26 lines
920 B
Python
26 lines
920 B
Python
from deltatrace.capture import create_reproducible_envelope, envelope_to_json
|
|
from deltatrace.event_graph import Event, EventType, EventGraph
|
|
from deltatrace.replay_engine import ReplayEngine
|
|
|
|
|
|
def test_reproducible_envelope_contains_snapshot_and_replays():
|
|
g = EventGraph()
|
|
t1 = Event.create(EventType.MD_TICK, {"v": 1})
|
|
s1 = Event.create(EventType.SIGNAL, {"v": 2})
|
|
g.add_event(t1)
|
|
g.add_event(s1)
|
|
g.add_edge(t1.id, s1.id)
|
|
|
|
# create checkpoint/snapshot
|
|
snapshot = g.to_dict()
|
|
|
|
env = create_reproducible_envelope(snapshot, adapters={"fix": "fix-adapter@0.1"}, rng_seed=42, container_image="sha256:deadbeef", cmdline="demo --run")
|
|
|
|
j = envelope_to_json(env)
|
|
assert isinstance(j, str) and 'snapshot' in j
|
|
|
|
# replay from snapshot inside envelope
|
|
engine = ReplayEngine()
|
|
res = engine.replay_from_snapshot(env.snapshot)
|
|
assert res.events_replayed == len(g.events)
|