72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
from state import SyncState
|
|
|
|
class TestSyncState:
|
|
def test_create_tables(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
assert True
|
|
state.close()
|
|
|
|
def test_upsert_and_get(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
state.upsert_event("uid-1", "hash-abc")
|
|
assert state.get_event_hash("uid-1") == "hash-abc"
|
|
state.close()
|
|
|
|
def test_get_uids(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
state.upsert_event("uid-1", "h1")
|
|
state.upsert_event("uid-2", "h2")
|
|
uids = state.get_event_uids()
|
|
assert "uid-1" in uids
|
|
assert "uid-2" in uids
|
|
assert len(uids) == 2
|
|
state.close()
|
|
|
|
def test_delete_event(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
state.upsert_event("uid-1", "h1")
|
|
state.delete_event("uid-1")
|
|
assert state.get_event_hash("uid-1") is None
|
|
state.close()
|
|
|
|
def test_clear_events(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
state.upsert_event("uid-1", "h1")
|
|
state.upsert_event("uid-2", "h2")
|
|
count = state.clear_events()
|
|
assert count == 2
|
|
assert len(state.get_event_uids()) == 0
|
|
state.close()
|
|
|
|
def test_ics_cache(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
state.set_ics_cache("hash-xyz", "etag-123")
|
|
h, e, _ = state.get_ics_cache()
|
|
assert h == "hash-xyz"
|
|
assert e == "etag-123"
|
|
state.clear_ics_cache()
|
|
h, e, _ = state.get_ics_cache()
|
|
assert h is None
|
|
assert e is None
|
|
state.close()
|
|
|
|
def test_snapshot_and_restore(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
state.upsert_event("uid-1", "h1")
|
|
state.upsert_event("uid-2", "h2")
|
|
snap = state.snapshot()
|
|
assert "uid-1" in snap["uids"]
|
|
state.clear_events()
|
|
assert len(state.get_event_uids()) == 0
|
|
state.restore_snapshot(snap)
|
|
assert len(state.get_event_uids()) == 2
|
|
assert state.get_event_hash("uid-1") == "h1"
|
|
state.close()
|
|
|
|
def test_empty_state(self, tmp_db):
|
|
state = SyncState(tmp_db)
|
|
assert len(state.get_event_uids()) == 0
|
|
h, e, _ = state.get_ics_cache()
|
|
assert h is None
|
|
state.close()
|