48 lines
2.0 KiB
Python
48 lines
2.0 KiB
Python
import time
|
|
import sys, os
|
|
# Ensure repo root is on PYTHONPATH for tests when executed in isolation
|
|
repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if repo_root not in sys.path:
|
|
sys.path.insert(0, repo_root)
|
|
from signalvault.schema import SignalNode, HedgePlan
|
|
from signalvault.registry import GraphOfContractsRegistry
|
|
from signalvault.replay import DeterministicReplayEngine
|
|
from signalvault.adapters.price_feed_adapter import PriceFeedAdapter
|
|
from signalvault.adapters.simulated_venue_adapter import SimulatedVenueAdapter
|
|
|
|
|
|
def test_basic_delta_application_and_replay():
|
|
# Setup adapters
|
|
price_adapter = PriceFeedAdapter(asset="AAPL", venue="TestVenue")
|
|
venue_adapter = SimulatedVenueAdapter()
|
|
|
|
# Create initial nodes via adapter
|
|
node1 = price_adapter.generate_signal(price=120.0, timestamp=1)
|
|
node2 = price_adapter.generate_signal(price=80.0, timestamp=2)
|
|
|
|
# Build a simple delta: add two nodes
|
|
delta1 = {"add_nodes": [node1, node2]}
|
|
|
|
engine = DeterministicReplayEngine()
|
|
hash1 = engine.apply_delta(delta1)
|
|
|
|
# Add an edge and a hedge plan delta
|
|
edge = (node1, node2, "derived_from", "hash123")
|
|
from signalvault.schema import Edge, HedgePlan
|
|
e = Edge(from_node=node1, to_node=node2, relation=edge[2], delta_hash=edge[3])
|
|
plan = HedgePlan(id="plan-1", delta={"example": True}, version=1, approvals=["alice"])
|
|
delta2 = {"add_edges": [e], "add_hedges": [plan]}
|
|
hash2 = engine.apply_delta(delta2)
|
|
|
|
# Replay to the latest hash (no state change for this test beyond cumulative delta)
|
|
state = engine.replay_to_hash(hash2)
|
|
assert isinstance(state, dict)
|
|
assert "nodes" in state and len(state["nodes"]) >= 2
|
|
assert len(state["edges"]) == 1
|
|
assert len(state["hedges"]) == 1
|
|
|
|
# Idempotence: applying the same delta again should result in a new hash but identical counts
|
|
hash3 = engine.apply_delta(delta2)
|
|
assert hash3 != hash2 # hashing includes state; re-applying delta changes hash
|
|
assert len(state["nodes"]) >= 2
|