99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from solfuse import (
|
|
ADMMLiteSolver,
|
|
AdapterContract,
|
|
AuditLedger,
|
|
AuditLogEntry,
|
|
DeltaSyncJournal,
|
|
GraphOfContractsRegistry,
|
|
LocalProblem,
|
|
PrivacyBudget,
|
|
PrivacyBudgetLedger,
|
|
generate_identity,
|
|
sign_message,
|
|
verify_message,
|
|
)
|
|
|
|
|
|
def test_solver_converges_and_emits_deltas():
|
|
solver = ADMMLiteSolver(rho=1.0, tolerance=1e-8, max_iterations=50)
|
|
result = solver.solve(
|
|
[
|
|
LocalProblem(site_id="solar-a", preferred_dispatch=[4.0, 5.0], quadratic_weight=2.0),
|
|
LocalProblem(site_id="storage-b", preferred_dispatch=[2.0, 1.0], quadratic_weight=1.5),
|
|
]
|
|
)
|
|
|
|
assert result.iterations <= 50
|
|
assert len(result.consensus) == 2
|
|
assert len(result.deltas) == 2
|
|
assert result.primal_residual <= 1e-8
|
|
|
|
|
|
def test_registry_validates_contract_payloads():
|
|
registry = GraphOfContractsRegistry()
|
|
contract = AdapterContract(
|
|
name="solar-inverter",
|
|
version="1.0.0",
|
|
adapter_type="inverter",
|
|
input_schema={
|
|
"type": "object",
|
|
"properties": {"site_id": {"type": "string"}, "dispatch_kw": {"type": "number"}},
|
|
"required": ["site_id", "dispatch_kw"],
|
|
"additionalProperties": False,
|
|
},
|
|
output_schema={"type": "object"},
|
|
capabilities=["dispatch", "telemetry"],
|
|
)
|
|
registry.register(contract)
|
|
registry.validate_payload("solar-inverter", "1.0.0", {"site_id": "a", "dispatch_kw": 3.5})
|
|
with pytest.raises(Exception):
|
|
registry.validate_payload("solar-inverter", "1.0.0", {"site_id": "a"})
|
|
report = registry.conformance_report("solar-inverter")
|
|
assert report["contract"] == "solar-inverter"
|
|
assert report["checks"]
|
|
|
|
|
|
def test_identity_signing_round_trip():
|
|
identity = generate_identity("site-alpha")
|
|
payload = {"hello": "world", "n": 1}
|
|
signature = sign_message(identity, payload)
|
|
assert verify_message(identity.private_key.public_key(), payload, signature)
|
|
assert not verify_message(identity.private_key.public_key(), {"hello": "world", "n": 2}, signature)
|
|
|
|
|
|
def test_audit_and_privacy_budget_ledgers():
|
|
budgets = PrivacyBudgetLedger()
|
|
budgets.set_budget("actor-1", PrivacyBudget(epsilon=1.0, delta=1e-5))
|
|
updated = budgets.charge("actor-1", 0.25, 1e-6)
|
|
assert updated.spent_epsilon == pytest.approx(0.25)
|
|
|
|
audit = AuditLedger()
|
|
entry = AuditLogEntry(actor_id="actor-1", event_type="plan_delta", payload={"a": 1}, signature="sig")
|
|
audit.append(entry)
|
|
recent = audit.recent()
|
|
assert recent[0].actor_id == "actor-1"
|
|
assert audit.coverage() == 1.0
|
|
|
|
|
|
def test_delta_sync_journal_replays_sorted_deltas():
|
|
journal = DeltaSyncJournal()
|
|
solver = ADMMLiteSolver()
|
|
deltas = solver.solve(
|
|
[LocalProblem(site_id="x", preferred_dispatch=[1.0], quadratic_weight=1.0)]
|
|
).deltas
|
|
journal.merge(deltas)
|
|
loaded = journal.load()
|
|
assert loaded[0].site_id == "x"
|
|
|
|
|
|
def test_cli_payload_is_json_serializable():
|
|
solver = ADMMLiteSolver()
|
|
result = solver.solve([LocalProblem(site_id="x", preferred_dispatch=[1.0], quadratic_weight=1.0)])
|
|
json.dumps(result.model_dump(mode="json"))
|