50 lines
1.7 KiB
Python
50 lines
1.7 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
|
|
from portaledger.governance import 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)
|
|
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"
|
|
|
|
|
|
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)
|
|
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
|