"""Step definitions for tui_session_export_import.feature. Tests for: - Session.as_export_markdown() domain method - CLI export --format md flag - TUI TuiCommandRouter /session export and /session import commands """ from __future__ import annotations import json import os import tempfile from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from unittest.mock import MagicMock from behave import given, then, when from behave.runner import Context from typer.testing import CliRunner from ulid import ULID from cleveragents.cli.commands import session as session_mod from cleveragents.cli.commands.session import app as session_app from cleveragents.domain.models.core.session import ( MessageRole, Session, SessionMessage, SessionTokenUsage, ) from cleveragents.tui.commands import TuiCommandRouter # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_session( *, session_id: str | None = None, actor_name: str | None = None, messages: list[SessionMessage] | None = None, linked_plan_ids: list[str] | None = None, ) -> Session: return Session( session_id=session_id or str(ULID()), actor_name=actor_name, namespace="local", messages=messages or [], linked_plan_ids=linked_plan_ids or [], token_usage=SessionTokenUsage( input_tokens=10, output_tokens=5, estimated_cost=0.001, ), created_at=datetime(2026, 1, 1, 12, 0, 0), updated_at=datetime(2026, 1, 1, 12, 30, 0), ) def _make_message( role: MessageRole = MessageRole.USER, content: str = "Hello", sequence: int = 0, ) -> SessionMessage: return SessionMessage( message_id=str(ULID()), role=role, content=content, sequence=sequence, timestamp=datetime(2026, 1, 1, 12, sequence, 0), ) @dataclass class FakePersona: name: str @dataclass class FakePersonaRegistry: _personas: list[FakePersona] = field(default_factory=list) personas_dir: Path = field(default_factory=lambda: Path("/tmp/fake-personas")) def list_personas(self) -> list[FakePersona]: return list(self._personas) @dataclass class FakePersonaState: _active: dict[str, FakePersona] = field(default_factory=dict) def set_active_persona(self, session_id: str, name: str) -> FakePersona: persona = FakePersona(name=name) self._active[session_id] = persona return persona def active_name(self, session_id: str) -> str | None: p = self._active.get(session_id) return p.name if p else None # --------------------------------------------------------------------------- # Domain model: as_export_markdown() # --------------------------------------------------------------------------- @given("a session with two messages for markdown export") def step_session_two_messages(context: Context) -> None: msgs = [ _make_message(MessageRole.USER, "Hello from user", 0), _make_message(MessageRole.ASSISTANT, "Hello from assistant", 1), ] context.md_session = _make_session(messages=msgs) @given("a session with no messages for markdown export") def step_session_no_messages(context: Context) -> None: context.md_session = _make_session() @given('a session with actor "{actor}" for markdown export') def step_session_with_actor(context: Context, actor: str) -> None: context.md_session = _make_session(actor_name=actor) @given("a session with linked plans for markdown export") def step_session_with_linked_plans(context: Context) -> None: context.md_session = _make_session(linked_plan_ids=[str(ULID()), str(ULID())]) @when("I call as_export_markdown on the session") def step_call_as_export_markdown(context: Context) -> None: context.md_output = context.md_session.as_export_markdown() @then('the markdown output should start with "{prefix}"') def step_md_starts_with(context: Context, prefix: str) -> None: assert context.md_output.startswith(prefix), ( f"Expected markdown to start with {prefix!r}, got: {context.md_output[:80]!r}" ) @then('the markdown output should contain "{text}"') def step_md_contains(context: Context, text: str) -> None: assert text in context.md_output, ( f"Expected {text!r} in markdown output.\nGot:\n{context.md_output[:500]}" ) # --------------------------------------------------------------------------- # CLI export: --format md # --------------------------------------------------------------------------- @given("there is a mocked session for markdown export") def step_mocked_session_for_md_export(context: Context) -> None: context.runner = CliRunner() context.mock_service = MagicMock() session_id = str(ULID()) context.session_id = session_id msgs = [ _make_message(MessageRole.USER, "Test user message", 0), _make_message(MessageRole.ASSISTANT, "Test assistant reply", 1), ] session = _make_session(session_id=session_id, messages=msgs) export_data = session.as_export_dict() context.mock_service.export_session.return_value = export_data context.mock_service.get.return_value = session session_mod._service = context.mock_service def _teardown_mock_service(context: Context) -> None: session_mod._service = None @when("I run session CLI export with --format md and no output file") def step_export_md_stdout(context: Context) -> None: context.result = context.runner.invoke( session_app, ["export", context.session_id, "--format", "md"] ) _teardown_mock_service(context) @when("I run session CLI export with --format md to a temp file") def step_export_md_to_file(context: Context) -> None: fd, path = tempfile.mkstemp(suffix=".md") os.close(fd) os.unlink(path) # Remove so export can create it context.export_md_path = path context.result = context.runner.invoke( session_app, ["export", context.session_id, "--format", "md", "--output", path], ) _teardown_mock_service(context) @when("I run session CLI export with no format flag") def step_export_default_format(context: Context) -> None: context.result = context.runner.invoke(session_app, ["export", context.session_id]) _teardown_mock_service(context) @when("I run session CLI export with --format xml") def step_export_invalid_format(context: Context) -> None: context.result = context.runner.invoke( session_app, ["export", context.session_id, "--format", "xml"] ) _teardown_mock_service(context) @then("the md export CLI result code should be zero") def step_cli_md_export_succeeds(context: Context) -> None: assert context.result.exit_code == 0, ( f"Expected exit code 0, got {context.result.exit_code}.\n" f"Output: {context.result.output}" ) @then("the md export CLI result code should be nonzero") def step_cli_md_exit_with_error(context: Context) -> None: assert context.result.exit_code != 0, ( f"Expected non-zero exit code, got {context.result.exit_code}.\n" f"Output: {context.result.output}" ) @then('the md export CLI output should include "{text}"') def step_cli_md_output_contains(context: Context, text: str) -> None: assert text in context.result.output, ( f"Expected {text!r} in CLI output.\nGot: {context.result.output}" ) @then("the md export CLI output should be parseable as JSON") def step_cli_md_output_valid_json(context: Context) -> None: try: json.loads(context.result.output) except json.JSONDecodeError as exc: raise AssertionError( f"CLI output is not valid JSON: {exc}\nOutput: {context.result.output}" ) from exc @then("the exported markdown file should exist") def step_exported_md_file_exists(context: Context) -> None: assert os.path.exists(context.export_md_path), ( f"Markdown export file not found: {context.export_md_path}" ) @then('the exported markdown file should contain "{text}"') def step_exported_md_file_contains(context: Context, text: str) -> None: content = Path(context.export_md_path).read_text(encoding="utf-8") assert text in content, ( f"Expected {text!r} in exported markdown file.\nGot: {content[:500]}" ) # --------------------------------------------------------------------------- # TUI command router: /session export and /session import # --------------------------------------------------------------------------- def _make_tui_router_with_export_mock( session_id: str, export_data: dict[str, object], session_obj: Session, ) -> TuiCommandRouter: """Build a TuiCommandRouter with a mocked container/service for export. The mock container is injected via the ``container_factory`` constructor parameter so the mock survives ``multiprocessing.fork()`` boundaries used by the parallel test runner. """ mock_service = MagicMock() mock_service.export_session.return_value = export_data mock_service.get.return_value = session_obj mock_container = MagicMock() mock_container.session_service.return_value = mock_service registry = FakePersonaRegistry() state = FakePersonaState() router = TuiCommandRouter( persona_registry=registry, persona_state=state, container_factory=lambda: mock_container, ) return router @given("a TUI command router with mocked session service for export") def step_tui_router_for_export(context: Context) -> None: session_id = str(ULID()) context.tui_session_id = session_id msgs = [ _make_message(MessageRole.USER, "TUI user message", 0), _make_message(MessageRole.ASSISTANT, "TUI assistant reply", 1), ] session = _make_session(session_id=session_id, messages=msgs) export_data = session.as_export_dict() context.tui_router = _make_tui_router_with_export_mock( session_id, export_data, session ) @given("a TUI command router with mocked session service for import") def step_tui_router_for_import(context: Context) -> None: session_id = str(ULID()) context.tui_session_id = session_id imported_session = _make_session(session_id=str(ULID())) mock_service = MagicMock() mock_service.import_session.return_value = imported_session mock_container = MagicMock() mock_container.session_service.return_value = mock_service registry = FakePersonaRegistry() state = FakePersonaState() router = TuiCommandRouter( persona_registry=registry, persona_state=state, container_factory=lambda: mock_container, ) context.tui_router = router context.tui_imported_session = imported_session @given("there is a valid JSON export file for TUI import") def step_valid_json_export_for_tui(context: Context) -> None: msgs = [_make_message(MessageRole.USER, "Import test", 0)] session = _make_session(messages=msgs) export_data = session.as_export_dict() fd, path = tempfile.mkstemp(suffix=".json") with os.fdopen(fd, "w") as fh: json.dump(export_data, fh, default=str) context.tui_import_path = path @given("there is an invalid JSON file for TUI import") def step_invalid_json_for_tui(context: Context) -> None: fd, path = tempfile.mkstemp(suffix=".json") with os.fdopen(fd, "w") as fh: fh.write("{ not valid json }") context.tui_invalid_json_path = path @when('I call TUI handle with "{command}" for the current session') def step_tui_handle_command(context: Context, command: str) -> None: context.tui_result = context.tui_router.handle( command, session_id=context.tui_session_id ) @when('I call TUI handle with "session import " for the import file') def step_tui_handle_import_valid(context: Context) -> None: command = f"session import {context.tui_import_path}" context.tui_result = context.tui_router.handle( command, session_id=context.tui_session_id ) @when('I call TUI handle with "session import " for the import file') def step_tui_handle_import_invalid(context: Context) -> None: command = f"session import {context.tui_invalid_json_path}" context.tui_result = context.tui_router.handle( command, session_id=context.tui_session_id ) @then('the TUI handle result should contain "{text}"') def step_tui_result_contains(context: Context, text: str) -> None: assert text in context.tui_result, ( f"Expected {text!r} in TUI result.\nGot: {context.tui_result!r}" ) @then('the TUI handle result should be "{text}"') def step_tui_result_equals(context: Context, text: str) -> None: assert context.tui_result == text, ( f"Expected TUI result {text!r}.\nGot: {context.tui_result!r}" ) @then('the tui exported file "{path}" should exist') def step_tui_file_exists(context: Context, path: str) -> None: import contextlib assert os.path.exists(path), f"Expected file to exist: {path}" # Cleanup with contextlib.suppress(OSError): os.unlink(path) # --------------------------------------------------------------------------- # Domain model: as_export_plain_text() # --------------------------------------------------------------------------- @given("a session with two messages for plain text export") def step_session_two_messages_plain(context: Context) -> None: msgs = [ _make_message(MessageRole.USER, "Hello from user", 0), _make_message(MessageRole.ASSISTANT, "Hello from assistant", 1), ] context.txt_session = _make_session(messages=msgs) @given("a session with no messages for plain text export") def step_session_no_messages_plain(context: Context) -> None: context.txt_session = _make_session() @when("I call as_export_plain_text on the session") def step_call_as_export_plain_text(context: Context) -> None: context.txt_output = context.txt_session.as_export_plain_text() @then('the plain text output should start with "{prefix}"') def step_txt_starts_with(context: Context, prefix: str) -> None: assert context.txt_output.startswith(prefix), ( f"Expected plain text to start with {prefix!r}, got: {context.txt_output[:80]!r}" ) @then('the plain text output should contain "{text}"') def step_txt_contains(context: Context, text: str) -> None: assert text in context.txt_output, ( f"Expected {text!r} in plain text output.\nGot:\n{context.txt_output[:500]}" )