Refactored sdp serializers
This commit is contained in:
+12
-8
@@ -1,27 +1,30 @@
|
||||
import pytest
|
||||
import os
|
||||
from os import path
|
||||
import shutil
|
||||
|
||||
from core.concept import Concept
|
||||
from core.sheerka import Sheerka
|
||||
|
||||
tests_root = path.abspath("../build/tests")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def init_test():
|
||||
print("Before yield")
|
||||
print("Current folder " + os.getcwd())
|
||||
if not os.path.exists("build/tests"):
|
||||
os.makedirs("build/tests")
|
||||
if path.exists(tests_root):
|
||||
shutil.rmtree(tests_root)
|
||||
|
||||
if not path.exists(tests_root):
|
||||
os.makedirs(tests_root)
|
||||
current_pwd = os.getcwd()
|
||||
os.chdir("build/tests")
|
||||
os.chdir(tests_root)
|
||||
|
||||
yield None
|
||||
|
||||
os.chdir(current_pwd)
|
||||
print("After yield")
|
||||
print("Current folder " + os.getcwd())
|
||||
|
||||
|
||||
def test_root_folder_is_created_after_initialization():
|
||||
print("Before yield")
|
||||
root_folder = "init_folder"
|
||||
|
||||
return_value = Sheerka().initialize(root_folder)
|
||||
@@ -49,3 +52,4 @@ def test_null_concept_are_equals():
|
||||
|
||||
assert Sheerka.concept_equals(None, None)
|
||||
assert Sheerka.concept_equals(concept1, concept1)
|
||||
|
||||
|
||||
@@ -0,0 +1,573 @@
|
||||
import pytest
|
||||
import os
|
||||
from os import path
|
||||
from sdp.sheerkaDataProvider import SheerkaDataProvider, Event
|
||||
from datetime import date, datetime
|
||||
import shutil
|
||||
import json
|
||||
|
||||
tests_root = path.abspath("../build/tests")
|
||||
|
||||
|
||||
def read_text_file(file_name):
|
||||
with open(file_name, "r") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
def read_json_file(file_name):
|
||||
with open(file_name, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
class ObjWithKey:
|
||||
def __init__(self, a, b):
|
||||
self.a = a
|
||||
self.b = b
|
||||
|
||||
def __eq__(self, obj):
|
||||
return isinstance(obj, ObjWithKey) and \
|
||||
self.a == obj.a and \
|
||||
self.b == obj.b
|
||||
|
||||
def __repr__(self):
|
||||
return f"ObjWithKey({self.a}, {self.b})"
|
||||
|
||||
def get_key(self):
|
||||
return self.a
|
||||
|
||||
|
||||
class ObjSetKey:
|
||||
def __init__(self, value, key=None):
|
||||
self.value = value
|
||||
self.key = key
|
||||
|
||||
def __eq__(self, obj):
|
||||
return isinstance(obj, ObjSetKey) and \
|
||||
self.key == obj.key and \
|
||||
self.value == obj.value
|
||||
|
||||
def __repr__(self):
|
||||
return f"ObjSetKey({self.key}, {self.value})"
|
||||
|
||||
def set_key(self, key):
|
||||
self.key = key
|
||||
|
||||
|
||||
class ObjNoKey:
|
||||
def __init__(self, a, b):
|
||||
self.a = a
|
||||
self.b = b
|
||||
|
||||
def __hash__(self):
|
||||
return hash((self.a, self.b))
|
||||
|
||||
def __eq__(self, obj):
|
||||
return isinstance(obj, ObjNoKey) and \
|
||||
self.a == obj.a and \
|
||||
self.b == obj.b
|
||||
|
||||
def __repr__(self):
|
||||
return f"ObjNoKey({self.a}, {self.b})"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def init_test():
|
||||
if path.exists(tests_root):
|
||||
shutil.rmtree(tests_root)
|
||||
|
||||
if not path.exists(tests_root):
|
||||
os.makedirs(tests_root)
|
||||
current_pwd = os.getcwd()
|
||||
os.chdir(tests_root)
|
||||
|
||||
yield None
|
||||
|
||||
os.chdir(current_pwd)
|
||||
|
||||
|
||||
def test_i_can_init_the_data_provider():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
|
||||
assert sdp.root == path.abspath(path.join(tests_root, ".sheerka"))
|
||||
assert path.exists(path.join(tests_root, ".sheerka"))
|
||||
|
||||
|
||||
def test_i_can_add_and_retrieve_an_event():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("hello world", date=date(year=2007, month=9, day=10), user="kodjo")
|
||||
|
||||
evt_digest = sdp.save_event(event)
|
||||
evt = sdp.load_event(evt_digest)
|
||||
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.EventFolder, evt_digest[0:24], evt_digest))
|
||||
assert evt.version == 1
|
||||
assert evt.date == datetime(year=2007, month=9, day=10)
|
||||
assert evt.user == "kodjo"
|
||||
assert evt.message == "hello world"
|
||||
|
||||
|
||||
def test_i_can_add_an_object():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("cmd add 'foo => bar'")
|
||||
event_digest = event.get_digest()
|
||||
obj = "foo => bar"
|
||||
|
||||
state_digest = sdp.add(event, "entry", obj)
|
||||
state = sdp.load_state(state_digest)
|
||||
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.EventFolder, event_digest[0:24], event_digest))
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.StateFolder, state_digest[0:24], state_digest))
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.HeadFile))
|
||||
|
||||
assert state.date is not None
|
||||
assert state.parents == []
|
||||
assert state.events == [event_digest]
|
||||
assert state.data == {"entry": "foo => bar"}
|
||||
|
||||
assert read_text_file(path.join(sdp.root, SheerkaDataProvider.HeadFile)) == state_digest
|
||||
|
||||
|
||||
def test_i_can_add_multiple_elements_in_an_entry():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("cmd add 'foo => bar'")
|
||||
|
||||
state_digest1 = sdp.add(event, "entry", 1)
|
||||
state1 = sdp.load_state(state_digest1)
|
||||
|
||||
state_digest2 = sdp.add(event, "entry", 2)
|
||||
state2 = sdp.load_state(state_digest2)
|
||||
|
||||
state_digest3 = sdp.add(event, "entry", 3)
|
||||
state3 = sdp.load_state(state_digest3)
|
||||
|
||||
assert state1.data == {"entry": 1}
|
||||
assert state2.data == {"entry": [1, 2]}
|
||||
assert state3.data == {"entry": [1, 2, 3]}
|
||||
|
||||
|
||||
def test_i_can_add_element_using_auto_generated_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("cmd add 'foo => bar'")
|
||||
key_file = path.join(sdp.root, SheerkaDataProvider.KeysFile)
|
||||
|
||||
sdp.add_with_auto_key(event, "entry1", "foo")
|
||||
sdp.add_with_auto_key(event, "entry1", "bar")
|
||||
sdp.add_with_auto_key(event, "entry2", "baz")
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
|
||||
assert path.exists(key_file)
|
||||
assert read_json_file(key_file) == {"entry1": 2, "entry2": 1}
|
||||
assert state.data == {"entry1": {"1": "foo", "2": "bar"}, "entry2": {"1": "baz"}}
|
||||
|
||||
|
||||
def test_i_can_add_and_auto_set_the_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("cmd add 'foo => bar'")
|
||||
key_file = path.join(sdp.root, SheerkaDataProvider.KeysFile)
|
||||
|
||||
sdp.add_with_auto_key(event, "entry1", ObjSetKey("foo"))
|
||||
sdp.add_with_auto_key(event, "entry1", ObjSetKey("bar"))
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
|
||||
assert path.exists(key_file)
|
||||
assert read_json_file(key_file) == {"entry1": 2}
|
||||
assert state.data == {"entry1": {"1": ObjSetKey("foo", "1"), "2": ObjSetKey("bar", "2")}}
|
||||
|
||||
|
||||
def test_i_can_add_an_object_with_its_own_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("cmd add 'foo => bar'")
|
||||
|
||||
sdp.add(event, "entry", ObjWithKey(1, "foo"))
|
||||
sdp.add(event, "entry", ObjWithKey(2, "bar"))
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
assert state.data == {"entry": {"1": ObjWithKey(1, "foo"), "2": ObjWithKey(2, "bar")}}
|
||||
|
||||
|
||||
def test_i_can_add_dictionary():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
event = Event("cmd add 'foo => bar'")
|
||||
|
||||
sdp.add(event, "entry", {"1": "foo"})
|
||||
sdp.add(event, "entry", {"2": "bar"})
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
assert state.data == {"entry": {"1": "foo", "2": "bar"}}
|
||||
|
||||
|
||||
def test_i_cannot_add_the_same_key_twice():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry", {"1": "foo"})
|
||||
|
||||
with pytest.raises(IndexError):
|
||||
sdp.add(Event("event"), "entry", {"1": "foo"})
|
||||
|
||||
|
||||
def test_i_cannot_add_the_same_element_twice():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry", ObjWithKey(1, "foo"))
|
||||
|
||||
with pytest.raises(IndexError):
|
||||
sdp.add(Event("event"), "entry", ObjWithKey(1, "foo"))
|
||||
|
||||
|
||||
def test_i_can_set_objects_with_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.set(Event("event"), "entry", ObjWithKey(1, "foo"))
|
||||
sdp.set(Event("event"), "entry", ObjWithKey(2, "foo"))
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
assert state.data == {"entry": {"2": ObjWithKey(2, "foo")}}
|
||||
|
||||
|
||||
def test_i_can_set_objects_with_no_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.set(Event("event"), "entry", ObjNoKey(1, "foo"))
|
||||
sdp.set(Event("event"), "entry", ObjNoKey(2, "foo"))
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
assert state.data == {"entry": ObjNoKey(2, "foo")}
|
||||
|
||||
|
||||
def test_i_can_set_from_list_to_dict():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.set(Event("event"), "entry", [ObjNoKey(1, "foo"), ObjNoKey(2, "foo")])
|
||||
sdp.set(Event("event"), "entry", {"1": ObjNoKey(1, "foo"), "2": ObjNoKey(2, "foo")})
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
assert state.data == {"entry": {"1": ObjNoKey(1, "foo"), "2": ObjNoKey(2, "foo")}}
|
||||
|
||||
|
||||
def test_i_can_add_unique():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add_unique(Event("event"), "entry", ObjNoKey(1, "foo"))
|
||||
sdp.add_unique(Event("event"), "entry", ObjNoKey(1, "foo"))
|
||||
sdp.add_unique(Event("event"), "entry", ObjNoKey(2, "bar"))
|
||||
sdp.add_unique(Event("event"), "entry", ObjNoKey(2, "bar"))
|
||||
|
||||
state = sdp.load_state(sdp.get_snapshot())
|
||||
assert state.data == {"entry": {ObjNoKey(1, "foo"), ObjNoKey(2, "bar")}}
|
||||
|
||||
|
||||
def test_i_can_keep_state_history():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
|
||||
event1 = Event("cmd add 'foo => bar'")
|
||||
event_digest1 = event1.get_digest()
|
||||
obj1 = "foo => bar"
|
||||
state_digest1 = sdp.add(event1, "entry1", obj1)
|
||||
|
||||
event2 = Event("cmd add 'foo => baz'")
|
||||
event_digest2 = event2.get_digest()
|
||||
obj2 = "foo => baz"
|
||||
state_digest2 = sdp.add(event2, "entry2", obj2)
|
||||
|
||||
state2 = sdp.load_state(state_digest2)
|
||||
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.EventFolder, event_digest1[0:24], event_digest1))
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.StateFolder, state_digest1[0:24], state_digest1))
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.EventFolder, event_digest2[0:24], event_digest2))
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.StateFolder, state_digest2[0:24], state_digest2))
|
||||
assert state2.date is not None
|
||||
assert state2.parents == [state_digest1]
|
||||
assert state2.events == [event_digest2]
|
||||
assert state2.data == {"entry1": "foo => bar", "entry2": "foo => baz"}
|
||||
|
||||
|
||||
def test_i_can_list_elements_when_there_is_nothing_to_list():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
|
||||
result = sdp.list("entry")
|
||||
|
||||
assert list(result) == []
|
||||
|
||||
|
||||
def test_i_can_list_when_no_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry1", "bar")
|
||||
sdp.add(Event("event"), "entry2", "baz")
|
||||
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == ["foo", "bar"]
|
||||
|
||||
|
||||
def test_i_can_list_when_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": "foo"})
|
||||
sdp.add(Event("event"), "entry1", {"2": "bar"})
|
||||
sdp.add(Event("event"), "entry2", {"3": "baz"})
|
||||
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == ["foo", "bar"]
|
||||
|
||||
|
||||
def test_i_can_list_when_one_element():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry2", "baz")
|
||||
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == ["foo"]
|
||||
|
||||
|
||||
def test_i_can_filter_on_key_for_dict():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": "foo"})
|
||||
sdp.add(Event("event"), "entry1", {"2": "bar"})
|
||||
|
||||
result = sdp.list("entry1", lambda k, o: k == "1")
|
||||
|
||||
assert list(result) == ["foo"]
|
||||
|
||||
|
||||
def test_i_can_filter_on_key_for_objects():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", ObjWithKey("a1", "b1"))
|
||||
sdp.add(Event("event"), "entry1", ObjWithKey("a2", "b2"))
|
||||
|
||||
result = sdp.list("entry1", lambda k, o: k == "a1")
|
||||
|
||||
assert list(result) == [ObjWithKey("a1", "b1")]
|
||||
|
||||
|
||||
def test_i_can_filter_on_attribute_for_dict():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": {"a": "a1", "b": "b1"}})
|
||||
sdp.add(Event("event"), "entry1", {"2": {"a": "a2", "b": "b2"}})
|
||||
|
||||
result = sdp.list("entry1", lambda k, o: o["a"] == "a2")
|
||||
|
||||
assert list(result) == [{"a": "a2", "b": "b2"}]
|
||||
|
||||
|
||||
def test_i_can_filter_on_attribute_for_object():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", ObjWithKey("a1", "b1"))
|
||||
sdp.add(Event("event"), "entry1", ObjWithKey("a2", "b2"))
|
||||
|
||||
result = sdp.list("entry1", lambda k, o: o.b == "b2")
|
||||
|
||||
assert list(result) == [ObjWithKey("a2", "b2")]
|
||||
|
||||
|
||||
def test_i_can_filter_a_list():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry1", "bar")
|
||||
|
||||
result = sdp.list("entry1", lambda o: o == "bar")
|
||||
|
||||
assert list(result) == ["bar"]
|
||||
|
||||
|
||||
def test_i_can_filter_a_list_of_object():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", ObjNoKey("a1", "b1"))
|
||||
sdp.add(Event("event"), "entry1", ObjNoKey("a2", "b2"))
|
||||
|
||||
result = sdp.list("entry1", lambda o: o.b == "b1")
|
||||
|
||||
assert list(result) == [ObjNoKey("a1", "b1")]
|
||||
|
||||
|
||||
def test_i_can_remove_all_elements():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry1", "bar")
|
||||
|
||||
state_digest = sdp.remove(Event("event"), "entry1")
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert read_text_file(path.join(sdp.root, SheerkaDataProvider.HeadFile)) == state_digest
|
||||
assert list(result) == []
|
||||
|
||||
|
||||
def test_i_can_remove_a_element():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry1", "bar")
|
||||
|
||||
sdp.remove(Event("event"), "entry1", lambda o: o == "foo")
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == ["bar"]
|
||||
|
||||
|
||||
def test_i_can_remove_dict_by_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": ObjNoKey("a1", "b1")})
|
||||
sdp.add(Event("event"), "entry1", {"2": ObjNoKey("a2", "b2")})
|
||||
|
||||
sdp.remove(Event("event"), "entry1", lambda k, o: k == "2")
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == [ObjNoKey("a1", "b1")]
|
||||
|
||||
|
||||
def test_i_can_remove_when_only_one_element():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
|
||||
sdp.remove(Event("event"), "entry1", lambda o: o == "foo")
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == []
|
||||
|
||||
|
||||
def test_i_cannot_remove_if_entry_does_not_exist():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
with pytest.raises(IndexError) as e:
|
||||
sdp.remove(Event("event"), "entry")
|
||||
assert str(e) == "entry"
|
||||
|
||||
|
||||
def test_i_can_replace_an_entry():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry1", "bar")
|
||||
|
||||
sdp.modify(Event("event"), "entry1", None, "baz")
|
||||
result = sdp.list("entry1")
|
||||
|
||||
assert list(result) == ["baz"]
|
||||
|
||||
|
||||
def test_i_cannot_update_an_entry_that_does_not_exist():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
|
||||
with pytest.raises(IndexError) as e:
|
||||
sdp.modify(Event("event"), "entry", "key", "foo")
|
||||
assert str(e) == "entry"
|
||||
|
||||
|
||||
def test_i_cannot_update_a_key_that_does_not_exist():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": "foo"})
|
||||
|
||||
with pytest.raises(IndexError) as e:
|
||||
sdp.modify(Event("event"), "entry1", "2", "bar")
|
||||
assert str(e) == "entry.1"
|
||||
|
||||
|
||||
def test_i_can_get_the_entire_entry():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
sdp.add(Event("event"), "entry1", "bar")
|
||||
|
||||
result = sdp.get("entry1")
|
||||
result_safe = sdp.get_safe("entry1")
|
||||
|
||||
assert result == ["foo", "bar"]
|
||||
assert result_safe == ["foo", "bar"]
|
||||
|
||||
|
||||
def test_i_can_get_an_entry_with_on_object():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", "foo")
|
||||
|
||||
result = sdp.get("entry1")
|
||||
result_safe = sdp.get_safe("entry1")
|
||||
|
||||
assert result == "foo"
|
||||
assert result_safe == "foo"
|
||||
|
||||
|
||||
def test_i_can_get_an_entry_by_key():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": "foo"})
|
||||
sdp.add(Event("event"), "entry1", {"2": "bar"})
|
||||
|
||||
result = sdp.get("entry1", "2")
|
||||
result_safe = sdp.get_safe("entry1", "2")
|
||||
|
||||
assert result == "bar"
|
||||
assert result_safe == "bar"
|
||||
|
||||
|
||||
def test_i_cannot_get_an_entry_that_does_not_exist():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
|
||||
assert sdp.get_safe("entry") is None
|
||||
with pytest.raises(IndexError) as e:
|
||||
sdp.get("entry")
|
||||
assert str(e) == "entry"
|
||||
|
||||
|
||||
def test_i_cannot_get_a_key_that_does_not_exist():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
sdp.add(Event("event"), "entry1", {"1": "foo"})
|
||||
|
||||
assert sdp.get_safe("entry1", "2") is None
|
||||
with pytest.raises(IndexError) as e:
|
||||
sdp.get("entry1", "2")
|
||||
assert str(e) == "entry.1"
|
||||
|
||||
|
||||
def test_i_can_save_and_retrieve_cache():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
txt = "foo bar baz foo bar baz foo bar baz"
|
||||
key = "key_to_use"
|
||||
category = "cache_category"
|
||||
|
||||
assert not sdp.in_cache(category, key)
|
||||
digest = sdp.add_to_cache(category, key, txt)
|
||||
assert path.exists(path.join(sdp.root, SheerkaDataProvider.CacheFolder, digest[0:24], digest))
|
||||
assert sdp.in_cache(category, key)
|
||||
|
||||
from_cache = sdp.load_from_cache(category, key)
|
||||
assert from_cache == txt
|
||||
|
||||
|
||||
def test_cache_is_not_updated_by_default():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
txt = "foo bar baz foo bar baz foo bar baz"
|
||||
txt2 = "foo foo foo foo foo foo foo foo foo"
|
||||
key = "key_to_use"
|
||||
category = "cache_category"
|
||||
|
||||
sdp.add_to_cache(category, key, txt)
|
||||
sdp.add_to_cache(category, key, txt2)
|
||||
|
||||
from_cache = sdp.load_from_cache(category, key)
|
||||
assert from_cache == txt
|
||||
|
||||
|
||||
def test_i_can_update_cache():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
txt = "foo bar baz foo bar baz foo bar baz"
|
||||
txt2 = "foo foo foo foo foo foo foo foo foo"
|
||||
key = "key_to_use"
|
||||
category = "cache_category"
|
||||
|
||||
sdp.add_to_cache(category, key, txt)
|
||||
sdp.add_to_cache(category, key, txt2, update=True)
|
||||
|
||||
from_cache = sdp.load_from_cache(category, key)
|
||||
assert from_cache == txt2
|
||||
|
||||
|
||||
def test_i_can_remove_from_cache():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
txt = "foo bar baz foo bar baz foo bar baz"
|
||||
key = "key_to_use"
|
||||
category = "cache_category"
|
||||
|
||||
sdp.add_to_cache(category, key, txt)
|
||||
digest = sdp.remove_from_cache(category, key)
|
||||
assert not path.exists(path.join(sdp.root, SheerkaDataProvider.CacheFolder, digest[0:24], digest))
|
||||
assert not sdp.in_cache(category, key)
|
||||
|
||||
|
||||
def test_i_can_test_than_an_entry_exits():
|
||||
sdp = SheerkaDataProvider(".sheerka")
|
||||
|
||||
assert not sdp.exists("entry")
|
||||
sdp.add(Event("event"), "entry", "value")
|
||||
assert sdp.exists("entry")
|
||||
@@ -0,0 +1,16 @@
|
||||
from sdp.sheerkaDataProvider import Event
|
||||
from sdp.sheerkaSerializer import Serializer
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
def test_i_can_serialize_an_event():
|
||||
event = Event("test", user="user", date=datetime.fromisoformat("2019-10-21T10:20:30.999"))
|
||||
serializer = Serializer()
|
||||
|
||||
stream = serializer.serialize(event)
|
||||
loaded = serializer.deserialize(stream)
|
||||
|
||||
assert event.version == loaded.version
|
||||
assert event.user == loaded.user
|
||||
assert event.date == loaded.date
|
||||
assert event.message == loaded.message
|
||||
@@ -0,0 +1,12 @@
|
||||
import core.utils
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lst, as_string", [
|
||||
(None, "",),
|
||||
([], ""),
|
||||
(["hello", "world"], "hello world"),
|
||||
(["hello world", "my friend"], '"hello world" "my friend"')
|
||||
])
|
||||
def test_i_can_create_string_from_a_list(lst, as_string):
|
||||
assert core.utils.sysarg_to_string(lst) == as_string
|
||||
Reference in New Issue
Block a user