96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
"""Tests for BeVault core behavior."""
|
|
|
|
from idea172_bevault_verifiable_best import (
|
|
AuditLog,
|
|
DeltaSyncEngine,
|
|
GovernanceLedger,
|
|
GoCRegistry,
|
|
HedgeDelta,
|
|
LocalArbProblem,
|
|
PrivacyBudget,
|
|
PriceFeedAdapter,
|
|
BrokerAdapter,
|
|
SharedSignals,
|
|
canonical_payload,
|
|
sign_delta,
|
|
to_canonical_ir,
|
|
verify_delta,
|
|
)
|
|
|
|
|
|
def test_sign_and_verify_delta_deterministically():
|
|
delta = HedgeDelta(asset="AAPL", hedge_size=1.0, timestamp=1.0, policy_tag="test")
|
|
key = b"secret-key"
|
|
sig1 = sign_delta(delta, key)
|
|
sig2 = sign_delta(delta, key)
|
|
assert sig1 == sig2
|
|
assert verify_delta(delta, key, sig1)
|
|
|
|
|
|
def test_registry_reports_adapter_conformance():
|
|
reg = GoCRegistry()
|
|
price = PriceFeedAdapter()
|
|
broker = BrokerAdapter()
|
|
reg.register_adapter(price)
|
|
reg.register_adapter(broker)
|
|
reg.register_contract("LocalArbProblem", "1.0")
|
|
|
|
report = reg.conformance_report(price.name)
|
|
assert report["conforms"] is True
|
|
assert report["registered"] is True
|
|
assert reg.get_adapter(broker.name) is broker
|
|
assert reg.get_contract("LocalArbProblem") == "1.0"
|
|
|
|
|
|
def test_canonical_ir_includes_governance_metadata():
|
|
local = LocalArbProblem(asset="MSFT", objective="min_cost", risk_budget=10.0)
|
|
signals = SharedSignals(asset="MSFT", price=250.0, latency_ms=2.0, timestamp=1.0)
|
|
audit = AuditLog()
|
|
audit.record("approve", {"desk": "equities"})
|
|
budget = PrivacyBudget(asset="MSFT", remaining=5.0)
|
|
|
|
ir = to_canonical_ir(local, signals, audit_log=audit, privacy_budget=budget)
|
|
payload = canonical_payload(ir)
|
|
|
|
assert ir.version == "1.0"
|
|
assert "audit_log" in payload
|
|
assert "privacy_budget" in payload
|
|
|
|
|
|
def test_governance_ledger_hash_chain_verifies():
|
|
ledger = GovernanceLedger()
|
|
first = ledger.append("approve", {"ticket": "abc"}, timestamp=1.0)
|
|
second = ledger.append("execute", {"ticket": "abc"}, timestamp=2.0)
|
|
|
|
assert ledger.verify() is True
|
|
assert ledger.anchor() == second.entry_hash
|
|
assert first.previous_hash == ""
|
|
|
|
ledger.entries[0].payload["ticket"] = "mutated"
|
|
assert ledger.verify() is False
|
|
|
|
|
|
def test_delta_sync_replays_deterministically():
|
|
engine = DeltaSyncEngine(signing_key=b"sync-key")
|
|
delta_one = HedgeDelta(asset="AAPL", hedge_size=1.0, timestamp=1.0, policy_tag="a")
|
|
delta_two = HedgeDelta(asset="AAPL", hedge_size=2.0, timestamp=0.5, policy_tag="b")
|
|
|
|
engine.record(delta_one, actor="desk-a")
|
|
engine.record(delta_two, actor="desk-b")
|
|
|
|
replayed = engine.replay()
|
|
assert [delta.policy_tag for delta in replayed] == ["b", "a"]
|
|
assert replayed[0].signature is not None
|
|
|
|
|
|
def test_price_feed_builds_delta_and_broker_sends_it():
|
|
adapter = PriceFeedAdapter()
|
|
local = LocalArbProblem(asset="AAPL", objective="min_cost", risk_budget=10.0)
|
|
signals = adapter.generate_signal(asset="AAPL", price=101.5, latency_ms=1.2)
|
|
delta = adapter.build_delta(local, signals)
|
|
|
|
broker = BrokerAdapter()
|
|
assert broker.send_delta(delta) is True
|
|
assert broker.sent_deltas[-1] is delta
|
|
assert delta.audit_ref is not None
|