38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
import random
|
|
from interplanetary_edge_orchestrator_privacy.ir import (
|
|
plan_delta_to_json,
|
|
plan_delta_from_json,
|
|
merge_plan_deltas,
|
|
)
|
|
|
|
|
|
def make_pd(delta: dict, timestamp: float, author: str):
|
|
return {"delta": delta, "timestamp": timestamp, "author": author, "contract_id": "c1", "signature": f"sig-{author}"}
|
|
|
|
|
|
def test_plan_delta_json_roundtrip():
|
|
pd = make_pd({"x": 1, "y": 2}, timestamp=123.45, author="agent-A")
|
|
s = plan_delta_to_json(pd)
|
|
pd2 = plan_delta_from_json(s)
|
|
assert pd2["delta"]["x"] == 1
|
|
assert pd2["author"] == "agent-A"
|
|
|
|
|
|
def test_merge_plan_deltas_is_deterministic():
|
|
# Create three deltas with overlapping keys and different timestamps
|
|
p1 = make_pd({"a": 1}, timestamp=1.0, author="A")
|
|
p2 = make_pd({"b": 2}, timestamp=2.0, author="B")
|
|
p3 = make_pd({"a": 3}, timestamp=3.0, author="C")
|
|
|
|
baseline = merge_plan_deltas([p1, p2, p3])
|
|
# expected: a overwritten by p3, b from p2
|
|
assert baseline["delta"]["a"] == 3
|
|
assert baseline["delta"]["b"] == 2
|
|
|
|
# Shuffle inputs many times and assert merge result is identical
|
|
for _ in range(10):
|
|
arr = [p1, p2, p3][:]
|
|
random.shuffle(arr)
|
|
m = merge_plan_deltas(arr)
|
|
assert m == baseline
|