59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from idea109_secureex_hardware_accelerated.adapters import FixAdapter, WebSocketAdapter
|
|
from idea109_secureex_hardware_accelerated.delta_sync import DeltaSync, reconcile_markets
|
|
from idea109_secureex_hardware_accelerated.enclave import SecureEnclave, verify_attestation
|
|
from idea109_secureex_hardware_accelerated.governance import GovernanceLedger
|
|
|
|
|
|
class SecureExTests(unittest.TestCase):
|
|
def test_adapters_parse_canonical_markets(self) -> None:
|
|
fix_market = FixAdapter(venue="VENUE_A").parse_market("35=W|55=ABC|132=10.1|133=10.2|134=100|135=120|34=7|52=1.5")
|
|
ws_market = WebSocketAdapter(venue="VENUE_B").parse_market(
|
|
{"asset": "ABC", "bid": 10.0, "ask": 10.3, "bid_size": 80, "ask_size": 60, "sequence": 8, "timestamp": 1.7}
|
|
)
|
|
self.assertEqual(fix_market.asset, "ABC")
|
|
self.assertEqual(ws_market.venue, "VENUE_B")
|
|
self.assertAlmostEqual(fix_market.mid, 10.15)
|
|
|
|
def test_enclave_attestation_and_deterministic_decision(self) -> None:
|
|
markets = [
|
|
FixAdapter(venue="VENUE_A").parse_market("55=ABC|132=10.1|133=10.2|134=100|135=120|34=7|52=1.5"),
|
|
WebSocketAdapter(venue="VENUE_B").parse_market("{\"asset\": \"ABC\", \"bid\": 10.0, \"ask\": 10.3, \"bid_size\": 80, \"ask_size\": 60, \"sequence\": 8, \"timestamp\": 1.7}")
|
|
]
|
|
enclave = SecureEnclave(enclave_id="edge-node-1", secret="demo-secret")
|
|
decision1 = enclave.compute(markets)
|
|
decision2 = enclave.compute(markets)
|
|
self.assertEqual(decision1.plan, decision2.plan)
|
|
report = enclave.attest("nonce-1")
|
|
self.assertTrue(verify_attestation(report, enclave.measurement, "demo-secret", enclave.enclave_id))
|
|
|
|
def test_ledger_chain_and_replay(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
ledger = GovernanceLedger(Path(tmp) / "ledger.sqlite3")
|
|
enclave = SecureEnclave(enclave_id="edge-node-1", secret="demo-secret")
|
|
report = enclave.attest("nonce-2")
|
|
ledger.append("plan_delta", {"plan_id": "x"}, 1.0, report)
|
|
self.assertTrue(ledger.verify_chain())
|
|
|
|
sync = DeltaSync()
|
|
markets = [
|
|
FixAdapter(venue="VENUE_A").parse_market("55=ABC|132=10.1|133=10.2|134=100|135=120|34=7|52=1.5"),
|
|
WebSocketAdapter(venue="VENUE_B").parse_market({"asset": "ABC", "bid": 10.0, "ask": 10.3, "sequence": 8}),
|
|
]
|
|
first = sync.snapshot(markets)
|
|
second = sync.snapshot(markets)
|
|
self.assertEqual(len(first), 2)
|
|
self.assertEqual(second, [])
|
|
delta = reconcile_markets(markets, markets[:1])
|
|
self.assertEqual(len(delta), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|