diff --git a/tests/cassettes/portability/echo_cli.json b/.cassettes/cli/portability/echo_cli.json similarity index 100% rename from tests/cassettes/portability/echo_cli.json rename to .cassettes/cli/portability/echo_cli.json diff --git a/tests/cassettes/test_create_cli_tool_project.json b/.cassettes/cli/test_create_cli_tool_project.json similarity index 100% rename from tests/cassettes/test_create_cli_tool_project.json rename to .cassettes/cli/test_create_cli_tool_project.json diff --git a/tests/cassettes/test_create_library_project.json b/.cassettes/cli/test_create_library_project.json similarity index 100% rename from tests/cassettes/test_create_library_project.json rename to .cassettes/cli/test_create_library_project.json diff --git a/tests/cassettes/test_create_web_project.json b/.cassettes/cli/test_create_web_project.json similarity index 100% rename from tests/cassettes/test_create_web_project.json rename to .cassettes/cli/test_create_web_project.json diff --git a/README.md b/README.md index b56a86d..1dec99d 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Think of it as a Stagehand for interactive CLIs. ```python def test_create_web_project(): - with Flow.spawn('python setup_wizard.py', cassette=cassette) as f: + with Flow.spawn('python setup_wizard.py') as f: f.expect('Welcome to Project Setup Wizard') f.step("Enter project name 'mywebapp' and press enter") @@ -47,3 +47,7 @@ Example - CI mode (fail if cassette is missing): ```bash RECORD_MODE=none pytest tests/test_cli.py ``` + +Cassettes are stored in `/.cassettes/`: +- CLI cassettes (LLM responses): `.cassettes/cli/` +- HTTP cassettes (API recordings): `.cassettes/http/` diff --git a/src/noot/addons/spy_mode.py b/src/noot/addons/spy_mode.py index 7fa51aa..522b69d 100644 --- a/src/noot/addons/spy_mode.py +++ b/src/noot/addons/spy_mode.py @@ -1,12 +1,12 @@ """ -Mitmproxy addon implementing mode-aware API spy. +Mitmproxy addon implementing mode-aware HTTP cassette recording/replay. Behavior by mode (RECORD_MODE env var): - once (default): Record if no recordings exist, replay if they do - none: Replay only, fail if no match found (use in CI) - all: Always re-record, overwriting existing recordings -On startup → load existing recordings from most recent file +On startup → load existing HTTP cassettes from most recent file On request → check for match, replay if found On response → record new interaction (when recording) On shutdown → save recordings (when recording) @@ -102,46 +102,47 @@ class SpyModeAddon: - all: Always re-record """ - def __init__(self, recordings_dir: Path | None = None): - # Check environment variable if recordings_dir not provided - if not recordings_dir: - env_path = os.environ.get("MITM_RECORDINGS_DIR") + def __init__(self, http_cassettes_dir: Path | None = None): + # Check environment variable if http_cassettes_dir not provided + if not http_cassettes_dir: + env_path = os.environ.get("MITM_HTTP_CASSETTES_DIR") if env_path: - recordings_dir = Path(env_path) + http_cassettes_dir = Path(env_path) - self.recordings_dir = recordings_dir + self.http_cassettes_dir = http_cassettes_dir # Read record mode from environment (default to "once") self.record_mode = os.environ.get("MITM_RECORD_MODE", "once").lower() - # Existing recordings (loaded from most recent file) - self.existing_recordings: list[RecordedInteraction] = [] + # Existing HTTP cassettes (loaded from most recent file) + self.existing_cassettes: list[RecordedInteraction] = [] - # New recordings made during this session - self.new_recordings: list[RecordedInteraction] = [] + # New HTTP cassettes recorded during this session + self.new_cassettes: list[RecordedInteraction] = [] # Determine if we should record based on mode and existing recordings self._should_record = False self._load_most_recent() + # Set recording flag based on mode and existing cassettes if self.record_mode == "all": # Always record, clear existing self._should_record = True - self.existing_recordings = [] + self.existing_cassettes = [] elif self.record_mode == "none": # Replay only self._should_record = False else: # "once" (default) - # Record if no existing recordings, otherwise replay - self._should_record = len(self.existing_recordings) == 0 + # Record if no existing cassettes, otherwise replay + self._should_record = len(self.existing_cassettes) == 0 - def _find_most_recent_recording(self) -> Path | None: - """Find the most recent recording file in the recordings directory.""" - if not self.recordings_dir or not self.recordings_dir.exists(): + def _find_most_recent_cassette(self) -> Path | None: + """Find the most recent HTTP cassette file in the cassettes directory.""" + if not self.http_cassettes_dir or not self.http_cassettes_dir.exists(): return None # Find all JSON files in the directory - json_files = list(self.recordings_dir.glob("*.json")) + json_files = list(self.http_cassettes_dir.glob("*.json")) if not json_files: return None @@ -150,80 +151,83 @@ def _find_most_recent_recording(self) -> Path | None: return json_files[0] def _load_most_recent(self) -> None: - """Load existing recordings from the most recent file.""" - recent_file = self._find_most_recent_recording() + """Load existing HTTP cassettes from the most recent file.""" + recent_file = self._find_most_recent_cassette() if not recent_file: - print(f"[SpyMode] No existing recordings found in {self.recordings_dir}") + print( + f"[SpyMode] No existing HTTP cassettes found in " + f"{self.http_cassettes_dir}" + ) return try: data = json.loads(recent_file.read_text()) - self.existing_recordings = [ + self.existing_cassettes = [ RecordedInteraction.from_dict(item) for item in data.get("interactions", []) ] print( - f"[SpyMode] Loaded {len(self.existing_recordings)} " - f"recordings from {recent_file.name}" + f"[SpyMode] Loaded {len(self.existing_cassettes)} " + f"HTTP cassettes from {recent_file.name}" ) except Exception as e: - print(f"[SpyMode] Error loading recordings: {e}") - self.existing_recordings = [] + print(f"[SpyMode] Error loading HTTP cassettes: {e}") + self.existing_cassettes = [] - def _save_new_recordings(self) -> None: - """Save new recordings to a timestamped file.""" - if not self.new_recordings or not self.recordings_dir: + def _save_new_cassettes(self) -> None: + """Save new HTTP cassettes to a timestamped file.""" + if not self.new_cassettes or not self.http_cassettes_dir: return - self.recordings_dir.mkdir(parents=True, exist_ok=True) + self.http_cassettes_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - output_file = self.recordings_dir / f"{timestamp}.json" + output_file = self.http_cassettes_dir / f"{timestamp}.json" - # Combine existing and new recordings - all_recordings = self.existing_recordings + self.new_recordings + # Combine existing and new cassettes + all_cassettes = self.existing_cassettes + self.new_cassettes data = { "version": "1.0", - "interactions": [r.to_dict() for r in all_recordings], + "interactions": [r.to_dict() for r in all_cassettes], } output_file.write_text(json.dumps(data, indent=2)) print( - f"[SpyMode] Saved {len(self.new_recordings)} " - f"new recordings to {output_file.name}" + f"[SpyMode] Saved {len(self.new_cassettes)} " + f"new HTTP cassettes to {output_file.name}" ) - print(f"[SpyMode] Total recordings: {len(all_recordings)}") + print(f"[SpyMode] Total HTTP cassettes: {len(all_cassettes)}") def load(self, loader: Loader) -> None: """Called when addon is loaded.""" loader.add_option( - name="spy_recordings_dir", + name="spy_http_cassettes_dir", typespec=str, default="", - help="Directory for spy mode recordings", + help="Directory for HTTP cassettes", ) def request(self, flow: http.HTTPFlow) -> None: """ Intercept incoming request. - Check if we have a recorded response for this request. + Check if we have a cassette response for this request. If yes, replay it. If no, let it pass through to real API. """ - # Check existing recordings first, then new ones - all_recordings = self.existing_recordings + self.new_recordings + # Check existing cassettes first, then new ones + all_cassettes = self.existing_cassettes + self.new_cassettes - for recording in all_recordings: - if recording.matches(flow): + for cassette in all_cassettes: + if cassette.matches(flow): url = flow.request.pretty_url print(f"[SpyMode] REPLAY: {flow.request.method} {url}") - # Create response from recording + # Create response from cassette flow.response = http.Response.make( - status_code=recording.response["status_code"], - content=recording.response["content"].encode("utf-8"), - headers=recording.response["headers"], + status_code=cassette.response["status_code"], + content=cassette.response["content"].encode("utf-8"), + headers=cassette.response["headers"], ) # Mark that this was replayed (prevent recording in response()) @@ -249,14 +253,14 @@ def response(self, flow: http.HTTPFlow) -> None: # Only record when in recording mode if self._should_record: - recording = RecordedInteraction.from_flow(flow) + cassette = RecordedInteraction.from_flow(flow) print(f"[SpyMode] RECORD: {flow.request.method} {flow.request.pretty_url}") - self.new_recordings.append(recording) + self.new_cassettes.append(cassette) def done(self) -> None: """Called when mitmproxy is shutting down.""" - if self._should_record and self.new_recordings: - self._save_new_recordings() + if self._should_record and self.new_cassettes: + self._save_new_cassettes() # Entry point for mitmproxy diff --git a/src/noot/cache.py b/src/noot/cache.py index 0abe3ca..b50cfb7 100644 --- a/src/noot/cache.py +++ b/src/noot/cache.py @@ -1,4 +1,4 @@ -"""LLM response caching for deterministic test replay.""" +"""CLI cassette caching for deterministic test replay.""" import json import os @@ -6,6 +6,47 @@ from enum import Enum from pathlib import Path +from noot.project import ProjectRootNotFoundError, find_project_root + + +class CassettePathError(Exception): + """Raised when cassette path cannot be determined.""" + + def __init__(self, message: str): + super().__init__(message) + + +def get_cli_cassettes_dir() -> Path: + """ + Get the default CLI cassettes directory. + + Resolution order: + 1. NOOT_CASSETTE_DIR environment variable + 2. /.cassettes/cli/ (based on .git location) + + Raises: + CassettePathError: If no .git directory is found and + NOOT_CASSETTE_DIR is not set. + """ + # Check for explicit env var first + env_dir = os.environ.get("NOOT_CASSETTE_DIR") + if env_dir: + return Path(env_dir) + + try: + root = find_project_root() + except ProjectRootNotFoundError as e: + raise CassettePathError( + "Cannot determine cassette directory: " + f"no .git found starting from {e.start_dir}.\n" + "Either:\n" + " 1. Run from within a git repository, or\n" + " 2. Set NOOT_CASSETTE_DIR environment variable, or\n" + " 3. Pass an explicit cassette path: " + "Flow.spawn(..., cassette='path/to/cassette.json')" + ) from e + return root / ".cassettes" / "cli" + class RecordMode(Enum): """Recording mode for cassettes.""" @@ -17,7 +58,7 @@ class RecordMode(Enum): @dataclass class CacheEntry: - """A single cached LLM response.""" + """A single cached LLM response in a CLI cassette.""" instruction: str screen: str @@ -29,9 +70,9 @@ class CacheEntry: @dataclass class Cache: """ - LLM response cache for deterministic replay. + CLI cassette cache for deterministic replay. - Caches are keyed by instruction only. For expect() calls, assertion code + CLI cassettes are keyed by instruction only. For expect() calls, assertion code is stored and replayed deterministically without screen comparison. """ @@ -45,6 +86,11 @@ def from_env(cls, cassette_path: Path | None = None) -> "Cache": """ Create cache based on RECORD_MODE environment variable. + Args: + cassette_path: Path to CLI cassette file. If not specified and + mode is record/replay, uses default directory based on + .git location. + Values: - "once" (default): Record if cassette missing, replay if exists - "none": Replay only, fail if request not found (use in CI) @@ -58,6 +104,10 @@ def from_env(cls, cassette_path: Path | None = None) -> "Cache": else: mode = RecordMode.ONCE + # Determine cassette path - always set a default path + if cassette_path is None: + cassette_path = get_cli_cassettes_dir() / "default.json" + cache = cls(mode=mode, path=cassette_path) # Determine behavior based on mode and cassette existence @@ -84,7 +134,7 @@ def from_env(cls, cassette_path: Path | None = None) -> "Cache": return cache def _load(self) -> None: - """Load cache entries from file.""" + """Load CLI cassette entries from file.""" if not self.path or not self.path.exists(): return data = json.loads(self.path.read_text()) @@ -100,7 +150,7 @@ def _load(self) -> None: ] def save(self) -> None: - """Save cache entries to file.""" + """Save CLI cassette entries to file.""" if not self.path: return self.path.parent.mkdir(parents=True, exist_ok=True) @@ -118,9 +168,7 @@ def save(self) -> None: data = {"entries": entries} self.path.write_text(json.dumps(data, indent=2)) - def get( - self, instruction: str, screen: str, method: str - ) -> str | None: + def get(self, instruction: str, screen: str, method: str) -> str | None: """ Look up a cached response. diff --git a/src/noot/flow.py b/src/noot/flow.py index 7d031fe..ef8358d 100644 --- a/src/noot/flow.py +++ b/src/noot/flow.py @@ -33,7 +33,7 @@ def __init__( pane_height: int = 40, stability_timeout: float = 5.0, cassette: str | Path | None = None, - api_recordings: str | Path | None = None, + http_cassettes: str | Path | None = None, mitmproxy_port: int = 8080, ): """ @@ -47,8 +47,8 @@ def __init__( stability_timeout: Default timeout for waiting for terminal stability cassette: Path to cassette file for caching LLM responses. Behavior controlled by RECORD_MODE env var (once/none/all). - api_recordings: Directory for API recordings (mitmproxy). - Defaults to recordings/api_replays/ + http_cassettes: Directory for HTTP cassettes (mitmproxy). + Defaults to /.cassettes/http/ mitmproxy_port: Port for mitmproxy to listen on (default 8080) """ self._command = command @@ -60,13 +60,14 @@ def __init__( self._steps: list[StepResult] = [] # Mitmproxy integration for API recording/replay - if api_recordings: - recordings_dir = Path(api_recordings) - config = MitmproxyConfig( - listen_port=mitmproxy_port, - recordings_dir=recordings_dir, - record_mode=self._cache.mode, - ) + http_cassettes_dir = Path(http_cassettes) if http_cassettes else None + config = MitmproxyConfig( + listen_port=mitmproxy_port, + http_cassettes_dir=http_cassettes_dir, + record_mode=self._cache.mode, + ) + # Only create mitmproxy manager if http_cassettes_dir was determined + if config.http_cassettes_dir: self._mitmproxy: MitmproxyManager | None = MitmproxyManager(config) else: self._mitmproxy = None @@ -80,7 +81,7 @@ def spawn( pane_height: int = 40, stability_timeout: float = 5.0, cassette: str | Path | None = None, - api_recordings: str | Path | None = None, + http_cassettes: str | Path | None = None, mitmproxy_port: int = 8080, ) -> Flow: """ @@ -94,8 +95,8 @@ def spawn( stability_timeout: Default timeout for stability cassette: Path to cassette file for caching LLM responses. Behavior controlled by RECORD_MODE env var (once/none/all). - api_recordings: Directory for API recordings (mitmproxy). - Defaults to recordings/api_replays/ + http_cassettes: Directory for HTTP cassettes (mitmproxy). + Defaults to /.cassettes/http/ mitmproxy_port: Port for mitmproxy to listen on (default 8080) Returns: @@ -108,7 +109,7 @@ def spawn( pane_height=pane_height, stability_timeout=stability_timeout, cassette=cassette, - api_recordings=api_recordings, + http_cassettes=http_cassettes, mitmproxy_port=mitmproxy_port, ) diff --git a/src/noot/mitmproxy_manager.py b/src/noot/mitmproxy_manager.py index b0f2df4..7bdb964 100644 --- a/src/noot/mitmproxy_manager.py +++ b/src/noot/mitmproxy_manager.py @@ -8,7 +8,8 @@ from dataclasses import dataclass from pathlib import Path -from noot.cache import RecordMode +from noot.cache import CassettePathError, RecordMode +from noot.project import ProjectRootNotFoundError, find_project_root def get_mitmproxy_ca_cert_path() -> Path: @@ -16,13 +17,36 @@ def get_mitmproxy_ca_cert_path() -> Path: return Path.home() / ".mitmproxy" / "mitmproxy-ca-cert.pem" -def get_recordings_dir() -> Path: +def get_http_cassettes_dir() -> Path: """ - Get the default API recordings directory. + Get the default HTTP cassettes directory. - Returns recordings/api_replays/ + Resolution order: + 1. NOOT_HTTP_CASSETTE_DIR environment variable + 2. /.cassettes/http/ (based on .git location) + + Raises: + CassettePathError: If no .git directory is found and + NOOT_HTTP_CASSETTE_DIR is not set. """ - return Path.cwd() / "recordings" / "api_replays" + # Check for explicit env var first + env_dir = os.environ.get("NOOT_HTTP_CASSETTE_DIR") + if env_dir: + return Path(env_dir) + + try: + root = find_project_root() + except ProjectRootNotFoundError as e: + raise CassettePathError( + "Cannot determine HTTP cassette directory: " + f"no .git found starting from {e.start_dir}.\n" + "Either:\n" + " 1. Run from within a git repository, or\n" + " 2. Set NOOT_HTTP_CASSETTE_DIR environment variable, or\n" + " 3. Pass an explicit http_cassettes path: " + "Flow.spawn(..., http_cassettes='path/to/dir')" + ) from e + return root / ".cassettes" / "http" @dataclass @@ -30,15 +54,20 @@ class MitmproxyConfig: """Mitmproxy configuration.""" listen_port: int = 8080 - recordings_dir: Path | None = None + http_cassettes_dir: Path | None = None addon_path: Path | None = None command: str | None = None # Kept for potential future use record_mode: RecordMode = RecordMode.ONCE # Controls spy mode behavior def __post_init__(self): - # Use default recordings directory if not specified - if self.recordings_dir is None: - self.recordings_dir = get_recordings_dir() + # Use default HTTP cassettes directory if not specified + if self.http_cassettes_dir is None: + try: + self.http_cassettes_dir = get_http_cassettes_dir() + except CassettePathError: + # Cannot determine default directory - leave as None + # Caller should check and skip mitmproxy if not needed + pass class MitmproxyManager: @@ -59,10 +88,10 @@ def start(self) -> None: if self._started: raise RuntimeError("Mitmproxy already started") - # Set recordings directory via environment variable + # Set HTTP cassettes directory via environment variable env = os.environ.copy() - if self._config.recordings_dir: - env["MITM_RECORDINGS_DIR"] = str(self._config.recordings_dir) + if self._config.http_cassettes_dir: + env["MITM_HTTP_CASSETTES_DIR"] = str(self._config.http_cassettes_dir) # Pass record mode to addon env["MITM_RECORD_MODE"] = self._config.record_mode.value @@ -76,9 +105,11 @@ def start(self) -> None: cmd = [ mitmdump_path, - "--listen-port", str(self._config.listen_port), + "--listen-port", + str(self._config.listen_port), "--ssl-insecure", # Don't verify upstream HTTPS certs - "-s", str(self._config.addon_path), + "-s", + str(self._config.addon_path), ] # Start mitmproxy diff --git a/src/noot/project.py b/src/noot/project.py new file mode 100644 index 0000000..344b1c5 --- /dev/null +++ b/src/noot/project.py @@ -0,0 +1,43 @@ +"""Project root discovery utilities.""" + +from pathlib import Path + + +class ProjectRootNotFoundError(Exception): + """Raised when no project root can be found.""" + + def __init__(self, start_dir: Path): + self.start_dir = start_dir + super().__init__( + f"Could not find project root (no .git directory) starting from {start_dir}" + ) + + +def find_project_root(start_dir: Path | None = None) -> Path: + """ + Find the project root by walking up from start_dir to find .git directory. + + Args: + start_dir: Directory to start searching from. Defaults to cwd. + + Returns: + Path to the project root directory. + + Raises: + ProjectRootNotFoundError: If no .git directory is found. + """ + if start_dir is None: + start_dir = Path.cwd() + + current = start_dir.resolve() + + while current != current.parent: + if (current / ".git").exists(): + return current + current = current.parent + + # Check root directory as well + if (current / ".git").exists(): + return current + + raise ProjectRootNotFoundError(start_dir) diff --git a/tests/test_api_replay.py b/tests/test_api_replay.py new file mode 100644 index 0000000..4da3505 --- /dev/null +++ b/tests/test_api_replay.py @@ -0,0 +1,195 @@ +""" +Tests for HTTP cassette replay functionality via mitmproxy. + +These tests verify that: +1. HTTP calls made by CLIs are recorded to cassettes +2. HTTP calls are replayed from cassettes + +Run with: + pytest tests/test_api_replay.py -v + +Requires ANTHROPIC_API_KEY for the initial recording. +""" + +import json +import os +import time +from pathlib import Path + +import pytest + +from noot import Flow + +# Test directories +TESTS_DIR = Path(__file__).parent +PROJECT_ROOT = TESTS_DIR.parent +HTTP_CASSETTES_DIR = PROJECT_ROOT / ".cassettes" / "http" +CLI_CASSETTES_DIR = PROJECT_ROOT / ".cassettes" / "cli" + + +@pytest.fixture +def storyteller_cli(): + """Path to the storyteller CLI.""" + cli_path = PROJECT_ROOT / "storyteller_cli.py" + if not cli_path.exists(): + pytest.skip("storyteller_cli.py not found in project root") + return cli_path + + +@pytest.fixture +def clean_http_cassettes(): + """Clean up HTTP cassettes directory before test.""" + # Clean up any existing cassettes for a fresh test + if HTTP_CASSETTES_DIR.exists(): + for f in HTTP_CASSETTES_DIR.glob("*.json"): + if not f.name.startswith("."): + f.unlink() + yield + # Optionally clean up after test too + + +class TestHttpCassetteStoryteller: + """Test HTTP cassette replay with storyteller CLI that makes Anthropic API calls.""" + + def test_storyteller_record_and_replay(self, storyteller_cli, clean_http_cassettes): + """ + Test that storyteller CLI HTTP calls are recorded and replayed. + + 1. RECORD: Makes real API call, saves to .cassettes/http/ + 2. REPLAY: Uses cached response, verifies no network needed + """ + # Check for API key + if "ANTHROPIC_API_KEY" not in os.environ: + pytest.skip("ANTHROPIC_API_KEY required for this test") + + story_beginning = "Once upon a time in a test" + cassette_path = CLI_CASSETTES_DIR / "test_storyteller_api.json" + + # --- PHASE 1: RECORD --- + os.environ["RECORD_MODE"] = "all" + + # Ensure directory exists + HTTP_CASSETTES_DIR.mkdir(parents=True, exist_ok=True) + print(f"HTTP_CASSETTES_DIR (absolute): {HTTP_CASSETTES_DIR.resolve()}") + + with Flow.spawn( + f"echo '{story_beginning}' | uv run python {storyteller_cli}", + cassette=cassette_path, + http_cassettes=HTTP_CASSETTES_DIR, + mitmproxy_port=8080, + stability_timeout=30.0, + ) as flow: + # Wait for the CLI to complete + max_wait = 60 + start = time.time() + while time.time() - start < max_wait: + output = flow.screen() + if "Story completed successfully" in output or "Error:" in output: + break + time.sleep(1) + + record_output = flow.screen() + print(f"\n=== RECORD PHASE OUTPUT ===\n{record_output}\n===") + + # Give mitmproxy a moment to save cassettes + time.sleep(1) + + # Verify cassette was created + cassette_files = [ + f for f in HTTP_CASSETTES_DIR.glob("*.json") if not f.name.startswith(".") + ] + print(f"HTTP_CASSETTES_DIR: {HTTP_CASSETTES_DIR}") + print(f"HTTP cassette files found: {cassette_files}") + assert len(cassette_files) > 0, ( + f"Expected HTTP cassette file to be created in {HTTP_CASSETTES_DIR}" + ) + + # Get the cassette file + latest_cassette = max(cassette_files, key=lambda f: f.stat().st_mtime) + data = json.loads(latest_cassette.read_text()) + interactions = data.get("interactions", []) + assert len(interactions) > 0, "Expected at least one HTTP interaction recorded" + + # Verify it's an Anthropic API call + first_req = interactions[0]["request"] + assert "api.anthropic.com" in first_req["url"], ( + f"Expected Anthropic API URL, got: {first_req['url']}" + ) + + # --- PHASE 2: REPLAY --- + os.environ["RECORD_MODE"] = "none" + + # Need to reimport to pick up new env var + import importlib + + import noot.cache + + importlib.reload(noot.cache) + + with Flow.spawn( + f"echo '{story_beginning}' | uv run python {storyteller_cli}", + cassette=cassette_path, + http_cassettes=HTTP_CASSETTES_DIR, + mitmproxy_port=8080, + stability_timeout=30.0, + ) as flow: + # Wait for the CLI to complete + max_wait = 30 + start = time.time() + while time.time() - start < max_wait: + output = flow.screen() + if "Story completed successfully" in output or "Error:" in output: + break + time.sleep(1) + + replay_output = flow.screen() + + # Verify replay worked - output should contain story completion + assert "Story completed successfully" in replay_output, ( + f"Expected story completion in replay. Output:\n{replay_output}" + ) + + # Clean up env var + del os.environ["RECORD_MODE"] + + +class TestHttpCassetteIntegration: + """Integration tests for HTTP cassette functionality.""" + + def test_cassette_file_format(self): + """Verify the structure of HTTP cassette files.""" + # Find any existing cassette file + if not HTTP_CASSETTES_DIR.exists(): + pytest.skip("No HTTP cassettes directory") + + cassette_files = [ + f for f in HTTP_CASSETTES_DIR.glob("*.json") if not f.name.startswith(".") + ] + + if not cassette_files: + pytest.skip("No HTTP cassette files found - run record test first") + + # Check the structure of the most recent cassette + latest = max(cassette_files, key=lambda f: f.stat().st_mtime) + data = json.loads(latest.read_text()) + + # Verify expected structure + assert "interactions" in data, "Cassette should have 'interactions' key" + assert isinstance(data["interactions"], list), "Interactions should be a list" + + if data["interactions"]: + interaction = data["interactions"][0] + assert "request" in interaction, "Interaction should have 'request'" + assert "response" in interaction, "Interaction should have 'response'" + + req = interaction["request"] + assert "method" in req, "Request should have 'method'" + assert "url" in req, "Request should have 'url'" + + resp = interaction["response"] + assert "status_code" in resp, "Response should have 'status_code'" + assert "content" in resp, "Response should have 'content'" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_cassette_portability.py b/tests/test_cassette_portability.py index 3457ead..366556d 100644 --- a/tests/test_cassette_portability.py +++ b/tests/test_cassette_portability.py @@ -1,7 +1,7 @@ """ -Integration tests for cassette portability across environments. +Integration tests for CLI cassette portability across environments. -Tests that cassettes recorded in one environment can replay in another +Tests that CLI cassettes recorded in one environment can replay in another (different paths, usernames, temp dirs, etc.) This test harness helps evaluate different f.expect implementations @@ -20,8 +20,10 @@ from noot import Flow -EXAMPLES_DIR = Path(__file__).parent.parent / "examples" -CASSETTE_DIR = Path(__file__).parent / "cassettes" / "portability" +TESTS_DIR = Path(__file__).parent +PROJECT_ROOT = TESTS_DIR.parent +EXAMPLES_DIR = PROJECT_ROOT / "examples" +CLI_CASSETTES_DIR = PROJECT_ROOT / ".cassettes" / "cli" / "portability" class EnvironmentTransformer: @@ -60,7 +62,7 @@ def simulate_ci_environment(self) -> dict: def test_same_environment_replay(): """Baseline: cassette replays in same environment.""" - cassette = CASSETTE_DIR / "echo_cli.json" + cassette = CLI_CASSETTES_DIR / "echo_cli.json" with tempfile.TemporaryDirectory() as tmpdir: os.chdir(tmpdir) @@ -80,7 +82,7 @@ def test_path_transformed_replay(): With assertion code approach, this should succeed because assertions match by instruction only, not screen similarity. """ - original_cassette = CASSETTE_DIR / "echo_cli.json" + original_cassette = CLI_CASSETTES_DIR / "echo_cli.json" if not original_cassette.exists(): pytest.fail("Cassette not recorded yet. Run with NOOT_CACHE=record first.") @@ -116,15 +118,13 @@ def test_path_transformed_replay(): ) def test_path_variations(old_path: str, new_path: str): """Parameterized test: various path transformations should succeed.""" - original_cassette = CASSETTE_DIR / "echo_cli.json" + original_cassette = CLI_CASSETTES_DIR / "echo_cli.json" if not original_cassette.exists(): pytest.fail("Cassette not recorded yet. Run with NOOT_CACHE=record first.") transformer = EnvironmentTransformer(original_cassette) - varied_data = transformer.transform_paths( - "/Users/seneca/Coding/goose", new_path - ) + varied_data = transformer.transform_paths("/Users/seneca/Coding/goose", new_path) with tempfile.NamedTemporaryFile( mode="w", suffix=".json", delete=False @@ -157,7 +157,7 @@ def test_random_environment_variations(): With assertion code approach, all variations should pass. """ num_variations = 5 - original_cassette = CASSETTE_DIR / "echo_cli.json" + original_cassette = CLI_CASSETTES_DIR / "echo_cli.json" if not original_cassette.exists(): pytest.fail("Cassette not recorded yet. Run with NOOT_CACHE=record first.") @@ -194,9 +194,10 @@ def test_random_environment_variations(): print(f"Portability test results: {results['pass']}/{num_variations} passed") # With assertion code approach, all variations should pass - assert ( - results["pass"] == num_variations - ), f"All variations should pass, but only {results['pass']}/{num_variations} passed" + assert results["pass"] == num_variations, ( + f"All variations should pass, but only " + f"{results['pass']}/{num_variations} passed" + ) if __name__ == "__main__": diff --git a/tests/test_sample_cli.py b/tests/test_sample_cli.py index bb6e16e..e0156ce 100644 --- a/tests/test_sample_cli.py +++ b/tests/test_sample_cli.py @@ -5,8 +5,10 @@ from noot import Flow -CASSETTE_DIR = Path(__file__).parent / 'cassettes' -EXAMPLES_DIR = Path(__file__).parent.parent / 'examples' +TESTS_DIR = Path(__file__).parent +PROJECT_ROOT = TESTS_DIR.parent +CLI_CASSETTES_DIR = PROJECT_ROOT / ".cassettes" / "cli" +EXAMPLES_DIR = PROJECT_ROOT / "examples" def test_create_web_project(): @@ -14,30 +16,30 @@ def test_create_web_project(): with tempfile.TemporaryDirectory() as tmpdir: os.chdir(tmpdir) - sample_cli = EXAMPLES_DIR / 'sample_cli.py' - cassette = CASSETTE_DIR / 'test_create_web_project.json' + sample_cli = EXAMPLES_DIR / "sample_cli.py" + cassette = CLI_CASSETTES_DIR / "test_create_web_project.json" - with Flow.spawn(f'python {sample_cli}', cassette=cassette) as f: - f.expect('Welcome to Project Setup Wizard') + with Flow.spawn(f"python {sample_cli}", cassette=cassette) as f: + f.expect("Welcome to Project Setup Wizard") f.step("Enter project name 'mywebapp' and press enter") - f.expect('Select project type') + f.expect("Select project type") # Web Application is the first option, so just press enter - f.step('Press enter to select Web Application') - f.expect('Enter author name') + f.step("Press enter to select Web Application") + f.expect("Enter author name") f.step("Enter author name 'Alice' and press enter") - f.expect('Project created successfully') + f.expect("Project created successfully") # Verify the config file was created correctly - config_path = Path(tmpdir) / 'project_config.json' - assert config_path.exists(), 'Config file should be created' + config_path = Path(tmpdir) / "project_config.json" + assert config_path.exists(), "Config file should be created" config = json.loads(config_path.read_text()) - assert config['name'] == 'mywebapp' - assert config['type'] == 'Web Application' - assert config['author'] == 'Alice' + assert config["name"] == "mywebapp" + assert config["type"] == "Web Application" + assert config["author"] == "Alice" def test_create_cli_tool_project(): @@ -45,29 +47,29 @@ def test_create_cli_tool_project(): with tempfile.TemporaryDirectory() as tmpdir: os.chdir(tmpdir) - sample_cli = EXAMPLES_DIR / 'sample_cli.py' - cassette = CASSETTE_DIR / 'test_create_cli_tool_project.json' + sample_cli = EXAMPLES_DIR / "sample_cli.py" + cassette = CLI_CASSETTES_DIR / "test_create_cli_tool_project.json" - with Flow.spawn(f'python {sample_cli}', cassette=cassette) as f: - f.expect('Welcome to Project Setup Wizard') + with Flow.spawn(f"python {sample_cli}", cassette=cassette) as f: + f.expect("Welcome to Project Setup Wizard") f.step("Enter project name 'mytool' and press enter") - f.expect('Select project type') + f.expect("Select project type") # CLI Tool is the second option, navigate down once - f.step('Press down arrow once and then press enter to select CLI Tool') - f.expect('Enter author name') + f.step("Press down arrow once and then press enter to select CLI Tool") + f.expect("Enter author name") f.step("Enter author name 'Bob' and press enter") - f.expect('Project created successfully') + f.expect("Project created successfully") - config_path = Path(tmpdir) / 'project_config.json' + config_path = Path(tmpdir) / "project_config.json" assert config_path.exists() config = json.loads(config_path.read_text()) - assert config['name'] == 'mytool' - assert config['type'] == 'CLI Tool' - assert config['author'] == 'Bob' + assert config["name"] == "mytool" + assert config["type"] == "CLI Tool" + assert config["author"] == "Bob" def test_create_library_project(): @@ -75,31 +77,32 @@ def test_create_library_project(): with tempfile.TemporaryDirectory() as tmpdir: os.chdir(tmpdir) - sample_cli = EXAMPLES_DIR / 'sample_cli.py' - cassette = CASSETTE_DIR / 'test_create_library_project.json' + sample_cli = EXAMPLES_DIR / "sample_cli.py" + cassette = CLI_CASSETTES_DIR / "test_create_library_project.json" - with Flow.spawn(f'python {sample_cli}', cassette=cassette) as f: - f.expect('Welcome to Project Setup Wizard') + with Flow.spawn(f"python {sample_cli}", cassette=cassette) as f: + f.expect("Welcome to Project Setup Wizard") f.step("Enter project name 'mylib' and press enter") - f.expect('Select project type') + f.expect("Select project type") # Library is the third option, navigate down twice - f.step('Press down arrow twice to select Library, then press enter') - f.expect('Enter author name') + f.step("Press down arrow twice to select Library, then press enter") + f.expect("Enter author name") f.step("Enter author name 'Charlie' and press enter") - f.expect('Project created successfully') + f.expect("Project created successfully") - config_path = Path(tmpdir) / 'project_config.json' + config_path = Path(tmpdir) / "project_config.json" assert config_path.exists() config = json.loads(config_path.read_text()) - assert config['name'] == 'mylib' - assert config['type'] == 'Library' - assert config['author'] == 'Charlie' + assert config["name"] == "mylib" + assert config["type"] == "Library" + assert config["author"] == "Charlie" -if __name__ == '__main__': +if __name__ == "__main__": import pytest - pytest.main([__file__, '-v']) + + pytest.main([__file__, "-v"])