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

31 lines
1.0 KiB
Python

from energiamesh.bridge import EnergiBridge
from energiamesh.core import LocalProblem, SharedVariables, PlanDelta, DualVariables, AuditLog
def test_energi_bridge_local_problem_translation():
lb = LocalProblem(site_id="site-1", objective="minimize_cost")
bridge = EnergiBridge()
data = bridge.to_catopt(lb)
assert data["type"] == "LocalProblem"
assert data["site_id"] == "site-1"
assert data["version"] == bridge.version
def test_energi_bridge_shared_variables_translation():
sv = SharedVariables(signals={"forecast": {"temp": 20}})
bridge = EnergiBridge()
data = bridge.to_catopt(sv)
assert data["type"] == "SharedVariables"
assert "signals" in data
def test_energi_bridge_batch_translation():
bridge = EnergiBridge()
lb = LocalProblem(site_id="site-2", objective="dispatch")
dv = DualVariables(multipliers={"lambda": 1.0})
ad = AuditLog()
results = bridge.translate_batch([lb, dv, ad])
assert isinstance(results, list)
assert len(results) == 3
assert results[0]["type"] == "LocalProblem"