63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
import time
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from energiamesh.core import LocalProblem, SharedVariables, PlanDelta, AuditLog, DualVariables
|
|
from energiamesh.transport import TLSChannel
|
|
from energiamesh.bridge import EnergiBridge
|
|
|
|
|
|
def test_local_problem_basic():
|
|
lp = LocalProblem(site_id="site-001", description="test", data={"load": 1.0})
|
|
assert lp.site_id == "site-001"
|
|
assert lp.data == {"load": 1.0}
|
|
assert isinstance(lp.parameters, dict)
|
|
|
|
|
|
def test_shared_variables_update_and_timestamp():
|
|
sv = SharedVariables(signals={"forecast_load": 10.0})
|
|
old_ts = sv.timestamp
|
|
time.sleep(0.01)
|
|
sv.update("forecast_load", 12.0)
|
|
assert sv.signals["forecast_load"] == 12.0
|
|
assert sv.timestamp >= old_ts
|
|
|
|
|
|
def test_plan_delta_and_metadata():
|
|
lp = LocalProblem(site_id="site-01")
|
|
sv = SharedVariables(signals={"forecast": 5.0})
|
|
delta = PlanDelta(delta={"update": True}, delta_id="d-1", source="unit-test")
|
|
assert delta.delta["update"] is True
|
|
assert delta.delta_id == "d-1"
|
|
assert delta.source == "unit-test"
|
|
|
|
|
|
def test_audit_log_basic():
|
|
al = AuditLog()
|
|
al.add("first entry")
|
|
al.log("event", {"k": 1})
|
|
assert len(al.entries) == 2
|
|
assert "ts" in al.entries[0]
|
|
|
|
|
|
def test_tls_channel_replay_protection():
|
|
chan = TLSChannel(peer_id="tester")
|
|
payload = {"hello": "world"}
|
|
envelope = chan.send(payload)
|
|
# First receive should succeed and return the payload
|
|
received = chan.receive(envelope)
|
|
assert received == payload
|
|
# Replay a second time should raise
|
|
with pytest.raises(ValueError):
|
|
chan.receive(envelope)
|
|
|
|
|
|
def test_energi_bridge_translation_basic():
|
|
bridge = EnergiBridge(version="0.1")
|
|
lp = LocalProblem(site_id="site-bridge", description="test bridge")
|
|
translated = bridge.to_catopt(lp)
|
|
assert translated["type"] == "LocalProblem"
|
|
assert translated["site_id"] == "site-bridge"
|
|
assert translated["version"] == "0.1"
|