whee

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

test_fs.py (1967B)

      1 import logging
      2 import unittest
      3 import tempfile
      4 import uuid
      5 import shutil
      6 
      7 from whee.fs import FsStore
      8 
      9 logging.basicConfig(level=logging.DEBUG)
     10 logg = logging.getLogger()
     11 
     12 
     13 class TestFs(unittest.TestCase):
     14 
     15     def setUp(self):
     16         self.base = tempfile.mkdtemp()
     17         uu = uuid.uuid4()
     18         self.store = FsStore(base=self.base, dbname=str(uu), levels=2)
     19 
     20 
     21     def tearDown(self):
     22         shutil.rmtree(self.base)
     23 
     24 
     25     def test_fs_get_put(self):
     26         r = self.store.have(b'foo')
     27         self.assertFalse(r)
     28         self.store.put(b'foo', b'bar')
     29         r = self.store.have(b'foo')
     30         self.assertTrue(r)
     31         r = self.store.get(b'foo')
     32         self.assertEqual(r, b'bar')
     33         r = self.store.get(b'foo'.hex())
     34         self.assertEqual(r, b'bar')
     35         with self.assertRaises(FileExistsError):
     36             self.store.put(b'foo', b'baz')
     37         self.store.put(b'foo', b'baz', exist_ok=True)
     38         self.store.delete(b'foo')
     39         with self.assertRaises(FileNotFoundError):
     40             self.store.delete(b'foo')
     41 
     42 
     43     def test_fs_short(self):
     44         self.store.put(b'f', b'oo')
     45         self.store.put(b'fo', b'o')
     46         self.store.put(b'foo', b'')
     47 
     48 
     49     def test_fs_match(self):
     50         self.store.put(b'afoo', b'inky')
     51         self.store.put(b'abar', b'pinky')
     52         self.store.put(b'abaz', b'blinky')
     53         self.store.put(b'barbarbar', b'blinky')
     54         self.store.match(b'a')
     55         r = list(self.store)
     56         self.assertEqual(len(r), 3)
     57 
     58         self.store.match(b'ab')
     59         r = list(self.store)
     60         self.assertEqual(len(r), 2)
     61 
     62 
     63     def test_fs_all(self):
     64         self.store.put(b'afoo', b'inky')
     65         self.store.put(b'abar', b'pinky')
     66         self.store.put(b'abaz', b'blinky')
     67         self.store.put(b'barbarbar', b'blinky')
     68         r = list(self.store)
     69         self.assertEqual(len(r), 4)
     70 
     71         r = list(self.store.iter())
     72         self.assertEqual(len(r), 4)
     73 
     74 
     75 if __name__ == '__main__':
     76     unittest.main()