diff --git a/tests/test_adapters_solver.py b/tests/test_adapters_solver.py new file mode 100644 index 0000000..bf1fc30 --- /dev/null +++ b/tests/test_adapters_solver.py @@ -0,0 +1,36 @@ +from gravityweave.adapters.sat_planner import create as create_sat +from gravityweave.adapters.relay_module import create as create_relay +from gravityweave.plan_delta import PlanDelta +from gravityweave.solver import ADMMNode + + +def test_sat_planner_apply_burn(): + sat = create_sat("satA") + pd = PlanDelta() + pd.commitments.add({"type": "burn", "fuel": 10}) + ok = sat.apply_delta(pd) + assert ok is True + # fuel should be deducted + state = sat.describe_local_problem()["state"] + assert state["fuel"] == 90.0 + + +def test_relay_accept_downlink(): + r = create_relay("relay1") + pd = PlanDelta() + pd.commitments.add({"type": "downlink", "window": "T1"}) + ok = r.apply_delta(pd) + assert ok is True + assert len(r.schedule) == 1 + + +def test_admm_node_step(): + node = ADMMNode("n1") + initial = node.local_x + z = 8.0 + x_new = node.step(z) + # check that local_x updated and dual updated + assert node.local_x == x_new + assert isinstance(node.dual, float) + # value should be finite and reasonable + assert 0.0 <= x_new <= 20.0 diff --git a/tests/test_delta_synopsis.py b/tests/test_delta_synopsis.py new file mode 100644 index 0000000..cb20a86 --- /dev/null +++ b/tests/test_delta_synopsis.py @@ -0,0 +1,25 @@ +from gravityweave.delta_synopsis import build_priority_key, build_synopsis, serialize_synopsis, deserialize_synopsis +from gravityweave.plan_delta import PlanDelta + + +def test_build_priority_key_valid(): + key = build_priority_key(3, 10) + assert isinstance(key, int) + assert key == (3 << 4) | 10 + + +def test_build_synopsis_and_serialization(): + pd = PlanDelta() + pd.commitments.add({"id": "x", "type": "downlink"}) + pd.priority.write(5) + syn = build_synopsis(pd, parent_version="v0", urgency_score=2, impact_score=7, affected_assets=["satA"], signer="node1", key=b"secret") + + assert "delta_id" in syn + assert syn["parent_version"] == "v0" + assert isinstance(syn["priority_key"], int) + assert syn["affected_assets"] == ["satA"] + assert syn["short_sig"] != "" + + b = serialize_synopsis(syn) + syn2 = deserialize_synopsis(b) + assert syn2["delta_id"] == syn["delta_id"] diff --git a/tests/test_plan_delta.py b/tests/test_plan_delta.py new file mode 100644 index 0000000..1c61aa3 --- /dev/null +++ b/tests/test_plan_delta.py @@ -0,0 +1,57 @@ +from gravityweave.plan_delta import ORSet, LWWRegister, PlanDelta + + +def test_orset_add_remove_merge(): + a = ORSet() + b = ORSet() + + e1 = {"id": 1, "type": "burn", "fuel": 5} + e2 = {"id": 2, "type": "downlink", "window": "T1"} + + a.add(e1) + b.add(e2) + + # both merged should contain both elements + m = a.merge(b) + vals = m.value() + assert any(v["id"] == 1 for v in vals) + assert any(v["id"] == 2 for v in vals) + + # remove in one, merge should reflect removal by timestamp + a.remove(e1) + m2 = a.merge(b) + vals2 = m2.value() + assert not any(v["id"] == 1 for v in vals2) + assert any(v["id"] == 2 for v in vals2) + + +def test_lww_register_merge(): + r1 = LWWRegister() + r2 = LWWRegister() + + r1.write("low") + # ensure r2 is newer + r2.write("high") + + merged = r1.merge(r2) + assert merged.read() == "high" + + +def test_plandelta_merge_priority_origin(): + d1 = PlanDelta() + d2 = PlanDelta() + + d1.commitments.add({"id": "c1"}) + d2.commitments.add({"id": "c2"}) + + d1.priority.write(1) + d2.priority.write(2) + d1.origin = "nodeA" + d2.origin = "nodeB" + + merged = d1.merge(d2) + # both commitments present + ids = {c.get("id") for c in merged.commitments.value()} + assert "c1" in ids and "c2" in ids + # origin should follow the latest priority's origin (d2) + assert merged.origin == "nodeB" diff --git a/tests/test_registry_ledger.py b/tests/test_registry_ledger.py new file mode 100644 index 0000000..5836f61 --- /dev/null +++ b/tests/test_registry_ledger.py @@ -0,0 +1,25 @@ +from gravityweave.registry import Registry +from gravityweave.ledger import GovernanceLedger + + +def test_registry_contract_and_adapter(): + reg = Registry() + schema = {"type": "object", "properties": {"node_id": {"type": "string"}}} + cid = reg.register_contract("LocalProblem", schema, version="0.1") + assert cid.startswith("LocalProblem:") + retrieved = reg.get_contract(cid) + assert retrieved["schema"]["type"] == "object" + + aid = reg.register_adapter("sat_planner", cid, {"desc": "toy"}) + adapters = reg.list_adapters() + assert any(a["name"] == "sat_planner" for a in adapters) + + +def test_governance_ledger_append_and_verify(): + key = b"node-secret" + ledger = GovernanceLedger(key) + e1 = ledger.append("node1", {"action": "register", "contract": "c1"}) + e2 = ledger.append("node1", {"action": "update", "contract": "c1"}) + assert ledger.verify_chain() is True + entries = ledger.entries() + assert entries[0]["author"] == "node1"