40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from energi_bridge.adapters import DERAggregatorAdapter, SolarMeterAdapter
|
|
from energi_bridge.orchestrator import VerifiableTelemOrchestrator
|
|
|
|
|
|
def test_orchestrator_emits_registry_and_signed_plan() -> None:
|
|
adapters = [
|
|
SolarMeterAdapter(site_id="site-a"),
|
|
DERAggregatorAdapter(aggregator_id="agg-a"),
|
|
]
|
|
orchestrator = VerifiableTelemOrchestrator(
|
|
adapters=adapters,
|
|
signing_key="unit-test",
|
|
clock=lambda: datetime(2026, 4, 25, 12, 0, 0),
|
|
)
|
|
|
|
result = orchestrator.run_once()
|
|
|
|
signals = result["shared_variables"]["signals"]
|
|
assert signals["solar_meter.site-a"]["solar_output_kw"] == 750.0
|
|
assert signals["der_aggregator.agg-a"]["total_power_kw"] == 500.0
|
|
|
|
payload = {
|
|
"signals": signals,
|
|
"ts": "2026-04-25T12:00:00",
|
|
}
|
|
expected_signature = hmac.new(
|
|
b"unit-test",
|
|
json.dumps(payload, sort_keys=True).encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
assert result["plan_delta"]["signature"] == expected_signature
|
|
|
|
contract_ids = [item["contract_id"] for item in result["contract_registry"]]
|
|
assert contract_ids == ["der_aggregator.agg-a", "solar_meter.site-a"]
|