▶ runtime-narrative

Testing with StoryRecorder

StoryRecorder is a dual sync/async context manager for use in tests. It creates the story context internally, records every emitted event, and provides assertion helpers that produce clear failure messages.

The key design rule

StoryRecorder IS the story. Functions under test must call stage() directly without their own story() wrapper. If the function under test wraps itself in story(), the story contexts will nest and stage events from the inner story will not be visible to the recorder.

python
# Correct — the function calls stage() directly, no story() inside
def import_users(path: str) -> None:
    from runtime_narrative import stage
    with stage("Read File"):
        data = open(path).read()
    with stage("Parse CSV"):
        rows = [line.split(",") for line in data.splitlines()]
    with stage("Upsert"):
        pass

# In the test:
from runtime_narrative.testing import StoryRecorder

def test_import_users(tmp_path):
    csv = tmp_path / "users.csv"
    csv.write_text("alice,alice@example.com\n")
    with StoryRecorder("User Import") as r:
        import_users(str(csv))
    r.assert_stages_completed(["Read File", "Parse CSV", "Upsert"])
    r.assert_no_failure()

All four assertion methods

assert_stages_completed

Verify that each named stage emitted StageCompleted, in order.

python
r.assert_stages_completed(["Load", "Validate", "Insert"])

AssertionError: "Expected stages not completed: ['Insert']. Completed stages were: ['Load', 'Validate']"

assert_no_failure

Verify that no FailureOccurred was emitted.

python
r.assert_no_failure()

AssertionError: "Expected no failure but got: ValueError: duplicate key at ..."

assert_stage_failed

Verify that a specific stage raised. error_type is a string — the class name — not the class itself.

python
r.assert_stage_failed("Insert")
r.assert_stage_failed("Insert", error_type="ValueError")

AssertionError: "Expected error type 'ValueError' at stage 'Insert' but got 'KeyError'"

assert_story_completed

Verify that StoryCompleted was emitted, optionally asserting its success value.

python
r.assert_story_completed()
r.assert_story_completed(success=True)
r.assert_story_completed(success=False)

error_type is a string, not the class itself

In assert_stage_failed, error_type is a string (e.g. "ValueError"), not the exception class. FailureOccurred.error_type is always a string, so error_type=ValueError will never match — pass error_type="ValueError".

Sync and async patterns

python
import asyncio
from runtime_narrative import stage
from runtime_narrative.testing import StoryRecorder

# Sync
def test_sync_success():
    with StoryRecorder("Pipeline", total_stages=2) as r:
        with stage("Step A"):
            pass
        with stage("Step B"):
            pass
    r.assert_stages_completed(["Step A", "Step B"])
    r.assert_no_failure()

# Async
async def _async_pipeline(fail: bool):
    async with stage("Fetch"):
        await asyncio.sleep(0)
    async with stage("Save"):
        if fail:
            raise IOError("disk full")

def test_async_failure():
    async def run():
        async with StoryRecorder("Async Pipeline") as r:
            await _async_pipeline(fail=True)
        return r

    import pytest
    with pytest.raises(IOError):
        r = asyncio.run(run())
    r.assert_stage_failed("Save", error_type="IOError")
    r.assert_story_completed(success=False)

Inspecting raw events

All emitted events are available on recorder.events.

python
with StoryRecorder("Pipeline") as r:
    with stage("Load"):
        pass

stage_events = [e for e in r.events if type(e).__name__ == "StageCompleted"]
assert stage_events[0].duration_seconds >= 0

story_started = next(e for e in r.events if type(e).__name__ == "StoryStarted")
assert story_started.story_name == "Pipeline"

Using dry_run with StoryRecorder

Combine dry_run with StoryRecorder to assert that stages are wired up correctly without needing real dependencies to succeed.

python
def test_stages_are_wired_up():
    with StoryRecorder("Pipeline", dry_run=True) as r:
        with stage("Fetch from DB"):
            raise RuntimeError("skipped — no DB in test")
        with stage("Write to S3"):
            raise RuntimeError("skipped — no S3 in test")
    r.assert_stages_completed(["Fetch from DB", "Write to S3"])
    r.assert_no_failure()

Next step

Read dry_run Mode for the full behavior of skeleton-mode stories — what runs, what's suppressed, and when to reach for it.