29 lines
901 B
Python
29 lines
901 B
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from energiamesh.core import LocalProblem, SharedVariables, PlanDelta
|
|
from energiamesh.sync import compute_delta, DeltaStore
|
|
|
|
|
|
def test_delta_store_basic_store_and_replay():
|
|
# Prepare sample problem/state
|
|
lp = LocalProblem(site_id="site-1", problem_id="p1", description="test")
|
|
sv = SharedVariables(signals={"load_forecast": 123.0})
|
|
|
|
# Create two deltas to simulate incremental updates
|
|
d1 = compute_delta(lp, sv)
|
|
time.sleep(0.001)
|
|
sv.update("load_forecast", 120.0)
|
|
d2 = compute_delta(lp, sv)
|
|
|
|
store = DeltaStore()
|
|
store.add_delta(lp.site_id, d1)
|
|
store.add_delta(lp.site_id, d2)
|
|
|
|
# Replay should yield deltas in order
|
|
deltas = store.replay(lp.site_id)
|
|
assert len(deltas) == 2
|
|
assert isinstance(deltas[0], PlanDelta)
|
|
assert deltas[0].delta == d1.delta
|
|
assert deltas[1].delta == d2.delta
|