build(agent): weasel-1#856f80 iteration

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

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"]

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"

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"