diff --git a/src/noot/cache.py b/src/noot/cache.py index b50cfb7..94820eb 100644 --- a/src/noot/cache.py +++ b/src/noot/cache.py @@ -51,7 +51,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 +92,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 +118,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/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_cache_modes.py b/tests/test_cache_modes.py new file mode 100644 index 0000000..74d5b36 --- /dev/null +++ b/tests/test_cache_modes.py @@ -0,0 +1,246 @@ +"""Tests for Cache recording modes.""" + +import json +from pathlib import Path + +from noot.cache import Cache, CacheMissError + + +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" 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"])