54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
import time
|
|
import sys
|
|
import os
|
|
|
|
# Ensure the package in src/ is importable for tests
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
SRC = os.path.join(ROOT, "src")
|
|
if SRC not in sys.path:
|
|
sys.path.insert(0, SRC)
|
|
|
|
from idea138_guardrailops_federated_verifiable import PlanDelta, DeltaSyncEngine
|
|
|
|
|
|
def make_delta(delta_id: str, parent_id: str | None, changes) -> PlanDelta:
|
|
return PlanDelta(
|
|
delta_id=delta_id,
|
|
parent_id=parent_id,
|
|
timestamp=time.time(),
|
|
nonce="nonce-123",
|
|
changes=changes,
|
|
)
|
|
|
|
|
|
def test_delta_offline_replay_deterministically():
|
|
# initial state
|
|
state = {
|
|
"host_group": [],
|
|
"status": "idle",
|
|
}
|
|
|
|
engine = DeltaSyncEngine()
|
|
|
|
delta1 = make_delta(
|
|
"delta-1",
|
|
None,
|
|
[
|
|
{"key": "status", "value": "scanned", "op": "set"},
|
|
{"key": "host_group", "value": "hostA", "op": "append"},
|
|
],
|
|
)
|
|
|
|
# apply delta once
|
|
state1 = engine.apply_delta(state, delta1)
|
|
log1 = engine.log("delta_applied", {"delta_id": delta1.delta_id})
|
|
|
|
# simulate offline save and replay with the same delta and state seed
|
|
state2 = engine.apply_delta(state, delta1)
|
|
log2 = engine.log("delta_applied_replay", {"delta_id": delta1.delta_id})
|
|
|
|
assert state1 == state2
|
|
assert delta1.digest() == delta1.digest() # deterministic digest
|
|
# Ledger should have two entries for two applications
|
|
assert len(engine.ledger) >= 2
|