Start implementing local storage

Local storage allows app and plugins to store persistent data

estimage/local_storage.py

@@ -0,0 +1,44 @@
+
+
+
+class Storage:
+    def __init__(self):
+        self.storage = dict()
+        self.requests = set()
+        self.ns_requests = set()
+
+    def feed_key(self, key, value, namespace=tuple()):
+        if namespace:
+            ns = namespace[0]
+            if ns not in self.storage:
+                self.storage[ns] = dict()
+            self.storage[ns][key] = value
+        else:
+            self.storage[key] = value
+
+    def get_key(self, key):
+        return self.storage[key]
+
+    def get_namespace(self, namespace):
+        ns = namespace[0]
+        return self.storage[ns]
+
+    def request_key(self, key):
+        return self.requests.add(key)
+
+    def request_namespace(self, namespace):
+        ns = namespace[0]
+        return self.ns_requests.add(ns)
+
+    def save(self, io):
+        io.save_dict(self.storage)
+
+    def load(self, io):
+        data = io.load_to_dict(self.requests)
+        for nsreq in self.ns_requests:
+            try:
+                more = io.load_to_dict((nsreq,))
+            except KeyError:
+                more = {nsreq: dict()}
+            data.update(more)
+        self.storage.update(data)

tests/test_local_storage.py

@@ -0,0 +1,87 @@
+import pytest
+
+import estimage.local_storage as tm
+
+
+class MemoryIO:
+    def __init__(self):
+        self.storage = dict()
+
+    def save_dict(self, data):
+        self.storage.update(data)
+
+    def load_to_dict(self, keys):
+        return {key: self.storage[key] for key in keys}
+
+
+def test_stores():
+    storage = tm.Storage()
+    storage.feed_key("key", "value")
+    storage.feed_key("key", "val")
+    assert storage.get_key("key") == "val"
+
+
+def test_raises_exception_on_nonexistence():
+    storage = tm.Storage()
+    with pytest.raises(KeyError):
+        assert storage.get_key("yek")
+
+
+def test_is_isolated():
+    storage = tm.Storage()
+    storage.feed_key("key", "value")
+    storage = tm.Storage()
+    with pytest.raises(KeyError):
+        assert storage.get_key("key")
+
+
+def test_is_isolated():
+    storage = tm.Storage()
+    storage.feed_key("key", "value")
+    storage = tm.Storage()
+    with pytest.raises(KeyError):
+        assert storage.get_key("key")
+
+
+@pytest.fixture(params=("memory",))
+def storage_io(request):
+    choices = dict(
+        memory=MemoryIO,
+    )
+    io = choices[request.param]()
+    yield io
+
+
+def test_wrong_load(storage_io):
+    storage_out = tm.Storage()
+    storage_out.load(storage_io)
+    storage_out.request_key("ke-y")
+    with pytest.raises(KeyError, match="ke-y"):
+        storage_out.load(storage_io)
+
+
+def test_persistence(storage_io):
+    storage_in = tm.Storage()
+    storage_in.feed_key("key", "value")
+    storage_in.save(storage_io)
+    storage_out = tm.Storage()
+    storage_out.request_key("key")
+    storage_out.load(storage_io)
+    assert storage_out.get_key("key") == "value"
+
+
+def test_namespaces(storage_io):
+    storage_in = tm.Storage()
+    storage_in.feed_key("key", "value", namespace=("ns",))
+    storage_in.save(storage_io)
+
+    storage_out = tm.Storage()
+    storage_out.request_namespace(("ns",))
+    storage_out.load(storage_io)
+    assert storage_out.get_namespace(("ns",))["key"] == "value"
+    storage_out.request_namespace(("one", "two"))
+    storage_out.request_namespace(("one", "one"))
+    storage_out.load(storage_io)
+    assert not storage_out.get_namespace(("one", "one"))
+    assert storage_out.get_namespace(("one", "two"))["key"] == "value"
+