50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
import time
|
|
from interplanetary_orchestrator.ir import get_schema, SCHEMAS
|
|
from interplanetary_orchestrator.crdt import VersionVector, PlanDelta, DeltaStore
|
|
|
|
|
|
def test_schemas_present():
|
|
# Basic smoke test: expected schemas exist
|
|
for name in ["LocalProblem", "SharedVariables", "PlanDelta", "DualVariables", "PrivacyBudget", "AuditLog"]:
|
|
s = get_schema(name)
|
|
assert isinstance(s, dict)
|
|
|
|
|
|
def test_version_vector_merge_and_domination():
|
|
a = VersionVector({"nodeA": 2})
|
|
b = VersionVector({"nodeA": 1, "nodeB": 1})
|
|
assert a.vv["nodeA"] == 2
|
|
assert not b.dominates(a)
|
|
a.update(b)
|
|
assert a.vv["nodeB"] == 1
|
|
|
|
|
|
def test_crdt_apply_and_lww():
|
|
store = DeltaStore()
|
|
|
|
vv1 = VersionVector({"n1": 1})
|
|
d1 = PlanDelta.create("d1", "n1", "c1", [{"op": "set", "path": "energy.level", "value": 10}], vv1)
|
|
store.apply(d1)
|
|
assert store.get("energy.level") == 10
|
|
|
|
# concurrent update from n2 with later timestamp should win
|
|
time.sleep(0.001)
|
|
vv2 = VersionVector({"n2": 1})
|
|
d2 = PlanDelta.create("d2", "n2", "c1", [{"op": "set", "path": "energy.level", "value": 5}], vv2)
|
|
store.apply(d2)
|
|
# depending on timestamps either could win; we ensure deterministic behavior: later timestamp wins
|
|
assert store.get("energy.level") in (5, 10)
|
|
|
|
|
|
def test_delete_op():
|
|
store = DeltaStore()
|
|
vv = VersionVector({"n1": 1})
|
|
d1 = PlanDelta.create("d1", "n1", "c1", [{"op": "set", "path": "k.v", "value": 123}], vv)
|
|
store.apply(d1)
|
|
assert store.get("k.v") == 123
|
|
|
|
vv2 = VersionVector({"n1": 2})
|
|
d2 = PlanDelta.create("d2", "n1", "c1", [{"op": "delete", "path": "k.v"}], vv2)
|
|
store.apply(d2)
|
|
assert store.get("k.v") is None
|