whee

Unified interface to key value with locking and transactions.
Log | Files | Refs | README | LICENSE

test_valkey.py (1616B)

      1 import logging
      2 import unittest
      3 
      4 from whee.valkey import ValkeyStore
      5 
      6 logging.basicConfig(level=logging.DEBUG)
      7 logg = logging.getLogger()
      8 
      9 class TestValkey(unittest.TestCase):
     10 
     11     def setUp(self):
     12         self.store = ValkeyStore(124)
     13 
     14     def test_valkey_get_put(self):
     15         # remove the initial delete once we are creating temporary test dbs 
     16         try:
     17             self.store.delete(b'foo')
     18         except:
     19             pass
     20         r = self.store.have(b'foo')
     21         self.assertFalse(r)
     22         self.store.put(b'foo', b'bar')
     23         r = self.store.have(b'foo')
     24         self.assertTrue(r)
     25         r = self.store.get(b'foo')
     26         self.assertEqual(r, b'bar')
     27         r = self.store.get(b'foo'.hex())
     28         self.assertEqual(r, b'bar')
     29         with self.assertRaises(FileExistsError):
     30             self.store.put(b'foo', b'baz')
     31         self.store.put(b'foo', b'baz', exist_ok=True)
     32         self.store.delete(b'foo')
     33         with self.assertRaises(FileNotFoundError):
     34             self.store.delete(b'foo')
     35 
     36 
     37     def test_valkey_fs_match(self):
     38         self.store.put(b'afoo', b'inky')
     39         self.store.put(b'abar', b'pinky')
     40         self.store.put(b'abaz', b'blinky')
     41         self.store.put(b'barbarbar', b'blinky')
     42         self.store.match(b'a')
     43         r = list(self.store)
     44         self.assertEqual(len(r), 3)
     45 
     46         self.store.match(b'ab')
     47         r = list(self.store)
     48         self.assertEqual(len(r), 2)
     49 
     50         self.store.delete(b'afoo')
     51         self.store.delete(b'abar')
     52         self.store.delete(b'abaz')
     53         self.store.delete(b'barbarbar')
     54 
     55 
     56 if __name__ == '__main__':
     57     unittest.main()