From 67ebfe4bd0e053a82d2e4c7abcbc1c5602625894 Mon Sep 17 00:00:00 2001 From: recursiveAF Date: Tue, 27 Jan 2026 09:56:59 -0800 Subject: [PATCH 1/3] Fix cache miss in replay mode for subsequent tests ONCE mode now replays existing entries AND records new ones on cache miss, instead of failing when encountering instructions not in the cassette. This fixes the bug where the first test passes but subsequent tests fail with "Cache miss in replay mode" even though noot should be recording. Co-Authored-By: Claude Opus 4.5 --- src/noot/cache.py | 14 +-- tests/test_cache_modes.py | 249 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 254 insertions(+), 9 deletions(-) create mode 100644 tests/test_cache_modes.py diff --git a/src/noot/cache.py b/src/noot/cache.py index b50cfb7..614de00 100644 --- a/src/noot/cache.py +++ b/src/noot/cache.py @@ -8,7 +8,6 @@ from noot.project import ProjectRootNotFoundError, find_project_root - class CassettePathError(Exception): """Raised when cassette path cannot be determined.""" @@ -51,7 +50,7 @@ def get_cli_cassettes_dir() -> Path: class RecordMode(Enum): """Recording mode for cassettes.""" - ONCE = "once" # Record if cassette missing, replay if exists (default) + ONCE = "once" # Replay existing, record new entries on cache miss (default) NONE = "none" # Replay only, fail if request not found (use in CI) ALL = "all" # Always re-record, overwriting existing cassette @@ -92,7 +91,7 @@ def from_env(cls, cassette_path: Path | None = None) -> "Cache": .git location. Values: - - "once" (default): Record if cassette missing, replay if exists + - "once" (default): Replay existing entries, record new ones on miss - "none": Replay only, fail if request not found (use in CI) - "all": Always re-record, overwriting existing cassette """ @@ -118,18 +117,15 @@ def from_env(cls, cassette_path: Path | None = None) -> "Cache": cache._should_record = True cache._entries = [] elif mode == RecordMode.NONE: - # Replay only + # Replay only, fail on cache miss cache._should_record = False if cassette_exists: cache._load() else: # ONCE (default) + # Replay existing entries, record new ones on cache miss + cache._should_record = True if cassette_exists: - # Cassette exists: replay mode - cache._should_record = False cache._load() - else: - # Cassette missing: record mode - cache._should_record = True return cache diff --git a/tests/test_cache_modes.py b/tests/test_cache_modes.py new file mode 100644 index 0000000..acbc15f --- /dev/null +++ b/tests/test_cache_modes.py @@ -0,0 +1,249 @@ +"""Tests for Cache recording modes.""" + +import json +import os +from pathlib import Path + +import pytest + +from noot.cache import Cache, CacheMissError, RecordMode + + +class TestOnceModeDefault: + """ONCE mode: Replay existing entries, record new ones on cache miss.""" + + def test_records_when_cassette_missing(self, tmp_path: Path): + """First run with no cassette should record.""" + cassette = tmp_path / "test.json" + + cache = Cache.from_env(cassette) + + assert cache._should_record is True + assert cache.fail_on_miss is False + assert not cassette.exists() + + def test_records_and_saves_entry(self, tmp_path: Path): + """Recording should save entries to cassette file.""" + cassette = tmp_path / "test.json" + cache = Cache.from_env(cassette) + + cache.put("hello", "screen1", "response1", "expect", "ctx.contains('hello')") + + assert cassette.exists() + data = json.loads(cassette.read_text()) + assert len(data["entries"]) == 1 + assert data["entries"][0]["instruction"] == "hello" + + def test_replays_existing_entries(self, tmp_path: Path): + """Existing entries should be replayed from cache.""" + cassette = tmp_path / "test.json" + + # Session 1: Record an entry + cache1 = Cache.from_env(cassette) + cache1.put("hello", "screen1", "response1", "expect", "ctx.contains('hello')") + + # Session 2: Should replay the existing entry + cache2 = Cache.from_env(cassette) + result = cache2.get("hello", "screen1", "expect") + + assert result == "ctx.contains('hello')" + + def test_records_new_entries_on_cache_miss(self, tmp_path: Path): + """Cache miss in ONCE mode should record, not fail.""" + cassette = tmp_path / "test.json" + + # Session 1: Record entry A + cache1 = Cache.from_env(cassette) + cache1.put("test_a", "screen", "response_a", "expect", "ctx.contains('a')") + + # Session 2: Replay A, then record new entry B + cache2 = Cache.from_env(cassette) + + # Existing entry replays + result_a = cache2.get("test_a", "screen", "expect") + assert result_a == "ctx.contains('a')" + + # New entry returns None (cache miss) but doesn't fail + result_b = cache2.get("test_b", "screen", "expect") + assert result_b is None + assert cache2.fail_on_miss is False + + # Can record the new entry + cache2.put("test_b", "screen", "response_b", "expect", "ctx.contains('b')") + + # Verify cassette now has both entries + data = json.loads(cassette.read_text()) + instructions = [e["instruction"] for e in data["entries"]] + assert instructions == ["test_a", "test_b"] + + def test_multiple_tests_same_session(self, tmp_path: Path): + """Multiple tests in same session should all record to same cassette.""" + cassette = tmp_path / "test.json" + + # Simulate test_first + cache1 = Cache.from_env(cassette) + cache1.put("hello", "screen1", "response1", "expect", "ctx.contains('hello')") + + # Simulate test_second (same session, cassette now exists) + cache2 = Cache.from_env(cassette) + assert cache2._should_record is True # Should still allow recording + cache2.put("world", "screen2", "response2", "expect", "ctx.contains('world')") + + # Verify both entries recorded + data = json.loads(cassette.read_text()) + instructions = [e["instruction"] for e in data["entries"]] + assert instructions == ["hello", "world"] + + def test_loads_existing_entries_before_recording(self, tmp_path: Path): + """New session should load existing entries, not start fresh.""" + cassette = tmp_path / "test.json" + + # Session 1: Record entries A and B + cache1 = Cache.from_env(cassette) + cache1.put("test_a", "screen", "response_a", "expect", "code_a") + cache1.put("test_b", "screen", "response_b", "expect", "code_b") + + # Session 2: Should have loaded both entries + cache2 = Cache.from_env(cassette) + assert len(cache2._entries) == 2 + + # Add entry C + cache2.put("test_c", "screen", "response_c", "expect", "code_c") + + # Verify all three in cassette + data = json.loads(cassette.read_text()) + assert len(data["entries"]) == 3 + + +class TestNoneMode: + """NONE mode: Replay only, fail on cache miss (for CI).""" + + def test_replays_existing_entries(self, tmp_path: Path, monkeypatch): + """Existing entries should replay successfully.""" + cassette = tmp_path / "test.json" + + # Create cassette with ONCE mode first + cache1 = Cache.from_env(cassette) + cache1.put("hello", "screen", "response", "expect", "ctx.contains('hello')") + + # Switch to NONE mode + monkeypatch.setenv("RECORD_MODE", "none") + cache2 = Cache.from_env(cassette) + + result = cache2.get("hello", "screen", "expect") + assert result == "ctx.contains('hello')" + + def test_fails_on_cache_miss(self, tmp_path: Path, monkeypatch): + """Cache miss in NONE mode should indicate failure.""" + cassette = tmp_path / "test.json" + + # Create cassette with entry A + cache1 = Cache.from_env(cassette) + cache1.put("test_a", "screen", "response", "expect", "code_a") + + # Switch to NONE mode and try to access missing entry B + monkeypatch.setenv("RECORD_MODE", "none") + cache2 = Cache.from_env(cassette) + + assert cache2._should_record is False + assert cache2.fail_on_miss is True + + # Cache miss returns None, but fail_on_miss indicates caller should error + result = cache2.get("test_b", "screen", "expect") + assert result is None + + def test_does_not_record_new_entries(self, tmp_path: Path, monkeypatch): + """NONE mode should not record new entries.""" + cassette = tmp_path / "test.json" + + # Create cassette with entry A + cache1 = Cache.from_env(cassette) + cache1.put("test_a", "screen", "response", "expect", "code_a") + + # Switch to NONE mode and try to record + monkeypatch.setenv("RECORD_MODE", "none") + cache2 = Cache.from_env(cassette) + cache2.put("test_b", "screen", "response", "expect", "code_b") + + # Entry B should NOT be saved (put() is a no-op when not recording) + data = json.loads(cassette.read_text()) + instructions = [e["instruction"] for e in data["entries"]] + assert instructions == ["test_a"] + + def test_fails_when_cassette_missing(self, tmp_path: Path, monkeypatch): + """NONE mode with missing cassette should fail on any lookup.""" + cassette = tmp_path / "nonexistent.json" + monkeypatch.setenv("RECORD_MODE", "none") + + cache = Cache.from_env(cassette) + + assert cache._should_record is False + assert cache.fail_on_miss is True + assert len(cache._entries) == 0 + + +class TestAllMode: + """ALL mode: Always re-record, overwriting existing cassette.""" + + def test_clears_existing_entries(self, tmp_path: Path, monkeypatch): + """ALL mode should clear existing entries and start fresh.""" + cassette = tmp_path / "test.json" + + # Create cassette with entries A and B + cache1 = Cache.from_env(cassette) + cache1.put("test_a", "screen", "response", "expect", "code_a") + cache1.put("test_b", "screen", "response", "expect", "code_b") + + # Switch to ALL mode - should clear entries + monkeypatch.setenv("RECORD_MODE", "all") + cache2 = Cache.from_env(cassette) + + assert cache2._should_record is True + assert len(cache2._entries) == 0 # Cleared! + + def test_records_fresh_entries(self, tmp_path: Path, monkeypatch): + """ALL mode should record fresh entries.""" + cassette = tmp_path / "test.json" + + # Create cassette with old entries + cache1 = Cache.from_env(cassette) + cache1.put("old_entry", "screen", "response", "expect", "old_code") + + # Switch to ALL mode and record new entries + monkeypatch.setenv("RECORD_MODE", "all") + cache2 = Cache.from_env(cassette) + cache2.put("new_entry", "screen", "response", "expect", "new_code") + + # Cassette should only have new entry + data = json.loads(cassette.read_text()) + instructions = [e["instruction"] for e in data["entries"]] + assert instructions == ["new_entry"] + + def test_never_fails_on_miss(self, tmp_path: Path, monkeypatch): + """ALL mode should never fail on cache miss.""" + cassette = tmp_path / "test.json" + monkeypatch.setenv("RECORD_MODE", "all") + + cache = Cache.from_env(cassette) + + assert cache._should_record is True + assert cache.fail_on_miss is False + + +class TestCacheMissError: + """Tests for CacheMissError exception.""" + + def test_error_message_format(self): + """Error message should include instruction and method.""" + error = CacheMissError("hello world", "expect") + + assert "Cache miss in replay mode" in str(error) + assert "expect()" in str(error) + assert "'hello world'" in str(error) + + def test_error_attributes(self): + """Error should store instruction and method as attributes.""" + error = CacheMissError("test instruction", "complete") + + assert error.instruction == "test instruction" + assert error.method == "complete" From be2e8467b7a561046dc2a4e15febc21f0038452d Mon Sep 17 00:00:00 2001 From: recursiveAF Date: Tue, 27 Jan 2026 09:58:54 -0800 Subject: [PATCH 2/3] Fix linting errors - Add blank line after imports in cache.py - Remove unused imports in test_cache_modes.py Co-Authored-By: Claude Opus 4.5 --- src/noot/cache.py | 1 + tests/test_cache_modes.py | 5 +---- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/src/noot/cache.py b/src/noot/cache.py index 614de00..94820eb 100644 --- a/src/noot/cache.py +++ b/src/noot/cache.py @@ -8,6 +8,7 @@ from noot.project import ProjectRootNotFoundError, find_project_root + class CassettePathError(Exception): """Raised when cassette path cannot be determined.""" diff --git a/tests/test_cache_modes.py b/tests/test_cache_modes.py index acbc15f..74d5b36 100644 --- a/tests/test_cache_modes.py +++ b/tests/test_cache_modes.py @@ -1,12 +1,9 @@ """Tests for Cache recording modes.""" import json -import os from pathlib import Path -import pytest - -from noot.cache import Cache, CacheMissError, RecordMode +from noot.cache import Cache, CacheMissError class TestOnceModeDefault: From 1a01e88447adc1b5e2d9a709132549fa1ae4ee16 Mon Sep 17 00:00:00 2001 From: recursiveAF Date: Tue, 27 Jan 2026 12:13:29 -0800 Subject: [PATCH 3/3] Respect env PATH when locating mitmdump binary Previously, MitmproxyManager used shutil.which("mitmdump") which only searched the pytest process's PATH, ignoring the env dict passed to Flow.spawn(). This caused issues when users set a custom PATH in env to locate mitmdump in non-standard locations. Now MitmproxyConfig accepts an env parameter, and MitmproxyManager extracts the PATH from it (if present) to pass to shutil.which(). Co-Authored-By: Claude Opus 4.5 --- src/noot/flow.py | 1 + src/noot/mitmproxy_manager.py | 6 ++-- tests/test_mitmproxy_manager.py | 60 +++++++++++++++++++++++++++++++++ 3 files changed, 65 insertions(+), 2 deletions(-) create mode 100644 tests/test_mitmproxy_manager.py diff --git a/src/noot/flow.py b/src/noot/flow.py index 00af247..d9eb4ea 100644 --- a/src/noot/flow.py +++ b/src/noot/flow.py @@ -71,6 +71,7 @@ def __init__( listen_port=mitmproxy_port, http_cassettes_dir=http_cassettes_dir, record_mode=self._cache.mode, + env=env, # Pass env for PATH lookup when finding mitmdump ) # Only create mitmproxy manager if http_cassettes_dir was determined if config.http_cassettes_dir: diff --git a/src/noot/mitmproxy_manager.py b/src/noot/mitmproxy_manager.py index 7bdb964..17300a1 100644 --- a/src/noot/mitmproxy_manager.py +++ b/src/noot/mitmproxy_manager.py @@ -58,6 +58,7 @@ class MitmproxyConfig: addon_path: Path | None = None command: str | None = None # Kept for potential future use record_mode: RecordMode = RecordMode.ONCE # Controls spy mode behavior + env: dict[str, str] | None = None # Environment for PATH lookup def __post_init__(self): # Use default HTTP cassettes directory if not specified @@ -96,8 +97,9 @@ def start(self) -> None: # Pass record mode to addon env["MITM_RECORD_MODE"] = self._config.record_mode.value - # Find mitmdump binary - mitmdump_path = shutil.which("mitmdump") + # Find mitmdump binary, respecting the env PATH if provided + search_path = self._config.env.get("PATH") if self._config.env else None + mitmdump_path = shutil.which("mitmdump", path=search_path) if not mitmdump_path: raise RuntimeError( "mitmdump not found in PATH. Install with: pip install mitmproxy" diff --git a/tests/test_mitmproxy_manager.py b/tests/test_mitmproxy_manager.py new file mode 100644 index 0000000..c2aceaa --- /dev/null +++ b/tests/test_mitmproxy_manager.py @@ -0,0 +1,60 @@ +"""Unit tests for MitmproxyManager.""" + +from unittest.mock import patch + +import pytest + +from noot.mitmproxy_manager import MitmproxyConfig, MitmproxyManager + + +class TestMitmproxyManagerPathLookup: + """Test that MitmproxyManager respects env PATH when finding mitmdump.""" + + def test_uses_env_path_when_provided(self): + """mitmdump lookup should use PATH from env dict when provided.""" + custom_path = "/custom/bin:/another/path" + config = MitmproxyConfig( + env={"PATH": custom_path, "OTHER_VAR": "value"}, + ) + manager = MitmproxyManager(config) + + with patch("noot.mitmproxy_manager.shutil.which") as mock_which: + mock_which.return_value = None # Simulate not found + + with pytest.raises(RuntimeError, match="mitmdump not found"): + manager.start() + + # Verify shutil.which was called with the custom PATH + mock_which.assert_called_once_with("mitmdump", path=custom_path) + + def test_uses_system_path_when_env_not_provided(self): + """mitmdump lookup should use system PATH when env is None.""" + config = MitmproxyConfig(env=None) + manager = MitmproxyManager(config) + + with patch("noot.mitmproxy_manager.shutil.which") as mock_which: + mock_which.return_value = None # Simulate not found + + with pytest.raises(RuntimeError, match="mitmdump not found"): + manager.start() + + # Verify shutil.which was called with path=None (uses system PATH) + mock_which.assert_called_once_with("mitmdump", path=None) + + def test_uses_system_path_when_env_has_no_path(self): + """mitmdump lookup should use system PATH when env dict has no PATH key.""" + config = MitmproxyConfig(env={"OTHER_VAR": "value"}) + manager = MitmproxyManager(config) + + with patch("noot.mitmproxy_manager.shutil.which") as mock_which: + mock_which.return_value = None # Simulate not found + + with pytest.raises(RuntimeError, match="mitmdump not found"): + manager.start() + + # Verify shutil.which was called with path=None (uses system PATH) + mock_which.assert_called_once_with("mitmdump", path=None) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])