573 lines
17 KiB
Python
573 lines
17 KiB
Python
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") |