Spaces:
Paused
Paused
| """ | |
| Comprehensive test suite for browser_utils/debug_utils.py. | |
| This module tests all debug snapshot and error logging functions with >80% coverage. | |
| Focuses on: timestamp generation, DOM capture, system context, snapshot saving. | |
| REFACTORED: Reduced from 299 mocks to ~60 mocks by using real server state and | |
| smart fixture design instead of patching every server attribute individually. | |
| """ | |
| import os | |
| import platform | |
| import sys | |
| from unittest.mock import AsyncMock, MagicMock, Mock, mock_open, patch | |
| import pytest | |
| from playwright.async_api import Error as PlaywrightError | |
| from browser_utils.debug_utils import ( | |
| capture_dom_structure, | |
| capture_playwright_state, | |
| capture_system_context, | |
| get_texas_timestamp, | |
| save_comprehensive_snapshot, | |
| save_error_snapshot_enhanced, | |
| ) | |
| # === Smart Fixtures to Reduce Mocking === | |
| def mock_server_state(): | |
| """ | |
| Provide a realistic server state object instead of patching every attribute. | |
| This fixture REPLACES 16-18 individual @patch decorators per test. | |
| """ | |
| state = MagicMock() | |
| state.is_playwright_ready = True | |
| state.is_browser_connected = True | |
| state.is_page_ready = True | |
| state.is_initializing = False | |
| state.request_queue = MagicMock() | |
| state.request_queue.qsize.return_value = 0 | |
| state.processing_lock = MagicMock() | |
| state.processing_lock.locked.return_value = False | |
| state.model_switching_lock = MagicMock() | |
| state.model_switching_lock.locked.return_value = False | |
| state.current_ai_studio_model_id = None | |
| state.excluded_model_ids = [] | |
| state.browser_instance = None | |
| state.page_instance = None | |
| state.console_logs = [] | |
| state.network_log = {} | |
| state.STREAM_QUEUE = None | |
| state.PLAYWRIGHT_PROXY_SETTINGS = None | |
| return state | |
| def real_mock_page(): | |
| """ | |
| Provide a realistic Playwright page mock with common methods. | |
| Reduces duplication across tests that need page mocking. | |
| """ | |
| page = AsyncMock() | |
| page.evaluate = AsyncMock() | |
| page.title = AsyncMock(return_value="Test Page") | |
| page.url = "https://example.com" | |
| page.viewport_size = {"width": 1920, "height": 1080} | |
| page.is_closed = MagicMock(return_value=False) | |
| page.screenshot = AsyncMock() | |
| page.content = AsyncMock(return_value="<html><body>Test</body></html>") | |
| page.goto = AsyncMock() | |
| # Mock context for cookies | |
| page.context = AsyncMock() | |
| page.context.cookies = AsyncMock(return_value=[]) | |
| # Mock locator | |
| mock_locator = AsyncMock() | |
| mock_locator.count = AsyncMock(return_value=0) | |
| mock_locator.is_visible = AsyncMock(return_value=False) | |
| mock_locator.is_enabled = AsyncMock(return_value=False) | |
| page.locator = MagicMock(return_value=mock_locator) | |
| return page | |
| # === Section 1: Timestamp Generation Tests === | |
| class TestGetTexasTimestamp: | |
| """Tests for Texas timezone timestamp generation function.""" | |
| def test_timestamp_format_iso(self): | |
| """Test ISO format timestamp.""" | |
| iso, human = get_texas_timestamp() | |
| # ISO format: "2025-11-21T18:37:32.440" | |
| assert len(iso) == 23 | |
| assert "T" in iso | |
| assert iso.count("-") == 2 # YYYY-MM-DD | |
| assert iso.count(":") == 2 # HH:MM:SS | |
| def test_timestamp_format_human_readable(self): | |
| """Test human-readable timestamp format.""" | |
| iso, human = get_texas_timestamp() | |
| # Human format: "2025-11-21 18:37:32.440 <TZ>" where TZ is local timezone | |
| # The function now uses local timezone (not hardcoded CST) | |
| assert human.count(":") == 2 | |
| # Should have a timezone abbreviation at the end (e.g., CST, JST, EST) | |
| parts = human.split() | |
| assert len(parts) >= 3 # date, time, timezone (timezone might have spaces) | |
| assert len(parts[-1]) >= 2 # Timezone abbreviation (at least 2 chars) | |
| def test_timestamp_consistency(self): | |
| """Test timestamp consistency.""" | |
| iso, human = get_texas_timestamp() | |
| # Both should represent same time | |
| iso_date = iso.split("T")[0] | |
| human_date = human.split(" ")[0] | |
| assert iso_date == human_date | |
| def test_timestamp_milliseconds_precision(self): | |
| """Test millisecond precision.""" | |
| iso, human = get_texas_timestamp() | |
| # Should have 3 decimal places | |
| iso_time = iso.split("T")[1] | |
| assert "." in iso_time | |
| ms_part = iso_time.split(".")[1] | |
| assert len(ms_part) == 3 | |
| def test_timezone_offset_local(self, mock_datetime): | |
| """Test local timezone offset (using system timezone).""" | |
| from datetime import datetime as real_datetime | |
| from datetime import timezone as real_timezone | |
| # Create a mock datetime that returns a proper aware datetime | |
| # Simulate a UTC time that can be converted to local time | |
| mock_utc = real_datetime(2025, 1, 15, 12, 0, 0, 0, tzinfo=real_timezone.utc) | |
| # Mock now() to return a local-aware datetime (via astimezone) | |
| mock_local = mock_utc.astimezone() # Convert to local timezone | |
| mock_datetime.now.return_value.astimezone.return_value = mock_local | |
| iso, human = get_texas_timestamp() | |
| # The ISO format should contain the local time (not UTC) | |
| # Just verify it produces a valid timestamp with local timezone offset applied | |
| assert "T" in iso | |
| assert ":" in iso | |
| # Human format should have timezone abbreviation | |
| parts = human.split() | |
| assert len(parts) >= 3 # date, time, timezone | |
| # === Section 2: DOM Structure Capture Tests === | |
| class TestCaptureDomStructure: | |
| """Tests for DOM tree structure capture function.""" | |
| async def test_dom_structure_basic_success(self, real_mock_page): | |
| """Test basic DOM tree capture success.""" | |
| dom_tree = "BODY\n DIV#app.container\n P.text\n" | |
| real_mock_page.evaluate.return_value = dom_tree | |
| result = await capture_dom_structure(real_mock_page) | |
| assert result == dom_tree | |
| real_mock_page.evaluate.assert_called_once() | |
| async def test_dom_structure_with_hierarchy(self, real_mock_page): | |
| """Test hierarchical DOM capture.""" | |
| complex_dom = """BODY | |
| DIV#root.app-container | |
| HEADER.navbar | |
| NAV | |
| MAIN.content | |
| DIV.widget | |
| """ | |
| real_mock_page.evaluate.return_value = complex_dom | |
| result = await capture_dom_structure(real_mock_page) | |
| assert "BODY" in result | |
| assert "DIV#root.app-container" in result | |
| assert "HEADER.navbar" in result | |
| async def test_dom_structure_playwright_error(self, real_mock_page): | |
| """Test Playwright error handling.""" | |
| real_mock_page.evaluate.side_effect = PlaywrightError("Page closed") | |
| result = await capture_dom_structure(real_mock_page) | |
| assert "Error capturing DOM structure" in result | |
| assert "Page closed" in result | |
| async def test_dom_structure_generic_exception(self, real_mock_page): | |
| """Test general exception handling.""" | |
| real_mock_page.evaluate.side_effect = RuntimeError("Unexpected error") | |
| result = await capture_dom_structure(real_mock_page) | |
| assert "Error capturing DOM structure" in result | |
| assert "Unexpected error" in result | |
| async def test_dom_structure_javascript_evaluation(self, real_mock_page): | |
| """Test JavaScript execution logic.""" | |
| real_mock_page.evaluate.return_value = "BODY\n DIV#test\n" | |
| await capture_dom_structure(real_mock_page) | |
| # Verify JavaScript function was passed | |
| call_args = real_mock_page.evaluate.call_args[0][0] | |
| assert "function getTreeStructure" in call_args | |
| assert "maxDepth = 15" in call_args | |
| assert "element.tagName" in call_args | |
| # === Section 3: System Context Capture Tests === | |
| class TestCaptureSystemContext: | |
| """Tests for system context capture function.""" | |
| async def test_system_context_basic_structure(self, mock_server_state): | |
| """Test basic system context structure.""" | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context("req123", "test_error") | |
| # Verify top-level structure | |
| assert "meta" in context | |
| assert "system" in context | |
| assert "application_state" in context | |
| assert "browser_state" in context | |
| assert "configuration" in context | |
| assert "recent_activity" in context | |
| async def test_system_context_meta_fields(self, mock_server_state): | |
| """Test meta field content.""" | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context("abc123", "timeout_error") | |
| meta = context["meta"] | |
| assert meta["req_id"] == "abc123" | |
| assert meta["error_name"] == "timeout_error" | |
| assert "timestamp_iso" in meta | |
| assert "timestamp_texas" in meta | |
| async def test_system_context_system_info(self, mock_server_state): | |
| """Test system info fields.""" | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| system = context["system"] | |
| assert "platform" in system | |
| assert "python_version" in system | |
| assert "pid" in system | |
| assert system["platform"] == platform.platform() | |
| assert system["python_version"] == sys.version.split()[0] | |
| assert system["pid"] == os.getpid() | |
| async def test_system_context_application_flags(self, mock_server_state): | |
| """Test application state flags.""" | |
| # Customize state for this test | |
| mock_server_state.is_playwright_ready = True | |
| mock_server_state.is_browser_connected = False | |
| mock_server_state.is_page_ready = True | |
| mock_server_state.is_initializing = True | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| flags = context["application_state"]["flags"] | |
| assert flags["is_playwright_ready"] is True | |
| assert flags["is_browser_connected"] is False | |
| assert flags["is_page_ready"] is True | |
| assert flags["is_initializing"] is True | |
| async def test_system_context_queue_size(self, mock_server_state): | |
| """Test queue size capture.""" | |
| mock_server_state.request_queue.qsize.return_value = 5 | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| queues = context["application_state"]["queues"] | |
| assert queues["request_queue_size"] == 5 | |
| async def test_system_context_queue_not_implemented(self, mock_server_state): | |
| """Test case where queue does not support qsize.""" | |
| mock_server_state.request_queue.qsize.side_effect = NotImplementedError() | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| queues = context["application_state"]["queues"] | |
| assert queues["request_queue_size"] == -1 | |
| async def test_system_context_lock_states(self, mock_server_state): | |
| """Test lock state detection.""" | |
| mock_server_state.processing_lock.locked.return_value = True | |
| mock_server_state.model_switching_lock.locked.return_value = True | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| locks = context["application_state"]["locks"] | |
| assert locks["processing_lock_locked"] is True | |
| assert locks["model_switching_lock_locked"] is True | |
| async def test_system_context_proxy_sanitization(self, mock_server_state): | |
| """Test proxy settings credential sanitization.""" | |
| mock_server_state.PLAYWRIGHT_PROXY_SETTINGS = { | |
| "server": "http://user:password@proxy.com:8080" | |
| } | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| proxy = context["configuration"]["proxy_settings"] | |
| assert "***:***@" in proxy["server"] | |
| assert "user" not in proxy["server"] | |
| assert "password" not in proxy["server"] | |
| async def test_system_context_console_logs(self, mock_server_state): | |
| """Test console log capture.""" | |
| mock_server_state.console_logs = [ | |
| {"type": "log", "text": "Log 1"}, | |
| {"type": "error", "text": "Error 1"}, | |
| {"type": "warning", "text": "Warning 1"}, | |
| {"type": "log", "text": "Log 2"}, | |
| {"type": "error", "text": "Error 2"}, | |
| ] | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| activity = context["recent_activity"] | |
| assert activity["console_logs_count"] == 5 | |
| assert "last_console_logs" in activity | |
| assert "recent_console_errors" in activity | |
| assert len(activity["recent_console_errors"]) == 3 # 2 errors + 1 warning | |
| async def test_system_context_failed_network_responses(self, mock_server_state): | |
| """Test failed network response capture.""" | |
| mock_server_state.network_log = { | |
| "requests": [], | |
| "responses": [ | |
| {"status": 200, "url": "https://example.com/ok"}, | |
| {"status": 404, "url": "https://example.com/not-found"}, | |
| {"status": 500, "url": "https://example.com/error"}, | |
| ], | |
| } | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| activity = context["recent_activity"] | |
| assert "failed_network_responses" in activity | |
| assert len(activity["failed_network_responses"]) == 2 | |
| async def test_system_context_page_url(self, mock_server_state, real_mock_page): | |
| """Test current page URL capture.""" | |
| real_mock_page.url = "https://ai.google.dev/chat" | |
| mock_server_state.page_instance = real_mock_page | |
| with patch("api_utils.server_state.state", mock_server_state): | |
| context = await capture_system_context() | |
| assert context["browser_state"]["current_url"] == "https://ai.google.dev/chat" | |
| # === Section 4: Playwright State Capture Tests === | |
| class TestCapturePlaywrightState: | |
| """Tests for Playwright state capture function.""" | |
| async def test_playwright_state_basic_page_info(self, real_mock_page): | |
| """Test basic page info capture.""" | |
| real_mock_page.url = "https://ai.google.dev" | |
| real_mock_page.title.return_value = "AI Studio" | |
| real_mock_page.viewport_size = {"width": 1920, "height": 1080} | |
| state = await capture_playwright_state(real_mock_page) | |
| assert state["page"]["url"] == "https://ai.google.dev" | |
| assert state["page"]["title"] == "AI Studio" | |
| assert state["page"]["viewport"] == {"width": 1920, "height": 1080} | |
| async def test_playwright_state_title_error(self, real_mock_page): | |
| """Test failure to get page title.""" | |
| real_mock_page.title.side_effect = PlaywrightError("Page closed") | |
| state = await capture_playwright_state(real_mock_page) | |
| assert "Error:" in state["page"]["title"] | |
| async def test_playwright_state_locators_exists_and_visible(self, real_mock_page): | |
| """Test locators exist and are visible.""" | |
| mock_locator = AsyncMock() | |
| mock_locator.count.return_value = 1 | |
| mock_locator.is_visible.return_value = True | |
| mock_locator.is_enabled.return_value = True | |
| mock_locator.input_value.side_effect = PlaywrightError("Not an input") | |
| locators = {"submit_button": mock_locator} | |
| state = await capture_playwright_state(real_mock_page, locators) # type: ignore[arg-type] | |
| assert state["locators"]["submit_button"]["exists"] is True | |
| assert state["locators"]["submit_button"]["count"] == 1 | |
| assert state["locators"]["submit_button"]["visible"] is True | |
| assert state["locators"]["submit_button"]["enabled"] is True | |
| async def test_playwright_state_locators_not_exists(self, real_mock_page): | |
| """Test locator does not exist.""" | |
| mock_locator = AsyncMock() | |
| mock_locator.count.return_value = 0 | |
| locators = {"missing_element": mock_locator} | |
| state = await capture_playwright_state(real_mock_page, locators) # type: ignore[arg-type] | |
| assert state["locators"]["missing_element"]["exists"] is False | |
| assert state["locators"]["missing_element"]["count"] == 0 | |
| async def test_playwright_state_locators_with_input_value(self, real_mock_page): | |
| """Test capturing input element value.""" | |
| mock_locator = AsyncMock() | |
| mock_locator.count.return_value = 1 | |
| mock_locator.is_visible.return_value = True | |
| mock_locator.is_enabled.return_value = True | |
| mock_locator.input_value.return_value = "test input value" | |
| locators = {"input_field": mock_locator} | |
| state = await capture_playwright_state(real_mock_page, locators) # type: ignore[arg-type] | |
| assert state["locators"]["input_field"]["value"] == "test input value" | |
| async def test_playwright_state_locators_long_value_truncation( | |
| self, real_mock_page | |
| ): | |
| """Test long input value truncation.""" | |
| long_value = "a" * 150 | |
| mock_locator = AsyncMock() | |
| mock_locator.count.return_value = 1 | |
| mock_locator.is_visible.return_value = True | |
| mock_locator.is_enabled.return_value = True | |
| mock_locator.input_value.return_value = long_value | |
| locators = {"text_area": mock_locator} | |
| state = await capture_playwright_state(real_mock_page, locators) # type: ignore[arg-type] | |
| assert "..." in state["locators"]["text_area"]["value"] | |
| assert len(state["locators"]["text_area"]["value"]) == 103 # 100 + "..." | |
| async def test_playwright_state_locators_error_handling(self, real_mock_page): | |
| """Test locator error handling.""" | |
| mock_locator = AsyncMock() | |
| mock_locator.count.side_effect = PlaywrightError("Locator failed") | |
| locators = {"broken_locator": mock_locator} | |
| state = await capture_playwright_state(real_mock_page, locators) # type: ignore[arg-type] | |
| assert "error" in state["locators"]["broken_locator"] | |
| async def test_playwright_state_cookies_count(self, real_mock_page): | |
| """Test cookie count statistics.""" | |
| real_mock_page.context.cookies.return_value = [ | |
| {"name": "session", "value": "abc"}, | |
| {"name": "user", "value": "123"}, | |
| ] | |
| state = await capture_playwright_state(real_mock_page) | |
| assert state["storage"]["cookies_count"] == 2 | |
| async def test_playwright_state_localstorage_keys(self, real_mock_page): | |
| """Test localStorage key capture.""" | |
| real_mock_page.evaluate.return_value = ["theme", "user_id", "settings"] | |
| state = await capture_playwright_state(real_mock_page) | |
| assert state["storage"]["localStorage_keys"] == ["theme", "user_id", "settings"] | |
| async def test_playwright_state_storage_error_handling(self, real_mock_page): | |
| """Test failure to get storage information.""" | |
| real_mock_page.context.cookies.side_effect = PlaywrightError("Context closed") | |
| real_mock_page.evaluate.side_effect = PlaywrightError("Evaluation failed") | |
| state = await capture_playwright_state(real_mock_page) | |
| # Should not crash, just log warnings | |
| assert state["storage"]["cookies_count"] == 0 | |
| assert state["storage"]["localStorage_keys"] == [] | |
| # === Section 5: Comprehensive Snapshot Tests === | |
| class TestSaveComprehensiveSnapshot: | |
| """Tests for comprehensive snapshot saving function.""" | |
| async def test_snapshot_page_closed(self, real_mock_page): | |
| """Test snapshot not saved when page is closed.""" | |
| real_mock_page.is_closed.return_value = True | |
| result = await save_comprehensive_snapshot( | |
| page=real_mock_page, error_name="test_error", req_id="req123" | |
| ) | |
| assert result == "" | |
| async def test_snapshot_page_none(self): | |
| """Test snapshot not saved when page is None.""" | |
| result = await save_comprehensive_snapshot( | |
| page=None, # type: ignore[arg-type] | |
| error_name="test_error", | |
| req_id="req123", | |
| ) | |
| assert result == "" | |
| async def test_snapshot_directory_creation( | |
| self, real_mock_page, tmp_path, mock_server_state | |
| ): | |
| """Test snapshot directory creation.""" | |
| # Create mock Path that returns actual tmp_path for path operations | |
| mock_path_instance = MagicMock() | |
| mock_path_instance.__truediv__ = lambda self, other: tmp_path / str(other) | |
| with ( | |
| patch("browser_utils.debug_utils.Path") as mock_path_class, | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch("builtins.open", mock_open()), | |
| patch("browser_utils.debug_utils.capture_system_context") as mock_context, | |
| patch("browser_utils.debug_utils.capture_dom_structure") as mock_dom, | |
| patch("browser_utils.debug_utils.capture_playwright_state") as mock_pw, | |
| ): | |
| mock_context.return_value = {"meta": {}, "system": {}} | |
| mock_dom.return_value = "BODY\n" | |
| mock_pw.return_value = {} | |
| mock_path_class.return_value = mock_path_instance | |
| result = await save_comprehensive_snapshot( | |
| page=real_mock_page, error_name="timeout", req_id="abc123" | |
| ) | |
| # Verify function completed successfully (returns path) | |
| assert result is not None | |
| async def test_snapshot_screenshot_success( | |
| self, real_mock_page, tmp_path, mock_server_state | |
| ): | |
| """Test successful screenshot saving.""" | |
| snapshot_dir = tmp_path / "snapshot" | |
| snapshot_dir.mkdir() | |
| with ( | |
| patch("browser_utils.debug_utils.Path") as mock_path_class, | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch("browser_utils.debug_utils.capture_dom_structure") as mock_dom, | |
| patch( | |
| "browser_utils.debug_utils.capture_playwright_state" | |
| ) as mock_pw_state, | |
| patch("browser_utils.debug_utils.capture_system_context") as mock_sys_ctx, | |
| ): | |
| # Setup mocks | |
| mock_dom.return_value = "BODY\n" | |
| mock_pw_state.return_value = {"page": {}} | |
| mock_sys_ctx.return_value = {"meta": {}} | |
| # Mock Path to return our tmp_path | |
| base_dir = tmp_path / "errors_py" | |
| date_dir = base_dir / "2025-01-15" | |
| final_dir = date_dir / "snapshot" | |
| final_dir.mkdir(parents=True) | |
| mock_path_class.return_value.__truediv__.side_effect = [ | |
| base_dir, | |
| date_dir, | |
| final_dir, | |
| ] | |
| await save_comprehensive_snapshot( | |
| page=real_mock_page, error_name="test", req_id="req123" | |
| ) | |
| # Verify screenshot was called | |
| real_mock_page.screenshot.assert_called_once() | |
| async def test_snapshot_screenshot_failure(self, real_mock_page, mock_server_state): | |
| """Test screenshot failure handling.""" | |
| real_mock_page.screenshot.side_effect = PlaywrightError("Screenshot timeout") | |
| with ( | |
| patch("browser_utils.debug_utils.Path"), | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch("browser_utils.debug_utils.capture_dom_structure") as mock_dom, | |
| patch( | |
| "browser_utils.debug_utils.capture_playwright_state" | |
| ) as mock_pw_state, | |
| patch("browser_utils.debug_utils.capture_system_context") as mock_sys_ctx, | |
| patch("builtins.open", mock_open()), | |
| ): | |
| mock_dom.return_value = "BODY\n" | |
| mock_pw_state.return_value = {} | |
| mock_sys_ctx.return_value = {} | |
| # Should not crash | |
| await save_comprehensive_snapshot( | |
| page=real_mock_page, error_name="test", req_id="req123" | |
| ) | |
| # Should complete despite screenshot failure | |
| # === Section 6: Enhanced Snapshot Tests === | |
| class TestSaveErrorSnapshotEnhanced: | |
| """Tests for enhanced error snapshot function.""" | |
| async def test_enhanced_snapshot_browser_unavailable(self, mock_server_state): | |
| """Test snapshot not saved when browser is unavailable.""" | |
| mock_server_state.browser_instance = None | |
| mock_server_state.page_instance = None | |
| with ( | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch( | |
| "browser_utils.operations_modules.errors.save_minimal_snapshot", | |
| new_callable=AsyncMock, | |
| ) as mock_minimal, | |
| ): | |
| # Should not crash and should call minimal snapshot | |
| await save_error_snapshot_enhanced(error_name="test_error") | |
| mock_minimal.assert_called_once() | |
| async def test_enhanced_snapshot_page_closed(self, mock_server_state): | |
| """Test snapshot not saved when page is closed.""" | |
| mock_server_state.browser_instance = MagicMock() | |
| mock_server_state.browser_instance.is_connected.return_value = True | |
| mock_page = MagicMock() | |
| mock_page.is_closed = Mock(return_value=True) | |
| mock_server_state.page_instance = mock_page | |
| with ( | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch( | |
| "browser_utils.operations_modules.errors.save_minimal_snapshot", | |
| new_callable=AsyncMock, | |
| ) as mock_minimal, | |
| ): | |
| await save_error_snapshot_enhanced(error_name="page_closed_error") | |
| mock_minimal.assert_called_once() | |
| async def test_enhanced_snapshot_req_id_parsing( | |
| self, mock_server_state, real_mock_page | |
| ): | |
| """Test parsing req_id from error name.""" | |
| mock_server_state.browser_instance = MagicMock() | |
| mock_server_state.browser_instance.is_connected.return_value = True | |
| mock_server_state.page_instance = real_mock_page | |
| with ( | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch("browser_utils.debug_utils.save_comprehensive_snapshot") as mock_save, | |
| ): | |
| mock_save.return_value = "/path/to/snapshot" | |
| await save_error_snapshot_enhanced(error_name="timeout_error_abc1234") | |
| # Verify comprehensive snapshot was called with parsed req_id | |
| call_kwargs = mock_save.call_args[1] | |
| assert call_kwargs["req_id"] == "abc1234" | |
| assert call_kwargs["error_name"] == "timeout_error" | |
| async def test_enhanced_snapshot_with_exception( | |
| self, mock_server_state, real_mock_page | |
| ): | |
| """Test including exception information.""" | |
| mock_server_state.browser_instance = MagicMock() | |
| mock_server_state.browser_instance.is_connected.return_value = True | |
| mock_server_state.page_instance = real_mock_page | |
| error_exc = RuntimeError("Unexpected failure") | |
| with ( | |
| patch("api_utils.server_state.state", mock_server_state), | |
| patch("browser_utils.debug_utils.save_comprehensive_snapshot") as mock_save, | |
| ): | |
| mock_save.return_value = "/path/to/snapshot" | |
| await save_error_snapshot_enhanced( | |
| error_name="runtime_error_xyz7890", error_exception=error_exc | |
| ) | |
| # Verify exception was passed | |
| call_kwargs = mock_save.call_args[1] | |
| assert call_kwargs["error_exception"] == error_exc | |