42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
import pytest
|
|
|
|
from gridresilience_studio.energi_bridge import EnergiBridge, LocalProblem, SharedSignals
|
|
from gridresilience_studio.core import Object, Morphism, PlanDelta
|
|
|
|
|
|
def test_goC_object_mapping():
|
|
lp = LocalProblem(id="LP1", description="Test LocalProblem", resources={"foo": "bar"})
|
|
obj = EnergiBridge.map_object(lp)
|
|
assert isinstance(obj, Object)
|
|
goc = EnergiBridge.to_goc_object(obj)
|
|
assert goc["type"] == "Object"
|
|
assert goc["id"] == "LP1"
|
|
assert goc["class"] == "LocalProblem" or goc["class"] == obj.type
|
|
assert isinstance(goc["properties"], dict)
|
|
|
|
|
|
def test_goC_morphism_mapping():
|
|
sig = SharedSignals(id="SIG1", source="LP1", target="LP2", signals={"voltage": 1.0})
|
|
morph = EnergiBridge.map_signals(sig)
|
|
assert isinstance(morph, Morphism)
|
|
gocm = EnergiBridge.to_goc_morphism(morph)
|
|
assert gocm["type"] == "Morphism"
|
|
assert gocm["id"] == "SIG1"
|
|
assert gocm["source"] == "LP1"
|
|
assert gocm["signals"] == {"voltage": 1.0}
|
|
|
|
|
|
def test_delta_reconciliation():
|
|
d_local = PlanDelta(delta_id="D1", islanded=False, actions=[{"type": "noop"}], tags={})
|
|
d_remote = PlanDelta(delta_id="D1", islanded=True, actions=[{"type": "toggle"}], tags={"t": "v1"})
|
|
merged = EnergiBridge.reconcile_deltas([d_local], [d_remote])
|
|
assert len(merged) == 1
|
|
merged_delta = merged[0]
|
|
assert merged_delta.delta_id == "D1"
|
|
# both actions should be merged without duplicates
|
|
actions = merged_delta.actions
|
|
assert any(a.get("type") == "noop" for a in actions)
|
|
assert any(a.get("type") == "toggle" for a in actions)
|
|
# islanded should be True due to OR
|
|
assert merged_delta.islanded is True
|