44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
import tempfile
|
|
import sys
|
|
import os
|
|
|
|
# Ensure repository root is on sys.path for tests
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from idea70_narrativeweave_real_time.storage import EventLog
|
|
from idea70_narrativeweave_real_time.adapters import news_adapter_event, transcript_adapter_event, build_block_from_events
|
|
from idea70_narrativeweave_real_time.ledger import SimpleLedger
|
|
from idea70_narrativeweave_real_time.narrative import render_narrative_block
|
|
|
|
|
|
def test_deterministic_replay_and_ledger():
|
|
dbf = tempfile.NamedTemporaryFile()
|
|
el = EventLog(path=dbf.name)
|
|
|
|
# append deterministic events
|
|
el.append_event("news", news_adapter_event("Market rallies after earnings", "https://news/1", confidence=0.8))
|
|
el.append_event("transcript", transcript_adapter_event("CEO", "We delivered strong growth this quarter.", "2023-01-01T00:00:00Z"))
|
|
|
|
events = el.read_events()
|
|
payloads = events
|
|
|
|
# build block twice and ensure identical id and deterministic JSON
|
|
b1 = build_block_from_events(payloads)
|
|
b2 = build_block_from_events(payloads)
|
|
assert b1.id == b2.id
|
|
assert b1.timestamp == b2.timestamp
|
|
assert b1.topic == "news | transcript"
|
|
|
|
ledger = SimpleLedger()
|
|
import json
|
|
|
|
a1 = ledger.anchor(json.dumps(b1.model_dump(), sort_keys=True, default=str))
|
|
a2 = ledger.anchor(json.dumps(b2.model_dump(), sort_keys=True, default=str))
|
|
# Anchors should be chained and deterministic
|
|
assert a1 != ""
|
|
assert a2 != a1
|
|
|
|
rendered = render_narrative_block(b1)
|
|
assert "# news | transcript" in rendered
|
|
assert "word_count" in rendered
|