40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
import time
|
|
from interplanetary_edge_orchestrator_privacy.core import LocalProblem, SharedVariables, PlanDelta, DualVariables, PrivacyBudget, AuditLog
|
|
|
|
def test_local_problem_basic():
|
|
lp = LocalProblem(
|
|
id="lp1",
|
|
domain="rover",
|
|
assets=["rover1"],
|
|
objective="minimize_energy"
|
|
)
|
|
assert lp.id == "lp1"
|
|
assert lp.domain == "rover"
|
|
assert "rover1" in lp.assets
|
|
|
|
def test_shared_variables_versioning():
|
|
sv = SharedVariables()
|
|
v1 = sv.bump_version("forecasts")
|
|
v2 = sv.bump_version("forecasts")
|
|
assert v1 == 1 and v2 == 2
|
|
sv.payloads["forecasts"] = {"temp": 20}
|
|
d = sv.to_dict()
|
|
assert "forecasts" in d["payloads"]
|
|
|
|
def test_plan_delta_signature():
|
|
pd = PlanDelta(delta={"delta": "update"})
|
|
pd.sign("agent-A")
|
|
assert pd.author == "agent-A"
|
|
assert pd.signature is not None
|
|
|
|
def test_dual_and_privacy_and_audit():
|
|
dv = DualVariables()
|
|
dv.set("lambda", 1.23)
|
|
assert dv.multipliers["lambda"] == 1.23
|
|
|
|
pb = PrivacyBudget(signal="mesh-energy", budget=0.5, expiry=time.time() + 60)
|
|
assert not pb.is_expired()
|
|
|
|
al = AuditLog(entry="init", signer="agent-A", timestamp=time.time())
|
|
assert al.entry == "init"
|