shep

Multi-state key stores using bit masks for python3
git clone git://git.defalsify.org/shep.git
Log | Files | Refs | LICENSE

commit 3de0ad8ee357c2e77f6e7c400bdf04d4fe85d808
parent d232632c36d80514b792dce4a2e3633f16482f6f
Author: lash <dev@holbrook.no>
Date:   Mon, 31 Jan 2022 15:53:33 +0000

Add simple file store backend

Diffstat:
MCHANGELOG | 1+
Ashep/store/file.py | 33+++++++++++++++++++++++++++++++++
Atests/test_file.py | 42++++++++++++++++++++++++++++++++++++++++++
3 files changed, 76 insertions(+), 0 deletions(-)

diff --git a/CHANGELOG b/CHANGELOG @@ -1,5 +1,6 @@ - 0.0.6 * Update description + * Implement simple file store backend (add only) - 0.0.5 * Add persistence wrapper - 0.0.4 diff --git a/shep/store/file.py b/shep/store/file.py @@ -0,0 +1,33 @@ +# standard imports +import os + + +class SimpleFileStore: + + def __init__(self, path): + self.path = path + os.makedirs(self.path, exist_ok=True) + + + def add(self, v, contents=''): + fp = os.path.join(self.path, v) + try: + os.stat(fp) + raise FileExistsError(fp) + except FileNotFoundError: + pass + f = open(fp, 'w') + f.write(contents) + f.close() + + +class SimpleFileStoreFactory: + + def __init__(self, path): + self.path = path + + + def add(self, k): + k = str(k) + store_path = os.path.join(self.path, k) + return SimpleFileStore(store_path) diff --git a/tests/test_file.py b/tests/test_file.py @@ -0,0 +1,42 @@ +# standard imports +import unittest +import tempfile +import os + +# local imports +from shep.persist import PersistedState +from shep.store.file import SimpleFileStoreFactory +from shep.error import ( + StateExists, + StateInvalid, + ) + + +class TestStateReport(unittest.TestCase): + + def setUp(self): + self.d = tempfile.mkdtemp() + self.factory = SimpleFileStoreFactory(self.d) + self.states = PersistedState(self.factory.add, 4) + self.states.add('foo') + self.states.add('bar') + self.states.add('baz') + + + def test_add(self): + self.states.put('abcd', self.states.FOO) + fp = os.path.join(self.d, 'FOO', 'abcd') + f = open(fp, 'r') + v = f.read() + f.close() + self.assertEqual(len(v), 0) + + + def test_dup(self): + self.states.put('abcd', self.states.FOO) + with self.assertRaises(FileExistsError): + self.states.put('abcd', self.states.FOO) + + +if __name__ == '__main__': + unittest.main()