energiamesh-federated-contr.../tests/test_catopt_bridge.py

37 lines
1.2 KiB
Python

from energiamesh.catopt_bridge import CatOptBridge
from energiamesh.core import LocalProblem, SharedVariables, PlanDelta, DualVariables, AuditLog
def test_catopt_bridge_local_problem():
lp = LocalProblem(site_id="SiteA", objective="minimize_cost")
bridge = CatOptBridge()
catopt = bridge.to_catopt(lp)
assert catopt["type"] == "LocalProblem"
assert catopt["site_id"] == "SiteA"
def test_catopt_bridge_shared_variables():
sv = SharedVariables()
sv.update("forecast", {"temp": 20})
bridge = CatOptBridge()
catopt = bridge.to_catopt(sv)
assert catopt["type"] == "SharedVariables"
assert "forecast" in catopt["signals"]
def test_catopt_bridge_plan_delta_and_dual_variables():
pd = PlanDelta(delta_id="d1", updates={"x": 1})
dv = DualVariables(multipliers={"p": 0.5}, primal={"y": 2})
bridge = CatOptBridge()
assert bridge.to_catopt(pd)["type"] == "PlanDelta"
assert bridge.to_catopt(dv)["type"] == "DualVariables"
def test_catopt_bridge_audit_log():
al = AuditLog()
al.add_entry({"event": "start"})
bridge = CatOptBridge()
catopt = bridge.to_catopt(al)
assert catopt["type"] == "AuditLog"
assert isinstance(catopt["entries"], list)