TelTest Phase 3: Fixtures, YAML DSL & pytest Plugin — Implementation Plan¶

Design Document: docs/designs/teltest/README.md Priority: P1 — Critical for regression safety Estimated Duration: 6 weeks across 9 sections (3.1–3.9) Prerequisites: Phase 1 (MUDClient) and Phase 2 (Expectation Engine) must be complete

Summary¶

This phase delivers the integration layer that makes TelTest usable as a pytest-native E2E testing framework. Where Phase 1 provides the raw MUDClient telnet driver and Phase 2 provides the expectation engine (cursor-based expect(), expect_prompt(), expect_sequence(), etc.), Phase 3 builds everything on top:

pytest plugin — Marker registration, CLI options, YAML test collection, transcript-on-failure hook
AuthDriver protocol — Pluggable login flow abstraction keeping MUDClient portable
MaidAuthDriver — Concrete MAID login driver (register, login, character select/create)
Test fixtures — mud_server (session-scoped engine lifecycle), mud_client (function-scoped logged-in client), raw_client (bare connection), registered_account, _isolate_world (snapshot/restore)
YAML DSL — Declarative test scripts with Pydantic schema validation, supporting send/expect/branch/group/note steps
Script runner — Executes YAML scripts against a live server via MUDClient + AuthDriver
Session recorder — Captures interactive sessions to .teltest JSON files
Recording converter — Transforms recordings into YAML test scripts with fuzzy pattern detection
CLI — teltest validate, teltest run, teltest record, teltest convert commands

Cross-document dependencies:

Dependency	Document	What this phase needs
Phase 1	`docs/impl/teltest/01-mudclient.md`	`MUDClient`, `TelnetProtocol`, `OutputBuffer`, `TranscriptEntry`, `Match`
Phase 2	`docs/impl/teltest/02-expect-engine.md`	`expect()`, `expect_prompt()`, `expect_sequence()`, `expect_any()`, `expect_not()`, `ExpectTimeout`, `ConnectionClosed`, `UnexpectedMatch`
Phase 4	`docs/impl/teltest/04-ci-advanced.md`	CI workflow, pytest-xdist support, GMCP assertions (consumed by Phase 4, not required here)

3.1 pytest Plugin¶

Package: teltest | Priority: P0 | Dependencies: Phase 1 (MUDClient), Phase 2 (expect engine)

File: packages/teltest/src/teltest/pytest_plugin.py

Entry Point Registration¶

[ ] Register the pytest plugin entry point in packages/teltest/pyproject.toml under [project.entry-points.pytest11]:
```
[project.entry-points.pytest11]
teltest = "teltest.pytest_plugin"
```

TelTestConfig Dataclass¶

[ ] Define TelTestConfig dataclass in packages/teltest/src/teltest/pytest_plugin.py:

@dataclass
class TelTestConfig:
    timeout: float = 5.0
    timeout_multiplier: float = 1.0
    strip_ansi: bool = True
    script_dirs: list[str] = field(default_factory=lambda: ["tests/e2e/scripts"])
    recording_dir: str = "tests/e2e/recordings"
    verbose: bool = False
    record: bool = False

Hook Implementations¶

[ ] Implement pytest_addoption(parser: pytest.Parser) -> None:
--teltest-verbose — action="store_true", help: "Print full transcript to stdout on test failure"
--teltest-record — action="store_true", help: "Save .teltest recording for every test"
--teltest-timeout-multiplier — type=float, default=1.0, help: "Multiply all expect timeouts (useful for CI)"
--teltest-script-dir — type=str, default="tests/e2e/scripts", help: "Directory for YAML test scripts"
[ ] Implement pytest_configure(config: pytest.Config) -> None:
Register e2e marker via config.addinivalue_line("markers", "e2e: End-to-end test requiring a running server")
Read [tool.teltest] section from pyproject.toml (via tomllib or config stash)
Override with environment variables: TELTEST_TIMEOUT_MULTIPLIER, TELTEST_TIMEOUT, TELTEST_STRIP_ANSI, TELTEST_SCRIPT_DIR, TELTEST_RECORDING_DIR
Override with CLI options (highest precedence)
Store resulting TelTestConfig on config.stash using a pytest.StashKey[TelTestConfig]
[ ] Implement pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[None]) with @pytest.hookimpl(hookwrapper=True):
Yield to get the report
On failure (report.failed and call.when == "call"):
- Extract transcript from item._teltest_client if present (set by fixtures)
- Attach transcript as extra section via report.sections.append(("TelTest Transcript", transcript_text))
- If TelTestConfig.verbose is True, also print transcript to stdout
- If TelTestConfig.record is True, save recording to TelTestConfig.recording_dir
[ ] Implement pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
Support deselection of YAML scripts based on marker expressions
[ ] Implement pytest_collect_file(parent: pytest.Collector, file_path: Path) -> YAMLTestFile | None:
Check if file_path suffix is .yaml or .yml
Check if file_path is within any configured script_dirs
Return YAMLTestFile.from_parent(parent, path=file_path) if both conditions met, else None

YAML Test Collection Classes¶

[ ] Implement YAMLTestFile(pytest.File):
def collect(self) -> Iterator[YAMLTestItem]:
- Load YAML file via load_script(self.path) (from section 3.5)
- Yield a single YAMLTestItem per script (one script per file)
- On parse error, yield a YAMLTestItem that will fail with the parse error on runtest()
[ ] Implement YAMLTestItem(pytest.Item):
def __init__(self, name: str, parent: YAMLTestFile, script: TelTestScript) -> None — store script reference
def runtest(self) -> None — create ScriptRunner, execute runner.run(self.script), raise on failure
def repr_failure(self, excinfo: pytest.ExceptionInfo[BaseException]) -> str — format ScriptResult failure with step index and transcript context
def reportinfo(self) -> tuple[Path, int | None, str] — return (self.path, None, self.name)
Convert script.tags to pytest markers in __init__ via self.add_marker(tag) for each tag

Fixtures¶

[ ] Implement teltest_config fixture (session-scoped) in packages/teltest/src/teltest/pytest_plugin.py:

@pytest.fixture(scope="session")
def teltest_config(request: pytest.FixtureRequest) -> TelTestConfig:
    return request.config.stash[_teltest_config_key]

Tests¶

[ ] Write tests in packages/teltest/tests/test_pytest_plugin.py:
[ ] test_e2e_marker_registered — verify e2e marker is in registered markers after pytest_configure
[ ] test_addoption_registers_verbose — verify --teltest-verbose option exists
[ ] test_addoption_registers_record — verify --teltest-record option exists
[ ] test_addoption_registers_timeout_multiplier — verify --teltest-timeout-multiplier option with float type
[ ] test_addoption_registers_script_dir — verify --teltest-script-dir option
[ ] test_configure_reads_pyproject_toml — mock pyproject.toml with [tool.teltest] section, verify values read
[ ] test_configure_env_var_overrides_pyproject — set TELTEST_TIMEOUT_MULTIPLIER=3.0 env var, verify override
[ ] test_timeout_multiplier_applied_to_config — verify CLI --teltest-timeout-multiplier=2.0 sets config field
[ ] test_makereport_attaches_transcript_on_failure — simulate failed test with _teltest_client, verify transcript section added
[ ] test_makereport_noop_on_success — simulate passing test, verify no transcript section
[ ] test_makereport_saves_recording_when_flag_set — simulate failure with record=True, verify .teltest file written
[ ] test_yaml_file_discovered_in_script_dir — place .yaml in script_dir, verify pytest_collect_file returns YAMLTestFile
[ ] test_yaml_file_ignored_outside_script_dir — place .yaml outside script_dir, verify pytest_collect_file returns None
[ ] test_yaml_test_item_has_correct_nodeid — verify collected YAMLTestItem has expected pytest node ID
[ ] test_yaml_test_item_converts_tags_to_markers — script with tags: [smoke, login], verify markers on item
[ ] test_teltest_config_fixture_returns_config — request teltest_config fixture, verify returns TelTestConfig instance

3.2 AuthDriver Protocol¶

Package: teltest | Priority: P0 | Dependencies: Phase 1 (MUDClient)

File: packages/teltest/src/teltest/auth.py

Data Classes¶

[ ] Define AccountInfo dataclass in packages/teltest/src/teltest/auth.py:

@dataclass
class AccountInfo:
    username: str
    password: str
    email: str
    character_name: str | None = None

Protocol¶

[ ] Define AuthDriver as a @runtime_checkable Protocol in packages/teltest/src/teltest/auth.py:

@runtime_checkable
class AuthDriver(Protocol):
    async def register(
        self, client: MUDClient, username: str, password: str, email: str
    ) -> None: ...

    async def login(
        self, client: MUDClient, username: str, password: str
    ) -> None: ...

    async def select_character(
        self, client: MUDClient, character_name: str
    ) -> None: ...

Exceptions¶

[ ] Define AuthenticationError(Exception) in packages/teltest/src/teltest/auth.py:

class AuthenticationError(Exception):
    def __init__(self, stage: str, message: str) -> None:
        self.stage = stage
        self.message = message
        super().__init__(f"Authentication failed at {stage}: {message}")

Tests¶

[ ] Write tests in packages/teltest/tests/test_auth.py:
[ ] test_auth_driver_is_runtime_checkable — verify isinstance(obj, AuthDriver) works for conforming object
[ ] test_concrete_class_satisfies_protocol — create a class implementing all three methods, verify isinstance check passes
[ ] test_incomplete_class_fails_protocol_check — class missing select_character, verify isinstance returns False
[ ] test_account_info_dataclass_fields — verify username, password, email fields exist and are required
[ ] test_account_info_optional_character_name — verify character_name defaults to None
[ ] test_authentication_error_has_stage — create AuthenticationError("login", "bad password"), verify .stage and .message

3.3 MaidAuthDriver¶

Package: MAID repo (NOT teltest) | Priority: P0 | Dependencies: 3.2 (AuthDriver), Phase 1 (MUDClient)

File: tests/e2e/conftest.py

Implementation¶

[ ] Implement MaidAuthDriver class in tests/e2e/conftest.py implementing the AuthDriver protocol:

Class constants:

_DEFAULT_RACE: str = "1"    # Human
_DEFAULT_CLASS: str = "1"   # Warrior
_DEFAULT_GENDER: str = "3"  # Neutral

[ ] async def register(self, client: MUDClient, username: str, password: str, email: str) -> None:
- await client.expect("Please select an option")
- await client.send("R")
- await client.expect("Choose a username")
- await client.send(username)
- await client.expect("Email")
- await client.send(email)
- await client.expect("Choose a password")
- await client.send(password)
- await client.expect("Confirm")
- await client.send(password)
- await client.expect("Account created")
- Wrap in try/except ExpectTimeout → raise AuthenticationError("register", ...)
[ ] async def login(self, client: MUDClient, username: str, password: str) -> None:
- await client.expect("Please select an option")
- await client.send("L")
- await client.expect("Username")
- await client.send(username)
- await client.expect("Password")
- await client.send(password)
- await client.expect("Character Selection")
- Wrap in try/except ExpectTimeout → raise AuthenticationError("login", ...)
[ ] async def select_character(self, client: MUDClient, character_name: str) -> None:
- Try to find existing character by name in character list
- If found, send the selection number
- If not found, create new character:
- await client.send("C")
- await client.expect("Enter character name")
- await client.send(character_name)
- await client.expect("Select your race")
- await client.send(self._DEFAULT_RACE)
- await client.expect("Select your class")
- await client.send(self._DEFAULT_CLASS)
- await client.expect("Select your gender")
- await client.send(self._DEFAULT_GENDER)
- await client.expect("Character Summary")
- await client.send("Y")
- await client.expect("Entering World")
- Wrap in try/except ExpectTimeout → raise AuthenticationError("select_character", ...)

Tests¶

[ ] Write tests in tests/e2e/test_maid_auth_driver.py:
[ ] test_maid_auth_driver_satisfies_protocol — verify isinstance(MaidAuthDriver(), AuthDriver) is True
[ ] test_register_sends_correct_sequence — mock MUDClient, verify send/expect call sequence matches the MAID registration flow
[ ] test_login_sends_correct_sequence — mock MUDClient, verify send/expect sequence matches MAID login flow
[ ] test_select_character_creates_new — mock MUDClient with no existing characters, verify creation sequence
[ ] test_select_character_selects_existing — mock MUDClient with existing character list, verify selection
[ ] test_register_raises_on_duplicate_username — mock MUDClient that times out at "Account created", verify AuthenticationError with stage="register"
[ ] test_login_raises_on_bad_password — mock MUDClient that times out at "Character Selection", verify AuthenticationError with stage="login"

3.4 Fixtures¶

3.4a: teltest Package Fixtures¶

Package: teltest | Priority: P0 | Dependencies: 3.1

File: packages/teltest/src/teltest/pytest_plugin.py

[ ] teltest_config fixture (session-scoped) — returns TelTestConfig from config.stash (already defined in 3.1)

3.4b: MAID-Specific Fixtures¶

Package: MAID repo | Priority: P0 | Dependencies: 3.2 (AuthDriver), 3.3 (MaidAuthDriver), Phase 1 (MUDClient)

File: tests/e2e/conftest.py

Data Classes¶

[ ] Define WorldSnapshot dataclass in tests/e2e/conftest.py:

@dataclass
class WorldSnapshot:
    entities: dict[UUID, dict]
    room_index: dict
    document_collections: dict

[ ] Define MUDServer dataclass in tests/e2e/conftest.py:

@dataclass
class MUDServer:
    host: str
    port: int
    engine: GameEngine
    _task: asyncio.Task[None]
    _tick_event: asyncio.Event

    def is_alive(self) -> bool: ...
    async def wait_ticks(self, n: int = 1) -> None: ...
    async def snapshot_world(self) -> WorldSnapshot: ...
    async def restore_world(self, snapshot: WorldSnapshot) -> None: ...

Helper Functions¶

[ ] Implement _allocate_port() -> int in tests/e2e/conftest.py:
Use socket.bind(('', 0)) trick to get a free ephemeral port

Close socket immediately after reading port number

def _allocate_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("", 0))
        return s.getsockname()[1]

[ ] Implement async def _wait_for_ready(host: str, port: int, timeout: float = 15.0) -> None:
Exponential backoff retry loop attempting asyncio.open_connection(host, port)
On success: close writer, return
On ConnectionRefusedError/OSError: sleep with backoff, retry
On timeout: raise TimeoutError(f"Server not ready on {host}:{port} after {timeout}s")

Fixtures¶

[ ] Implement mud_server fixture (session-scoped, async) in tests/e2e/conftest.py:

@pytest.fixture(scope="session")
async def mud_server() -> AsyncGenerator[MUDServer, None]:

Allocate ephemeral port via _allocate_port()
Configure GameEngine with InMemoryDocumentStore and allocated port
Start engine in background asyncio.Task
Call await _wait_for_ready(host, port) before yielding
Yield MUDServer(host="localhost", port=port, engine=engine, _task=task, _tick_event=event)
On teardown: stop engine, cancel task
Register atexit handler as zombie prevention

[ ] Implement raw_client fixture (function-scoped, async) in tests/e2e/conftest.py:

@pytest.fixture
async def raw_client(mud_server: MUDServer) -> AsyncGenerator[MUDClient, None]:

Check mud_server.is_alive() → pytest.skip("Server crashed") if dead
Create MUDClient(host=mud_server.host, port=mud_server.port)
await client.connect()
Yield client
await client.disconnect()

[ ] Implement registered_account fixture (session-scoped, async) in tests/e2e/conftest.py:

@pytest.fixture(scope="session")
async def registered_account(mud_server: MUDServer) -> AccountInfo:

Create temporary MUDClient, connect
Use MaidAuthDriver().register(client, username, password, email) with generated unique credentials
Disconnect client
Return AccountInfo(username=username, password=password, email=email)

[ ] Implement mud_client fixture (function-scoped, async) in tests/e2e/conftest.py:

@pytest.fixture
async def mud_client(
    mud_server: MUDServer, registered_account: AccountInfo
) -> AsyncGenerator[MUDClient, None]:

Check mud_server.is_alive() → skip if dead
Create MUDClient, connect
Login via MaidAuthDriver().login(client, account.username, account.password)
Select character via MaidAuthDriver().select_character(client, account.character_name or "TestChar")
Set client._teltest_client = client for transcript hook
Yield client
Disconnect

[ ] Implement _isolate_world fixture (function-scoped, autouse, async) in tests/e2e/conftest.py:

@pytest.fixture(autouse=True)
async def _isolate_world(mud_server: MUDServer) -> AsyncGenerator[None, None]:
    snapshot = await mud_server.snapshot_world()
    yield
    await mud_server.restore_world(snapshot)

[ ] Implement unique_name fixture (function-scoped) in tests/e2e/conftest.py:

@pytest.fixture
def unique_name() -> str:
    return f"test_{uuid4().hex[:8]}"

[ ] Implement logged_in_client async context manager helper in tests/e2e/conftest.py:

@asynccontextmanager
async def logged_in_client(
    mud_server: MUDServer, name: str
) -> AsyncIterator[MUDClient]:

Create MUDClient, connect
Register unique account, login, select character with given name
Yield client
Disconnect on exit

Tests¶

[ ] Write tests in tests/e2e/test_fixtures.py:
[ ] test_mud_server_fixture_starts_engine — verify mud_server.engine is not None and port is > 0
[ ] test_mud_server_is_alive — verify mud_server.is_alive() returns True after startup
[ ] test_mud_server_wait_ticks — call mud_server.wait_ticks(2), verify returns without timeout
[ ] test_raw_client_connects — verify raw_client.is_connected is True
[ ] test_raw_client_skips_if_server_dead — mock is_alive() returning False, verify pytest.skip raised
[ ] test_mud_client_is_logged_in — verify mud_client receives game prompt after setup
[ ] test_registered_account_has_credentials — verify registered_account.username and .password are non-empty strings
[ ] test_isolate_world_restores_state — create entity in test, verify it is gone in next test
[ ] test_unique_name_is_unique — call fixture twice, verify different values
[ ] test_logged_in_client_context_manager — use logged_in_client, verify client is connected and logged in
[ ] test_allocate_port_returns_free_port — call _allocate_port(), verify returned port > 0 and < 65536
[ ] test_wait_for_ready_succeeds — start a TCP listener, call _wait_for_ready, verify returns
[ ] test_wait_for_ready_timeout_raises — call _wait_for_ready on unused port with short timeout, verify TimeoutError
[ ] test_world_snapshot_captures_entities — create entities, snapshot, verify snapshot.entities contains them
[ ] test_world_restore_removes_new_entities — snapshot, create entity, restore, verify entity gone

3.5 YAML Script DSL Schema¶

Package: teltest | Priority: P1 | Dependencies: none (pure Pydantic models)

File: packages/teltest/src/teltest/script/schema.py

Pydantic v2 Models¶

[ ] Define ExpectNotSpec(BaseModel) in packages/teltest/src/teltest/script/schema.py:
```
class ExpectNotSpec(BaseModel):
    pattern: str
    until: str
```
[ ] Define ScriptStep(BaseModel) in packages/teltest/src/teltest/script/schema.py:
```
                           
```
name="__codelineno-18-1" href="#__codelineno-18-1">class ScriptStep(BaseModel): send: str | None = None expect: str | None = None expect_re: str | None = None expect_all: list[str] | None = None expect_any: list[str] | None = None expect_not: ExpectNotSpec | None = None expect_prompt: str | None = None timeout: float | None = None delay: float | None = None note: str | None = None group: str | None = None steps: list[ScriptStep] | None = None branch: str | None = None cases: dict[str, list[ScriptStep]] | None = None @model_validator(mode="after") def validate_step_has_action(self) -> ScriptStep: ... # Must have at least one of: send, expect, expect_re, expect_all, # expect_any, expect_not, expect_prompt, delay, note, group, branch @model_validator(mode="after") def validate_branch_has_cases(self) -> ScriptStep: ... # branch requires cases @model_validator(mode="after") def validate_group_has_steps(self) -> ScriptStep: ... # group requires steps

[ ] Define AccountSetup(BaseModel):

class AccountSetup(BaseModel):
    username: str
    password: str
    email: str = "test@example.com"
    register: bool = False

[ ] Define CharacterSetup(BaseModel):

class CharacterSetup(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    name: str
    race: str = "human"
    class_type: str = Field(default="warrior", alias="class")
    gender: str = "neutral"
    select: int | None = None

[ ] Define ScriptSetup(BaseModel):

class ScriptSetup(BaseModel):
    account: AccountSetup | None = None
    character: CharacterSetup | None = None

[ ] Define TeardownStep(BaseModel):

class TeardownStep(BaseModel):
    send: str

[ ] Define TelTestScript(BaseModel):

class TelTestScript(BaseModel):
    name: str
    description: str = ""
    tags: list[str] = []
    timeout: float = 5.0
    setup: ScriptSetup | None = None
    steps: list[ScriptStep]
    teardown: list[TeardownStep] = []

    @model_validator(mode="after")
    def validate_steps_not_empty(self) -> TelTestScript: ...

Loader Functions¶

[ ] Implement def load_script(path: Path) -> TelTestScript in packages/teltest/src/teltest/script/schema.py:
Read YAML file via yaml.safe_load()
Validate via TelTestScript.model_validate(data)
Raise ValueError on invalid YAML or schema errors
[ ] Implement def load_scripts(directory: Path) -> list[TelTestScript]:
Discover all .yaml and .yml files in directory
Load each via load_script()
Return list of valid scripts
[ ] Implement def validate_scripts(directory: Path) -> list[tuple[Path, Exception]]:
Attempt load_script() on each YAML file
Collect (path, exception) tuples for failures
Return empty list if all valid

Tests¶

[ ] Write tests in packages/teltest/tests/test_script_schema.py:
[ ] test_minimal_script_name_and_steps — script with only name and one send step
[ ] test_full_script_all_fields — script with all fields populated
[ ] test_step_send_only — step with only send field
[ ] test_step_expect_only — step with only expect field
[ ] test_step_send_with_expect — step with both send and expect
[ ] test_step_expect_re — step with expect_re regex pattern
[ ] test_step_expect_all — step with expect_all list
[ ] test_step_expect_any — step with expect_any list
[ ] test_step_expect_not_with_until — step with expect_not containing pattern and until
[ ] test_step_expect_not_missing_until_fails — expect_not without until raises validation error
[ ] test_step_expect_prompt — step with expect_prompt string
[ ] test_step_delay — step with only delay float
[ ] test_step_note — step with only note string
[ ] test_step_group_with_nested_steps — step with group name and steps list
[ ] test_step_group_without_steps_fails — group without steps raises validation error
[ ] test_step_branch_with_cases — step with branch text and cases dict
[ ] test_step_branch_without_cases_fails — branch without cases raises validation error
[ ] test_step_no_action_fails_validation — empty step (no fields set) raises validation error
[ ] test_setup_account_register — setup with account.register: true
[ ] test_setup_account_login_only — setup with account but register: false
[ ] test_setup_character_with_class_alias — character with class: "mage" field alias
[ ] test_setup_character_select_existing — character with select: 1
[ ] test_teardown_steps — script with teardown list
[ ] test_empty_steps_fails_validation — script with steps: [] raises validation error
[ ] test_script_tags — script with tags: ["smoke", "login"]
[ ] test_script_timeout_override — script with timeout: 10.0
[ ] test_load_script_from_yaml_file — write YAML to tmp file, load via load_script(), verify parsed
[ ] test_load_script_invalid_yaml_raises — write invalid YAML, verify ValueError raised
[ ] test_load_scripts_directory — write multiple YAML files to tmp dir, verify all loaded
[ ] test_validate_scripts_returns_errors — directory with one valid and one invalid script, verify error list
[ ] test_validate_scripts_empty_dir_returns_empty — empty directory returns empty error list

3.6 Script Runner¶

Package: teltest | Priority: P1 | Dependencies: 3.5 (schema), 3.2 (AuthDriver), Phase 2 (expect engine)

File: packages/teltest/src/teltest/script/runner.py

Data Classes¶

[ ] Define StepResult dataclass in packages/teltest/src/teltest/script/runner.py:

@dataclass
class StepResult:
    step_index: int
    step_type: str
    passed: bool
    elapsed: float
    error: str | None = None
    note: str | None = None

[ ] Define ScriptResult dataclass:

@dataclass
class ScriptResult:
    script_name: str
    passed: bool
    steps_completed: int
    steps_total: int
    step_results: list[StepResult]
    elapsed: float
    transcript: list[TranscriptEntry]
    error: Exception | None = None
    failed_step: int | None = None

Exceptions¶

[ ] Define ScriptRunError(Exception):

class ScriptRunError(Exception):
    def __init__(self, step_index: int, step_type: str, cause: Exception) -> None:
        self.step_index = step_index
        self.step_type = step_type
        self.cause = cause
        super().__init__(f"Step {step_index} ({step_type}) failed: {cause}")

ScriptRunner Class¶

[ ] Implement ScriptRunner class in packages/teltest/src/teltest/script/runner.py:

class ScriptRunner:
    def __init__(
        self,
        client: MUDClient,
        auth_driver: AuthDriver | None = None,
        default_timeout: float = 5.0,
        timeout_multiplier: float = 1.0,
    ) -> None: ...

[ ] async def run(self, script: TelTestScript) -> ScriptResult:
- Record start time
- Run setup if present (wrapped in try/except)
- Run steps, collecting StepResult list
- Run teardown always (even on failure, in finally block)
- Build and return ScriptResult
[ ] async def _run_setup(self, setup: ScriptSetup) -> None:
- If setup.account is set and auth_driver is not None:
- If setup.account.register is True: call auth_driver.register(...)
- Else: call auth_driver.login(...)
- If setup.character is set and auth_driver is not None:
- Call auth_driver.select_character(client, setup.character.name)
[ ] async def _run_steps(self, steps: list[ScriptStep], results: list[StepResult]) -> None:
- Iterate steps with index, call _run_step() for each
- On failure: raise ScriptRunError
[ ] async def _run_step(self, step: ScriptStep, index: int) -> StepResult:
- Dispatch to appropriate _execute_* method based on which fields are set
- Record elapsed time and success/failure
[ ] async def _execute_send(self, step: ScriptStep) -> None:
- await self._client.send(step.send)
[ ] async def _execute_expect(self, step: ScriptStep) -> None:
- await self._client.expect(step.expect, timeout=self._effective_timeout(step))
[ ] async def _execute_expect_re(self, step: ScriptStep) -> None:
- await self._client.expect(re.compile(step.expect_re), timeout=self._effective_timeout(step))
[ ] async def _execute_expect_all(self, step: ScriptStep) -> None:
- Call await self._client.expect(pattern, timeout=...) for each pattern in step.expect_all
[ ] async def _execute_expect_any(self, step: ScriptStep) -> None:
- await self._client.expect_any(step.expect_any, timeout=self._effective_timeout(step))
[ ] async def _execute_expect_not(self, step: ScriptStep) -> None:
- await self._client.expect_not(step.expect_not.pattern, until=step.expect_not.until, timeout=self._effective_timeout(step))
[ ] async def _execute_expect_prompt(self, step: ScriptStep) -> None:
- await self._client.expect_prompt(step.expect_prompt, timeout=self._effective_timeout(step))
[ ] async def _execute_delay(self, step: ScriptStep) -> None:
- await asyncio.sleep(step.delay)
[ ] async def _execute_branch(self, step: ScriptStep, results: list[StepResult]) -> None:
- await self._client.send(step.branch)
- index, match = await self._client.expect_any(list(step.cases.keys()), timeout=self._effective_timeout(step))
- Get matching case key, run its steps via _run_steps()
- Raise ScriptRunError if no case matches
[ ] async def _execute_group(self, step: ScriptStep, results: list[StepResult]) -> None:
- Recursively call _run_steps(step.steps, results)
[ ] async def _run_teardown(self, steps: list[TeardownStep]) -> None:
- For each step: await self._client.send(step.send)
- Always runs, even on failure (called from finally block)
- Exceptions during teardown are logged but not raised
[ ] def _effective_timeout(self, step: ScriptStep) -> float:
- Return step.timeout if set, else self._default_timeout * self._timeout_multiplier

Tests¶

[ ] Write tests in packages/teltest/tests/test_script_runner.py:
[ ] test_run_simple_send_expect — script with send + expect step, mock client, verify calls
[ ] test_run_send_without_expect — step with only send, verify client.send called
[ ] test_run_expect_without_send — step with only expect, verify client.expect called
[ ] test_run_expect_re — step with expect_re, verify re.compile() passed to client.expect
[ ] test_run_expect_all_any_order — step with expect_all: ["a", "b"], verify both expected
[ ] test_run_expect_any_first_match — step with expect_any, verify client.expect_any called
[ ] test_run_expect_not_passes — step with expect_not, mock client.expect_not succeeds
[ ] test_run_expect_not_fails_on_match — mock client.expect_not raises UnexpectedMatch
[ ] test_run_expect_prompt — step with expect_prompt, verify client.expect_prompt called
[ ] test_run_delay_step — step with delay: 0.1, verify asyncio.sleep called
[ ] test_run_note_step_recorded — step with note, verify StepResult.note is set
[ ] test_run_group_executes_nested — group step with nested send/expect, verify all executed
[ ] test_run_branch_selects_matching_case — branch step, mock expect_any returns index 0, verify case 0 steps executed
[ ] test_run_branch_no_match_raises — branch step, mock expect_any raises timeout, verify ScriptRunError
[ ] test_run_setup_registers_account — script with setup.account.register: true, verify auth_driver.register called
[ ] test_run_setup_logs_in — script with setup.account.register: false, verify auth_driver.login called
[ ] test_run_setup_creates_character — script with setup.character, verify auth_driver.select_character called
[ ] test_run_setup_skipped_when_none — script with no setup, verify no auth calls
[ ] test_run_teardown_always_runs — script with teardown, verify teardown send calls after steps
[ ] test_run_teardown_runs_after_failure — step fails, verify teardown still executes
[ ] test_script_result_passed_on_success — all steps pass, verify result.passed is True
[ ] test_script_result_failed_on_error — step fails, verify result.passed is False and result.failed_step set
[ ] test_script_result_has_transcript — verify result.transcript is populated from client
[ ] test_script_result_step_count — verify steps_completed and steps_total are correct
[ ] test_effective_timeout_uses_multiplier — set timeout_multiplier=2.0, verify effective timeout doubled
[ ] test_effective_timeout_uses_step_override — step with timeout: 10.0, verify overrides default
[ ] test_timeout_propagated_to_expect — verify timeout value passed to client.expect call
[ ] test_global_script_timeout_enforced — script with timeout: 1.0, verify overall execution bounded

3.7 Session Recorder¶

Package: teltest | Priority: P2 | Dependencies: Phase 1 (MUDClient)

File: packages/teltest/src/teltest/recorder.py

Data Classes¶

[ ] Define RecordedEvent dataclass in packages/teltest/src/teltest/recorder.py:

@dataclass
class RecordedEvent:
    t: float  # seconds since recording start
    type: Literal["send", "recv", "gmcp_send", "gmcp_recv"]
    text: str

[ ] Define SessionRecording dataclass:

@dataclass
class SessionRecording:
    recorded_at: str  # ISO 8601
    server: str       # "host:port"
    duration: float
    events: list[RecordedEvent]
    metadata: dict[str, str] = field(default_factory=dict)

    def save(self, path: Path) -> None: ...
        # Serialize to .teltest JSON file

    @classmethod
    def load(cls, path: Path) -> SessionRecording: ...
        # Deserialize from .teltest JSON file

SessionRecorder Class¶

[ ] Implement SessionRecorder class in packages/teltest/src/teltest/recorder.py:

class SessionRecorder:
    def __init__(self, client: MUDClient) -> None:
        self._client = client
        self._events: list[RecordedEvent] = []
        self._start_time: float | None = None
        self._recording: bool = False

[ ] def start(self) -> None — set self._start_time = time.monotonic(), set self._recording = True
[ ] def stop(self) -> SessionRecording — set self._recording = False, poll final transcript, build SessionRecording from _events with duration = time.monotonic() - self._start_time
[ ] def _poll_transcript(self) -> None — sync client transcript entries to _events, converting TranscriptEntry to RecordedEvent with relative timestamps
[ ] Property is_recording → bool: return self._recording

Tests¶

[ ] Write tests in packages/teltest/tests/test_recorder.py:
[ ] test_recorder_start_sets_recording — call start(), verify is_recording is True
[ ] test_recorder_stop_returns_recording — start, simulate activity, stop, verify SessionRecording returned
[ ] test_recorder_captures_send_events — mock client with send transcript entries, verify send events in recording
[ ] test_recorder_captures_recv_events — mock client with recv transcript entries, verify recv events in recording
[ ] test_recording_save_creates_json_file — save to tmp path, verify file exists and is valid JSON
[ ] test_recording_load_restores_events — save then load, verify events match
[ ] test_recording_save_load_roundtrip — create recording, save, load, compare all fields
[ ] test_recording_timestamps_monotonic — verify event.t values are monotonically increasing
[ ] test_recording_metadata_preserved — recording with metadata dict, save/load, verify preserved
[ ] test_recording_server_field — verify server field is "host:port" format
[ ] test_recording_duration_calculated — verify duration equals time between start and stop
[ ] test_recorder_not_recording_after_stop — stop recorder, verify is_recording is False

3.8 Recording-to-Script Converter¶

Package: teltest | Priority: P2 | Dependencies: 3.5 (schema), 3.7 (recorder)

File: packages/teltest/src/teltest/script/converter.py

Data Classes¶

[ ] Define ConversionOptions dataclass in packages/teltest/src/teltest/script/converter.py:

@dataclass
class ConversionOptions:
    min_expect_length: int = 10  # Skip short recv lines
    fuzzy_numbers: bool = True    # Replace literal numbers with \d+ in expect_re
    detect_prompts: bool = True
    prompt_pattern: str = r"^>|^\w+> "

RecordingConverter Class¶

[ ] Implement RecordingConverter class:

class RecordingConverter:
    def __init__(self, options: ConversionOptions | None = None) -> None:
        self._options = options or ConversionOptions()

[ ] def convert(self, recording: SessionRecording) -> TelTestScript:
- Group events into send/recv pairs
- Generate ScriptStep for each pair
- Infer script name from recording metadata
- Return valid TelTestScript
[ ] def _group_recv_events(self, events: list[RecordedEvent]) -> list[list[RecordedEvent]]:
- Group consecutive recv events between send events
[ ] def _extract_expect_pattern(self, recv_group: list[RecordedEvent]) -> str:
- Pick the most distinctive line from a recv group (longest non-prompt line)
[ ] def _is_prompt(self, text: str) -> bool:
- Check text against self._options.prompt_pattern regex
[ ] def _make_fuzzy(self, text: str) -> str:
- Replace literal numbers with \d+
- Replace UUIDs with [0-9a-f-]+ pattern
[ ] def _infer_script_name(self, recording: SessionRecording) -> str:
- Generate name from recording metadata or filename

Tests¶

[ ] Write tests in packages/teltest/tests/test_script_converter.py:
[ ] test_convert_simple_send_recv — recording with one send + one recv, verify produces one send/expect step
[ ] test_convert_multiple_recv_grouped — multiple recv between sends, verify grouped into single expect
[ ] test_convert_preserves_send_order — verify send steps appear in chronological order
[ ] test_convert_generates_valid_script — converted script validates against TelTestScript schema
[ ] test_convert_detects_prompts — recv matching prompt pattern generates expect_prompt step
[ ] test_convert_fuzzy_numbers — recv containing "Level 5" generates expect_re with Level \d+
[ ] test_convert_skips_short_recv — recv shorter than min_expect_length is skipped
[ ] test_convert_infers_script_name — verify script name derived from recording metadata
[ ] test_convert_empty_recording — empty events list produces script with no steps (or raises)
[ ] test_convert_recv_only_recording — only recv events (no sends), verify expect-only steps
[ ] test_conversion_options_defaults — verify default option values
[ ] test_conversion_options_custom — custom options applied during conversion

3.9 CLI¶

Package: teltest | Priority: P1 | Dependencies: 3.5 (schema), 3.6 (runner), 3.7 (recorder), 3.8 (converter)

File: packages/teltest/src/teltest/cli.py

Entry Point¶

[ ] Register CLI entry point in packages/teltest/pyproject.toml:
```
[project.scripts]
teltest = "teltest.cli:app"
```

Commands¶

[ ] Define Typer app in packages/teltest/src/teltest/cli.py:

app = typer.Typer(name="teltest", help="E2E testing framework for MUDs")

[ ] Implement validate command:

@app.command()
def validate(
    directory: Path = typer.Argument(..., help="Directory containing YAML scripts"),
) -> None:

Load and validate all YAML scripts via validate_scripts(directory)
Print errors with file paths and error messages
Exit 0 if all valid, exit 1 on any failure

[ ] Implement run command:

@app.command()
def run(
    directory: Path = typer.Argument(..., help="Directory containing YAML scripts"),
    server: str = typer.Option("localhost:4000", help="Server host:port"),
    timeout: float = typer.Option(5.0, help="Default expect timeout"),
    timeout_multiplier: float = typer.Option(1.0, help="Timeout multiplier"),
) -> None:

Parse server string into host:port
Connect MUDClient to server
Load scripts via load_scripts(directory)
Run each via ScriptRunner(client, default_timeout=timeout, timeout_multiplier=timeout_multiplier)
Print results summary (passed/failed/total)
Exit 1 on any failure

[ ] Implement record command:

@app.command()
def record(
    host: str = typer.Argument(..., help="Server host"),
    port: int = typer.Argument(..., help="Server port"),
    output: Path = typer.Option(..., "-o", "--output", help="Output .teltest file"),
) -> None:

Connect MUDClient with SessionRecorder
Forward stdin to client.send()
Print received text to stdout
On Ctrl+C (KeyboardInterrupt): stop recorder, save recording to output path

[ ] Implement convert command:

@app.command()
def convert(
    recording_path: Path = typer.Argument(..., metavar="FILE", help="Input .teltest recording"),
    output: Path = typer.Option(..., "-o", "--output", help="Output YAML script path"),
) -> None:

Load recording via SessionRecording.load(recording_path)
Convert via RecordingConverter().convert(recording)
Dump as YAML to output path

Tests¶

[ ] Write tests in packages/teltest/tests/test_cli.py:
[ ] test_validate_valid_scripts_exit_0 — create valid YAML scripts in tmp dir, invoke validate, verify exit code 0
[ ] test_validate_invalid_scripts_exit_1 — create invalid YAML, invoke validate, verify exit code 1
[ ] test_validate_empty_dir_exit_0 — empty directory, verify exit code 0
[ ] test_validate_prints_error_details — invalid script, verify error message in output
[ ] test_run_executes_scripts — mock MUDClient, create valid scripts, invoke run, verify scripts executed
[ ] test_run_reports_failures — mock MUDClient with expect failure, verify failure reported
[ ] test_run_parses_server_host_port — invoke with --server=myhost:5000, verify client created with correct host/port
[ ] test_record_saves_on_interrupt — mock stdin/MUDClient, simulate KeyboardInterrupt, verify .teltest file saved
[ ] test_convert_produces_yaml — create .teltest recording, invoke convert, verify .yaml output exists
[ ] test_convert_output_validates — converted output passes load_script() validation
[ ] test_cli_help_text — invoke --help, verify "E2E testing framework" in output
[ ] test_cli_version — verify version output (if implemented)

3.10 Integration Tests¶

Package: teltest + MAID repo | Priority: P1 | Dependencies: 3.1–3.9

File: packages/teltest/tests/test_integration.py

End-to-End Pipeline Tests¶

[ ] Write integration tests in packages/teltest/tests/test_integration.py:
[ ] test_yaml_script_discovery_and_execution — create temp directory with valid YAML script, run pytest programmatically with --co to verify collection, then execute and verify pass/fail
[ ] test_yaml_script_with_setup_and_teardown — YAML script with setup (account + character) and teardown, verify full lifecycle executed
[ ] test_yaml_script_with_branch — YAML script using branch/cases, mock client to return specific pattern, verify correct case executed
[ ] test_yaml_script_failure_reports_step — YAML script where a step fails, verify failure report includes step index and transcript context
[ ] test_record_convert_validate_run_pipeline — record a mock session → convert to YAML → validate → run, verify full pipeline
[ ] test_multiple_scripts_in_directory — multiple YAML scripts in one directory, verify all discovered and executed
[ ] test_script_tags_become_pytest_markers — script with tags: ["smoke"], verify pytest marker smoke is on the collected item

Dependencies Summary¶

Phase 1 (MUDClient)
  │
  ├──► Phase 2 (Expectation Engine)
  │       │
  │       ├──► 3.1 pytest Plugin ◄─── Phase 1
  │       │       │
  │       │       ├──► 3.4a teltest Fixtures
  │       │       │
  │       │       └──► 3.10 Integration Tests
  │       │
  │       └──► 3.6 Script Runner ◄─── 3.5 Schema, 3.2 AuthDriver
  │               │
  │               └──► 3.9 CLI ◄─── 3.7 Recorder, 3.8 Converter
  │
  ├──► 3.2 AuthDriver Protocol
  │       │
  │       ├──► 3.3 MaidAuthDriver
  │       │       │
  │       │       └──► 3.4b MAID Fixtures
  │       │
  │       └──► 3.6 Script Runner
  │
  ├──► 3.7 Session Recorder
  │       │
  │       └──► 3.8 Recording Converter ◄─── 3.5 Schema
  │
  └──► 3.5 YAML Script DSL Schema (no dependencies — pure Pydantic)

Section parallelism:
  • 3.2 + 3.5 + 3.7 can start immediately (no inter-dependencies)
  • 3.1 can start once Phase 1 + Phase 2 are done
  • 3.3 depends on 3.2
  • 3.4b depends on 3.2 + 3.3
  • 3.6 depends on 3.2 + 3.5
  • 3.8 depends on 3.5 + 3.7
  • 3.9 depends on 3.5 + 3.6 + 3.7 + 3.8
  • 3.10 depends on all of 3.1–3.9

Success Criteria¶

[ ] pytest plugin registers e2e marker and all CLI options
[ ] [tool.teltest] config in pyproject.toml is read and merged with env vars
[ ] YAML scripts are auto-discovered as pytest test items with correct node IDs
[ ] Transcript attached to pytest failure report when --teltest-verbose set
[ ] AuthDriver protocol is runtime_checkable and MaidAuthDriver satisfies it
[ ] MaidAuthDriver drives the complete MAID login flow (register → login → select character)
[ ] mud_server fixture starts engine on ephemeral port with readiness probe
[ ] _isolate_world fixture snapshot/restores world state between tests
[ ] All YAML DSL step types (send, expect, expect_re, expect_all, expect_any, expect_not, expect_prompt, delay, branch, group, note) execute correctly
[ ] Script runner teardown always runs even on failure
[ ] teltest validate catches schema errors before execution
[ ] teltest run executes scripts standalone against a server
[ ] teltest record captures interactive sessions to .teltest format
[ ] teltest convert produces valid YAML scripts from recordings
[ ] MyPy strict mode passes for all teltest modules
[ ] Test coverage >90% for all teltest package code

Resource Allocation¶

Role	FTE	Primary Focus
Framework Developer	1.0	pytest plugin, fixtures, AuthDriver, YAML schema, script runner
Tools Developer	0.5	Session recorder, recording converter, CLI commands
Integration Developer	0.5	MaidAuthDriver, MAID-specific fixtures, world snapshot/restore
QA Engineer	0.5	Test authoring, integration tests, CI pipeline validation

Prerequisites / Blockers from Other Design Docs¶

Phase 1 (MUDClient) must be complete. The MUDClient class with connect(), disconnect(), send(), and the TranscriptEntry dataclass are required by all sections in this phase.
Phase 2 (Expectation Engine) must be complete. The expect(), expect_prompt(), expect_sequence(), expect_any(), expect_not() methods and the ExpectTimeout, ConnectionClosed, UnexpectedMatch exception classes are required by the script runner (3.6) and fixtures (3.4b).
No external blockers beyond Phases 1 and 2. This phase is self-contained within the teltest package and the MAID repo's tests/e2e/ directory.
Phase 4 (CI & Advanced Features) consumes this phase's output but is not required by it. GMCP assertion support, pytest-xdist parallel execution, and the CI workflow are all Phase 4 concerns.