77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
import json
|
|
import time
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))
|
|
from portaledger import (
|
|
LocalProblem,
|
|
PortaLedger,
|
|
PlanDelta,
|
|
ProvenanceEvent,
|
|
Attestation,
|
|
PortaBridge,
|
|
bridge_to_catopt,
|
|
GovernanceLedger,
|
|
)
|
|
|
|
|
|
def test_sign_and_log_provenance_roundtrip():
|
|
ledger = PortaLedger()
|
|
lp = LocalProblem(problem_id="p1", assets=["AAPL"], objectives=["maximize return"])
|
|
delta = PlanDelta(delta_id="d1", changes={"AAPL": {"hold": 10}}, timestamp=time.time())
|
|
signed = ledger.sign_delta(delta)
|
|
assert ledger.verify_delta(signed)
|
|
assert signed.attestation is not None
|
|
event = ProvenanceEvent(
|
|
event_id="ev1",
|
|
experiment_id="exp1",
|
|
actor="tester",
|
|
timestamp=time.time(),
|
|
contract_id="c1",
|
|
delta_hash=signed.delta_hash,
|
|
attestation=signed.attestation,
|
|
policy_flags=["default"],
|
|
)
|
|
ledger.log_provenance(event)
|
|
fetched = ledger.fetch_event("ev1")
|
|
assert fetched is not None
|
|
assert fetched.event_id == "ev1"
|
|
assert ledger.verify_event("ev1")
|
|
assert ledger.query_events(experiment_id="exp1")[0].event_id == "ev1"
|
|
|
|
|
|
def test_bridge_conversion():
|
|
ledger = PortaLedger()
|
|
lp = LocalProblem(problem_id="p1", assets=["BTC"], objectives=["minimize risk"])
|
|
delta = PlanDelta(delta_id="d1", changes={"BTC": {"alloc": 5}}, timestamp=time.time())
|
|
signed = ledger.sign_delta(delta)
|
|
assert signed.attestation is not None
|
|
event = ProvenanceEvent(
|
|
event_id="ev2",
|
|
experiment_id="exp2",
|
|
actor="tester",
|
|
timestamp=time.time(),
|
|
contract_id="c2",
|
|
delta_hash=signed.delta_hash,
|
|
attestation=signed.attestation,
|
|
policy_flags=["default"],
|
|
)
|
|
# simple bridge
|
|
composite = bridge_to_catopt(event, lp, {})
|
|
assert isinstance(composite, dict)
|
|
assert "LocalProblem" in composite
|
|
assert composite["graph_type"] == "GoC"
|
|
assert "PrivacyBudget" in composite
|
|
assert "PolicyBlock" in composite
|
|
|
|
|
|
def test_governance_versioning():
|
|
gov = GovernanceLedger()
|
|
version = gov.register_policy("tester", "policy-1", {"allow": True})
|
|
gov.record_decision("tester", "policy-1", "approve", {"reason": "ok"}, policy_version=version)
|
|
logs = gov.get_logs()
|
|
assert logs[0]["policy_version"] == version
|
|
latest = gov.latest_policy("policy-1")
|
|
assert latest is not None
|
|
assert latest["version"] == version
|