61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
import os
|
|
import secrets
|
|
from gravityweave.registry import Registry
|
|
from gravityweave.plan_delta import PlanDelta, ORSet, LWWRegister
|
|
from gravityweave.ledger import GovernanceLedger
|
|
from gravityweave.adapters.sat_planner import create as sat_create
|
|
from gravityweave.adapters.relay_module import create as relay_create
|
|
from gravityweave.solver import ADMMNode
|
|
|
|
|
|
def test_registry_and_adapters():
|
|
reg = Registry()
|
|
cid = reg.register_contract("sat_problem", {"fuel": "float"})
|
|
aid = reg.register_adapter("sat_planner", cid, {"version": "0.1"})
|
|
adapters = reg.list_adapters()
|
|
assert any(a["name"] == "sat_planner" for a in adapters)
|
|
|
|
|
|
def test_plan_delta_merge():
|
|
p1 = PlanDelta(origin="nodeA")
|
|
p1.commitments.add({"type": "burn", "fuel": 1.0})
|
|
p1.priority.write(5)
|
|
|
|
p2 = PlanDelta(origin="nodeB")
|
|
p2.commitments.add({"type": "downlink", "window": 42})
|
|
p2.priority.write(10)
|
|
|
|
merged = p1.merge(p2)
|
|
vals = merged.commitments.value()
|
|
assert any(isinstance(v, dict) for v in vals)
|
|
assert merged.priority.read() == 10
|
|
|
|
|
|
def test_ledger_append_and_verify():
|
|
key = secrets.token_bytes(32)
|
|
ledger = GovernanceLedger(key)
|
|
e1 = ledger.append("nodeA", {"action": "seed"})
|
|
e2 = ledger.append("nodeB", {"action": "update"})
|
|
assert ledger.verify_chain()
|
|
entries = ledger.entries()
|
|
assert entries[0]["author"] == "nodeA"
|
|
|
|
|
|
def test_adapters_apply_delta():
|
|
sat = sat_create("sat1")
|
|
relay = relay_create("relay1")
|
|
delta = PlanDelta(origin="test")
|
|
delta.commitments.add({"type": "burn", "fuel": 1.0})
|
|
delta.commitments.add({"type": "downlink", "window": 99})
|
|
|
|
assert sat.apply_delta(delta)
|
|
assert relay.apply_delta(delta)
|
|
|
|
|
|
def test_admm_node_step():
|
|
node = ADMMNode("n1")
|
|
z = 8.0
|
|
x_new = node.step(z)
|
|
# numeric sanity
|
|
assert isinstance(x_new, float)
|