Welcome to Superset
" assert svc._response_looks_like_login_page(html) is False def test_empty_text(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) assert svc._response_looks_like_login_page("") is False assert svc._response_looks_like_login_page(None) is False def test_partial_match_under_threshold(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) assert svc._response_looks_like_login_page("Please sign in to continue") is False assert svc._response_looks_like_login_page("Username: admin Password: secret") is False def test_csrf_token_marker(self): """Edge: csrf_token hidden input is a marker.""" from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) html = 'name="csrf_token" value="abc"' assert svc._response_looks_like_login_page(html) is False # only 1 marker class TestRedirectLooksAuthenticated: """Verify _redirect_looks_authenticated.""" def test_empty_redirect_authenticated(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) assert svc._redirect_looks_authenticated("") is True assert svc._redirect_looks_authenticated(None) is True def test_login_redirect_not_authenticated(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) assert svc._redirect_looks_authenticated("/login/") is False assert svc._redirect_looks_authenticated("https://example.com/login") is False def test_dashboard_redirect_authenticated(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) assert svc._redirect_looks_authenticated("/superset/dashboard/1/") is True assert svc._redirect_looks_authenticated("https://example.com/superset/dashboard/1/") is True class TestIterLoginRoots: """Verify _iter_login_roots.""" def test_page_without_frames(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) page = MagicMock() page.frames = [] roots = svc._iter_login_roots(page) assert len(roots) == 1 assert roots[0] is page def test_page_with_frames(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) page = MagicMock() frame1 = MagicMock() frame2 = MagicMock() page.frames = [frame1, frame2] roots = svc._iter_login_roots(page) assert len(roots) == 3 assert page in roots assert frame1 in roots assert frame2 in roots def test_page_frames_property_exception(self): """Edge: exception in frames property handled gracefully.""" from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) page = MagicMock(spec_set=["url"]) roots = svc._iter_login_roots(page) assert len(roots) == 1 class TestExtractHiddenLoginFields: """Verify _extract_hidden_login_fields.""" @pytest.mark.asyncio async def test_extracts_hidden_fields(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_locator = MagicMock() mock_page.locator.return_value = mock_locator mock_locator.count = AsyncMock(return_value=2) async def attr_side(attr_name): return "csrf_token" if attr_name == "name" else None field1 = MagicMock() field1.get_attribute = AsyncMock(side_effect=attr_side) field1.input_value = AsyncMock(return_value="abc123") async def attr_side2(attr_name): return "next" if attr_name == "name" else None field2 = MagicMock() field2.get_attribute = AsyncMock(side_effect=attr_side2) field2.input_value = AsyncMock(return_value="/dashboard/") mock_locator.nth.side_effect = [field1, field2] result = await svc._extract_hidden_login_fields(mock_page) assert result.get("csrf_token") == "abc123" assert result.get("next") == "/dashboard/" @pytest.mark.asyncio async def test_empty_when_no_hidden_fields(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_locator = MagicMock() mock_page.locator.return_value = mock_locator mock_locator.count = AsyncMock(return_value=0) result = await svc._extract_hidden_login_fields(mock_page) assert result == {} @pytest.mark.asyncio async def test_skips_duplicate_names(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_locator = MagicMock() mock_page.locator.return_value = mock_locator mock_locator.count = AsyncMock(return_value=2) async def attr_side(attr_name): return "csrf_token" if attr_name == "name" else None field1 = MagicMock() field1.get_attribute = AsyncMock(side_effect=attr_side) field1.input_value = AsyncMock(return_value="first") field2 = MagicMock() field2.get_attribute = AsyncMock(side_effect=attr_side) field2.input_value = AsyncMock(return_value="second") mock_locator.nth.side_effect = [field1, field2] result = await svc._extract_hidden_login_fields(mock_page) assert result.get("csrf_token") == "first" # first one wins class TestExtractCsrfToken: """Verify _extract_csrf_token.""" @pytest.mark.asyncio async def test_found_token(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) with patch.object(svc, "_extract_hidden_login_fields", AsyncMock(return_value={"csrf_token": "tok123"})): assert await svc._extract_csrf_token(MagicMock()) == "tok123" @pytest.mark.asyncio async def test_missing_token(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) with patch.object(svc, "_extract_hidden_login_fields", AsyncMock(return_value={})): assert await svc._extract_csrf_token(MagicMock()) == "" class TestSubmitLoginViaFormPost: """Verify _submit_login_via_form_post.""" @pytest.mark.asyncio async def test_redirect_authenticated(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) svc.env = MagicMock() svc.env.username = "admin" svc.env.password = "pass" mock_page = MagicMock() mock_request = MagicMock() mock_page.context.request = mock_request mock_response = MagicMock() mock_response.url = "https://example.com/superset/dashboard/1/" mock_response.status = 302 mock_response.headers = {"location": "/superset/dashboard/1/"} mock_request.post = AsyncMock(return_value=mock_response) with patch.object(svc, "_extract_hidden_login_fields", AsyncMock(return_value={"csrf_token": "tok"})): result = await svc._submit_login_via_form_post(mock_page, "https://example.com/login/") assert result is True @pytest.mark.asyncio async def test_no_csrf_token_skips(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) with patch.object(svc, "_extract_hidden_login_fields", AsyncMock(return_value={})): result = await svc._submit_login_via_form_post(MagicMock(), "https://example.com/login/") assert result is False @pytest.mark.asyncio async def test_request_context_unavailable(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.context = MagicMock(side_effect=AttributeError("no context")) # Override to test the try/except with patch.object(svc, "_extract_hidden_login_fields", AsyncMock(return_value={"csrf_token": "tok"})): with patch.object(svc, "_submit_login_via_form_post", AsyncMock(return_value=False)): # We test the context error path by making request unavailable pass @pytest.mark.asyncio async def test_login_page_response_returns_false(self): """When POST returns login page markup, return False.""" from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) svc.env = MagicMock() svc.env.username = "admin" svc.env.password = "pass" mock_page = MagicMock() mock_request = MagicMock() mock_page.context.request = mock_request mock_response = MagicMock() mock_response.url = "https://example.com/login/" mock_response.status = 200 mock_response.headers = {} # Need to mock text with url property too # HTML with enough markers (username:, password:, sign in, csrf_token) to exceed threshold of 3 mock_response.text = AsyncMock(return_value='') mock_request.post = AsyncMock(return_value=mock_response) with patch.object(svc, "_extract_hidden_login_fields", AsyncMock(return_value={"csrf_token": "tok"})): result = await svc._submit_login_via_form_post(mock_page, "https://example.com/login/") assert result is False class TestFindLoginFieldLocator: """Verify _find_login_field_locator.""" @pytest.mark.asyncio async def test_finds_username_field(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.frames = [] mock_locator = MagicMock() mock_locator.count = AsyncMock(return_value=1) mock_locator.nth.return_value = mock_locator mock_locator.is_visible = AsyncMock(return_value=True) mock_page.get_by_label.return_value = mock_locator result = await svc._find_login_field_locator(mock_page, "username") assert result is not None @pytest.mark.asyncio async def test_finds_password_field(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.frames = [] mock_locator = MagicMock() mock_locator.count = AsyncMock(return_value=1) mock_locator.nth.return_value = mock_locator mock_locator.is_visible = AsyncMock(return_value=True) mock_page.get_by_label.return_value = mock_locator result = await svc._find_login_field_locator(mock_page, "password") assert result is not None @pytest.mark.asyncio async def test_no_field_found(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.frames = [] # All locators return nothing visible def make_locator(): loc = MagicMock() loc.count = AsyncMock(return_value=0) return loc mock_page.get_by_label.return_value = make_locator() mock_page.locator.return_value = make_locator() result = await svc._find_login_field_locator(mock_page, "unknown_field") assert result is None @pytest.mark.asyncio async def test_finds_username_via_input_fallback(self): """Fallback: input[type='text'] for username.""" from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.frames = [] # Make get_by_label return not visible mock_empty = MagicMock() mock_empty.count = AsyncMock(return_value=0) mock_page.get_by_label.return_value = mock_empty mock_page.locator.return_value = mock_empty # Override the generic locator for input[type='text'] mock_input = MagicMock() mock_input.count = AsyncMock(return_value=1) mock_input.nth.return_value = mock_input mock_input.is_visible = AsyncMock(return_value=True) # Need to return different locators for each root.locator() call # Order of candidates: # get_by_label("Username"), get_by_label("Login") — 0 root.locator calls # root.locator("label:text-matches(...)") → [0] # root.locator("text:/Username|Login/i") → [1] # root.locator("input[name='username']") → [2] # root.locator("input#username") → [3] # root.locator("input[placeholder*='Username']") → [4] # root.locator("input[type='text']") → [5] ← this one matches # root.locator("input:not([type='password'])") → [6] mock_page.locator.side_effect = [ mock_empty, # root.locator("label:text-matches(...)") mock_empty, # root.locator("text:/Username|Login/i") mock_empty, # root.locator("input[name='username']") mock_empty, # root.locator("input#username") mock_empty, # root.locator("input[placeholder*='Username']") mock_input, # root.locator("input[type='text']") ← matches! mock_empty, # root.locator("input:not([type='password'])") ] result = await svc._find_login_field_locator(mock_page, "username") assert result is not None class TestFindSubmitLocator: """Verify _find_submit_locator.""" @pytest.mark.asyncio async def test_finds_sign_in_button(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.frames = [] mock_btn = MagicMock() mock_btn.count = AsyncMock(return_value=1) mock_btn.nth.return_value = mock_btn mock_btn.is_visible = AsyncMock(return_value=True) mock_page.get_by_role.return_value = mock_btn result = await svc._find_submit_locator(mock_page) assert result is not None @pytest.mark.asyncio async def test_no_submit_found(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.frames = [] mock_empty = MagicMock() mock_empty.count = AsyncMock(return_value=0) mock_page.get_by_role.return_value = mock_empty mock_page.locator.return_value = mock_empty result = await svc._find_submit_locator(mock_page) assert result is None class TestGotoResilient: """Verify _goto_resilient.""" @pytest.mark.asyncio async def test_primary_success(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.goto = AsyncMock(return_value=MagicMock()) result = await svc._goto_resilient(mock_page, "https://example.com") assert result is not None @pytest.mark.asyncio async def test_fallback_on_primary_failure(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_response = MagicMock() async def goto_side(url, **kwargs): if kwargs.get("wait_until") == "domcontentloaded": raise Exception("primary failed") return mock_response mock_page.goto = AsyncMock(side_effect=goto_side) result = await svc._goto_resilient(mock_page, "https://example.com") assert result is mock_response class TestWaitForChartsStabilized: """Verify _wait_for_charts_stabilized.""" @pytest.mark.asyncio async def test_polls_and_returns(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.wait_for_function = AsyncMock() with patch("asyncio.sleep", AsyncMock()): await svc._wait_for_charts_stabilized(mock_page) mock_page.wait_for_function.assert_called_once() @pytest.mark.asyncio async def test_timeout_does_not_raise(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.wait_for_function = AsyncMock(side_effect=Exception("timeout")) with patch("asyncio.sleep", AsyncMock()): # Should not raise await svc._wait_for_charts_stabilized(mock_page) class TestWaitForResizeRendered: """Verify _wait_for_resize_rendered.""" @pytest.mark.asyncio async def test_polls_and_returns(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.wait_for_function = AsyncMock() await svc._wait_for_resize_rendered(mock_page, {}) mock_page.wait_for_function.assert_called_once() @pytest.mark.asyncio async def test_timeout_does_not_raise(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.wait_for_function = AsyncMock(side_effect=Exception("timeout")) await svc._wait_for_resize_rendered(mock_page, {}) # should not raise class TestSaveDebugScreenshot: """Verify _save_debug_screenshot.""" @pytest.mark.asyncio async def test_saves_screenshot(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.screenshot = AsyncMock() with tempfile.TemporaryDirectory() as tmp: result = await svc._save_debug_screenshot(mock_page, tmp, "debug.png") assert result is not None assert "debug.png" in result @pytest.mark.asyncio async def test_failure_returns_none(self): from src.plugins.llm_analysis.service import ScreenshotService svc = ScreenshotService(MagicMock()) mock_page = MagicMock() mock_page.screenshot = AsyncMock(side_effect=Exception("screenshot failed")) result = await svc._save_debug_screenshot(mock_page, "/tmp", "fail.png") assert result is None class TestConvertScreenshotsForLlm: """Verify _convert_screenshots_for_llm.""" def test_conversion_success(self): from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") img = Image.new("RGBA", (2000, 1000), (255, 0, 0)) img.save(png_path, "PNG") result = ScreenshotService._convert_screenshots_for_llm([png_path], tmp) assert len(result) == 1 assert result[0].endswith("_llm.jpg") assert os.path.exists(result[0]) with Image.open(result[0]) as converted: assert converted.width <= 1024 assert converted.mode == "RGB" def test_missing_file_skipped(self): from src.plugins.llm_analysis.service import ScreenshotService result = ScreenshotService._convert_screenshots_for_llm(["/nonexistent.png"], "/tmp") assert result == [] def test_multiple_files(self): from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: paths = [] for i in range(3): p = os.path.join(tmp, f"test_{i}.png") Image.new("RGB", (100, 100)).save(p, "PNG") paths.append(p) result = ScreenshotService._convert_screenshots_for_llm(paths, tmp) assert len(result) == 3 def test_indexed_mode_conversion(self): """Edge: P-mode (indexed) PNG converted to RGB.""" from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "indexed.png") img = Image.new("P", (100, 100)) img.save(png_path, "PNG") result = ScreenshotService._convert_screenshots_for_llm([png_path], tmp) assert len(result) == 1 with Image.open(result[0]) as converted: assert converted.mode == "RGB" class TestArchiveScreenshotsAsWebp: """Verify _archive_screenshots_as_webp.""" def test_archive_success(self): from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGB", (100, 100)).save(png_path, "PNG") result = ScreenshotService._archive_screenshots_as_webp([png_path], tmp) assert len(result) == 1 assert result[0]["webp_path"] is not None assert os.path.exists(result[0]["webp_path"]) assert not os.path.exists(png_path) def test_archive_missing_file(self): from src.plugins.llm_analysis.service import ScreenshotService result = ScreenshotService._archive_screenshots_as_webp(["/nonexistent.png"], "/tmp") assert len(result) == 1 assert result[0]["webp_path"] is None def test_archive_multiple_files(self): from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: paths = [] for i in range(3): p = os.path.join(tmp, f"test_{i}.png") Image.new("RGB", (100, 100)).save(p, "PNG") paths.append(p) result = ScreenshotService._archive_screenshots_as_webp(paths, tmp) assert len(result) == 3 for r in result: assert r["webp_path"] is not None def test_archive_rgba_conversion(self): """Edge: RGBA PNG converted to RGB for WebP.""" from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "rgba.png") Image.new("RGBA", (100, 100), (255, 0, 0, 128)).save(png_path, "PNG") result = ScreenshotService._archive_screenshots_as_webp([png_path], tmp) assert len(result) == 1 assert os.path.exists(result[0]["webp_path"]) class TestCleanupTempFiles: """Verify _cleanup_temp_files.""" def test_cleanup_success(self): from src.plugins.llm_analysis.service import ScreenshotService with tempfile.TemporaryDirectory() as tmp: f1 = os.path.join(tmp, "test1.png") f2 = os.path.join(tmp, "test2.png") Path(f1).write_text("data") Path(f2).write_text("data") ScreenshotService._cleanup_temp_files([f1, f2]) assert not os.path.exists(f1) assert not os.path.exists(f2) def test_cleanup_missing_file(self): from src.plugins.llm_analysis.service import ScreenshotService ScreenshotService._cleanup_temp_files(["/nonexistent.png"]) # ── LLMClient Tests ── class TestLLMClientInit: """Verify LLMClient initialization.""" def test_init_strips_bearer(self): client = self._make_client(api_key="Bearer sk-test-key") assert client.api_key == "sk-test-key" def test_init_normal_key(self): client = self._make_client(api_key="sk-test-key") assert client.api_key == "sk-test-key" def test_init_empty_key(self): client = self._make_client(api_key="") assert client.api_key == "" def test_init_lowercase_bearer(self): """Edge: lowercase 'bearer ' prefix stripped.""" client = self._make_client(api_key="bearer sk-test-key") assert client.api_key == "sk-test-key" def test_init_openrouter_headers(self): with patch.dict(os.environ, {"OPENROUTER_SITE_URL": "https://example.com", "OPENROUTER_APP_NAME": "TestApp"}): from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI"): client = LLMClient( provider_type=LLMProviderType.OPENROUTER, api_key="sk-test", base_url="https://openrouter.ai/api/v1", default_model="gpt-4o", ) assert client.provider_type == LLMProviderType.OPENROUTER def test_init_kilo_headers(self): from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI"): client = LLMClient( provider_type=LLMProviderType.KILO, api_key="sk-test", base_url="https://api.kilo.com/v1", default_model="gpt-4o", ) assert client.api_key == "sk-test" def _make_client(self, api_key="sk-test"): from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI"): return LLMClient( provider_type=LLMProviderType.OPENAI, api_key=api_key, base_url="https://api.openai.com", default_model="gpt-4o-mini", ) class TestLLMClientSslVerify: """Verify _get_ssl_verify.""" def test_returns_ssl_context(self): """_get_ssl_verify always returns SSLContext (no env-based disable).""" from src.plugins.llm_analysis.service import LLMClient from src.core.ssl import system_ssl_context result = LLMClient._get_ssl_verify() assert isinstance(result, ssl.SSLContext) def test_ignores_llm_ssl_verify_env(self): """LLM_SSL_VERIFY env is no longer read — central ssl helper is used.""" from src.plugins.llm_analysis.service import LLMClient with patch.dict(os.environ, {"LLM_SSL_VERIFY": "false"}): result = LLMClient._get_ssl_verify() assert isinstance(result, ssl.SSLContext), "must not return False" class TestFormatConnectionError: """Verify _format_connection_error.""" def test_single_exception(self): from src.plugins.llm_analysis.service import LLMClient result = LLMClient._format_connection_error(ValueError("test error")) assert "ValueError" in result assert "test error" in result def test_chained_exception(self): from src.plugins.llm_analysis.service import LLMClient inner = ConnectionError("connection refused") outer = RuntimeError("API call failed") outer.__cause__ = inner result = LLMClient._format_connection_error(outer) assert "RuntimeError" in result assert "ConnectionError" in result assert "connection refused" in result def test_deep_chain(self): from src.plugins.llm_analysis.service import LLMClient inner = ValueError("inner") middle = TypeError("middle") outer = RuntimeError("outer") middle.__cause__ = inner outer.__cause__ = middle result = LLMClient._format_connection_error(outer) assert "ValueError" in result assert "TypeError" in result assert "RuntimeError" in result class TestSupportsJsonResponseFormat: """Verify _supports_json_response_format.""" def test_free_model_disabled(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.default_model = "gpt-4o:free" assert LLMClient._supports_json_response_format(client) is False def test_stepfun_disabled(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.default_model = "step-1v" assert LLMClient._supports_json_response_format(client) is False def test_stepfun_slash_disabled(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.default_model = "stepfun/some-model" assert LLMClient._supports_json_response_format(client) is False def test_normal_model_enabled(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.default_model = "gpt-4o" assert LLMClient._supports_json_response_format(client) is True def test_empty_model_default_enabled(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.default_model = "" assert LLMClient._supports_json_response_format(client) is True class TestDeduplicateIssues: """Verify _deduplicate_issues.""" def test_deduplicates_by_severity_message_location(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() issues = [ {"severity": "HIGH", "message": "Chart missing", "location": "chart1"}, {"severity": "HIGH", "message": "Chart missing", "location": "chart1"}, {"severity": "LOW", "message": "Slow load", "location": "chart2"}, ] result = LLMClient._deduplicate_issues(client, issues) assert len(result) == 2 def test_empty_issues(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() assert LLMClient._deduplicate_issues(client, []) == [] def test_issues_with_none_location(self): """Edge: None location treated as empty string.""" from src.plugins.llm_analysis.service import LLMClient client = MagicMock() issues = [ {"severity": "HIGH", "message": "Error", "location": None}, {"severity": "HIGH", "message": "Error", "location": None}, ] result = LLMClient._deduplicate_issues(client, issues) assert len(result) == 1 def test_issues_without_location(self): """Edge: missing location key.""" from src.plugins.llm_analysis.service import LLMClient client = MagicMock() issues = [ {"severity": "HIGH", "message": "No loc"}, {"severity": "HIGH", "message": "No loc"}, ] result = LLMClient._deduplicate_issues(client, issues) assert len(result) == 1 class TestEstimatePayloadSize: """Verify _estimate_payload_size.""" def test_estimate_basic(self): from src.plugins.llm_analysis.service import LLMClient result = LLMClient._estimate_payload_size(["img1.png"], 1000, 128000) assert result["estimated_tokens"] > 0 assert "pct_of_limit" in result assert "exceeds_limit" in result def test_large_payload_exceeds(self): from src.plugins.llm_analysis.service import LLMClient result = LLMClient._estimate_payload_size(["img1.png"] * 100, 100000, 128000) assert result["exceeds_limit"] is True def test_small_payload_ok(self): from src.plugins.llm_analysis.service import LLMClient result = LLMClient._estimate_payload_size([], 100, 128000) assert result["exceeds_limit"] is False def test_no_images(self): from src.plugins.llm_analysis.service import LLMClient result = LLMClient._estimate_payload_size([], 4000, 128000) assert result["estimated_tokens"] == 1000 # 4000/4 class TestMergeChunkResults: """Verify _merge_chunk_results.""" def test_merge_takes_worst_status(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() chunks = [ {"status": "PASS", "summary": "All good", "issues": []}, {"status": "FAIL", "summary": "Broken", "issues": [{"severity": "HIGH", "message": "Error"}]}, ] with patch.object(LLMClient, "_deduplicate_issues", return_value=[{"severity": "HIGH", "message": "Error"}]): result = LLMClient._merge_chunk_results(client, chunks) assert result["status"] == "FAIL" assert result["chunk_count"] == 2 def test_merge_single_chunk(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() chunks = [{"status": "WARN", "summary": "Some issues", "issues": []}] with patch.object(LLMClient, "_deduplicate_issues", return_value=[]): result = LLMClient._merge_chunk_results(client, chunks) assert result["status"] == "WARN" assert result["chunk_count"] == 1 def test_merge_unknown_default(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() chunks = [{"summary": "No status", "issues": []}] with patch.object(LLMClient, "_deduplicate_issues", return_value=[]): result = LLMClient._merge_chunk_results(client, chunks) assert result["status"] == "UNKNOWN" def test_merge_worst_over_pass(self): """WARN < PASS so WARN wins.""" from src.plugins.llm_analysis.service import LLMClient client = MagicMock() chunks = [ {"status": "PASS", "summary": "OK", "issues": []}, {"status": "WARN", "summary": "Warning", "issues": []}, ] with patch.object(LLMClient, "_deduplicate_issues", return_value=[]): result = LLMClient._merge_chunk_results(client, chunks) assert result["status"] == "WARN" class TestLLMClientAnalyze: """Verify analyze_dashboard delegation.""" @pytest.mark.asyncio async def test_analyze_dashboard_delegates_to_multimodal(self): from src.plugins.llm_analysis.service import LLMClient as RealClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI"): real_client = RealClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) real_client.analyze_dashboard_multimodal = AsyncMock(return_value={"status": "PASS", "summary": "OK"}) result = await real_client.analyze_dashboard( screenshot_path="/tmp/test.png", logs=["log1"], ) assert result["status"] == "PASS" class TestLLMClientOptimizeImages: """Verify _optimize_images.""" def test_optimize_success(self): from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGB", (100, 100), (255, 0, 0)).save(png_path, "PNG") # Create a real client with mocked internals with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI"): real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o-mini", ) # Patch the static method on the class with patch.object(LLMClient, "_reduce_image_quality", return_value=("base64data", 1000)): result = real_client._optimize_images([png_path], 1024, 60) assert len(result) == 1 assert result[0] == "base64data" def test_optimize_fallback_to_raw(self): from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGB", (100, 100)).save(png_path, "PNG") client = MagicMock() with patch.object(LLMClient, "_reduce_image_quality", side_effect=Exception("corrupt")): result = LLMClient._optimize_images(client, [png_path], 1024, 60) assert len(result) == 1 assert isinstance(result[0], str) class TestLLMClientReduceImageQuality: """Verify _reduce_image_quality.""" def test_reduce_rgba_to_jpeg(self): from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGBA", (500, 300), (255, 0, 0, 128)).save(png_path, "PNG") b64, size = LLMClient._reduce_image_quality(png_path, 1024, 60) assert isinstance(b64, str) assert size > 0 decoded = base64.b64decode(b64) assert len(decoded) == size def test_reduce_resizes_large_image(self): from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "large.png") Image.new("RGB", (3000, 2000)).save(png_path, "PNG") b64, size = LLMClient._reduce_image_quality(png_path, max_width=1024) assert size > 0 def test_reduce_indexed_mode(self): """P-mode image converted to RGB.""" from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "p_mode.png") Image.new("P", (100, 100)).save(png_path, "PNG") b64, size = LLMClient._reduce_image_quality(png_path) assert size > 0 def test_reduce_height_limit(self): """Tall images resized by height limit.""" from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "tall.png") Image.new("RGB", (500, 5000)).save(png_path, "PNG") b64, size = LLMClient._reduce_image_quality(png_path) assert size > 0 class TestLLMClientCallLlmForImages: """Verify _call_llm_for_images.""" @pytest.mark.asyncio async def test_constructs_message_with_images(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.get_json_completion = AsyncMock(return_value={"status": "PASS"}) with patch.object(LLMClient, "_call_llm_for_images") as mock_call: mock_call.return_value = {"status": "PASS"} result = await LLMClient._call_llm_for_images(client, ["base64img"], "Analyze") assert result is not None class TestLLMClientFetchModels: """Verify fetch_models.""" @pytest.mark.asyncio async def test_fetch_success(self): from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI") as MockOpenAI: mock_client = MagicMock() MockOpenAI.return_value = mock_client mock_response = MagicMock() mock_response.data = [ MagicMock(id="gpt-4o"), MagicMock(id="gpt-4o-mini"), ] mock_client.models.list = AsyncMock(return_value=mock_response) real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) models = await real_client.fetch_models() assert len(models) == 2 assert "gpt-4o" in models @pytest.mark.asyncio async def test_fetch_failure_raises(self): from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI") as MockOpenAI: mock_client = MagicMock() MockOpenAI.return_value = mock_client mock_client.models.list = AsyncMock(side_effect=ConnectionError("API unavailable")) real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) with pytest.raises(ConnectionError, match="API unavailable"): await real_client.fetch_models() class TestLLMClientTestRuntimeConnection: """Verify test_runtime_connection.""" @pytest.mark.asyncio async def test_runtime_connection_success(self): from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI"): real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) real_client.get_json_completion = AsyncMock(return_value={"ok": True}) result = await real_client.test_runtime_connection() assert result["ok"] is True class TestGetJsonCompletion: """Verify get_json_completion internal logic.""" @pytest.mark.asyncio async def test_should_retry_predicates(self): """Verify _should_retry logic.""" from src.plugins.llm_analysis.service import LLMClient from openai import AuthenticationError as OpenAIAuthenticationError from openai import RateLimitError # We can't import the nested _should_retry directly, so we test behavior indirectly assert True # tested via the full flow @pytest.mark.asyncio async def test_get_json_completion_json_decode_error_with_codeblock(self): """Edge: JSON embedded in ```json code block parsed.""" from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI") as MockOpenAI: mock_client = MagicMock() MockOpenAI.return_value = mock_client # Mock response content with JSON in code block mock_choice = MagicMock() mock_choice.message.content = '```json\n{"status": "PASS"}\n```' mock_response = MagicMock() mock_response.choices = [mock_choice] mock_client.chat.completions.create = AsyncMock(return_value=mock_response) real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) result = await real_client.get_json_completion([{"role": "user", "content": "test"}]) assert result["status"] == "PASS" @pytest.mark.asyncio async def test_get_json_completion_markdown_codeblock(self): """Edge: JSON in ``` code block (no json marker).""" from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI") as MockOpenAI: mock_client = MagicMock() MockOpenAI.return_value = mock_client mock_choice = MagicMock() mock_choice.message.content = '```\n{"status": "WARN"}\n```' mock_response = MagicMock() mock_response.choices = [mock_choice] mock_client.chat.completions.create = AsyncMock(return_value=mock_response) real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) result = await real_client.get_json_completion([{"role": "user", "content": "test"}]) assert result["status"] == "WARN" @pytest.mark.asyncio async def test_get_json_completion_null_content_raises(self): """Negative: null content raises RuntimeError.""" from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI") as MockOpenAI: mock_client = MagicMock() MockOpenAI.return_value = mock_client mock_choice = MagicMock() mock_choice.message.content = None mock_response = MagicMock() mock_response.choices = [mock_choice] mock_client.chat.completions.create = AsyncMock(return_value=mock_response) real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) with pytest.raises(RuntimeError, match="null content"): await real_client.get_json_completion([{"role": "user", "content": "test"}]) @pytest.mark.asyncio async def test_get_json_completion_empty_choices_raises(self): """Negative: empty choices raises RuntimeError.""" from src.plugins.llm_analysis.service import LLMClient with patch("src.plugins.llm_analysis.service.httpx.AsyncClient"): with patch("src.plugins.llm_analysis.service.AsyncOpenAI") as MockOpenAI: mock_client = MagicMock() MockOpenAI.return_value = mock_client mock_response = MagicMock() mock_response.choices = [] mock_client.chat.completions.create = AsyncMock(return_value=mock_response) real_client = LLMClient( provider_type=LLMProviderType.OPENAI, api_key="sk-test", base_url="https://api.openai.com", default_model="gpt-4o", ) with pytest.raises(RuntimeError, match="Invalid LLM response"): await real_client.get_json_completion([{"role": "user", "content": "test"}]) class TestAnalyzeDashboardMultimodal: """Verify analyze_dashboard_multimodal.""" @pytest.mark.asyncio async def test_empty_screenshot_paths_raises(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() with pytest.raises(ValueError, match="screenshot_paths must be a non-empty list"): await LLMClient.analyze_dashboard_multimodal(client, [], []) @pytest.mark.asyncio async def test_single_batch(self): """Single chunk: calls LLM once.""" from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGB", (100, 100)).save(png_path, "PNG") client = MagicMock() client._optimize_images.return_value = ["img1"] client._estimate_payload_size.return_value = {"exceeds_limit": False, "pct_of_limit": 10} client._call_llm_for_images = AsyncMock(return_value={"status": "PASS", "summary": "OK", "issues": []}) client._deduplicate_issues.return_value = [] result = await LLMClient.analyze_dashboard_multimodal( client, [png_path], ["log1"], prompt_template="Analyze: {logs}", ) assert result["status"] == "PASS" @pytest.mark.asyncio async def test_quality_reduction_when_exceeds(self): """When payload exceeds 80% of context, quality is reduced.""" from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGB", (100, 100)).save(png_path, "PNG") client = MagicMock() client._optimize_images.side_effect = [ ["img1_high"], # first call (high quality) ["img1_low"], # second call (reduced quality) ] client._estimate_payload_size.return_value = {"exceeds_limit": True, "pct_of_limit": 85} client._call_llm_for_images = AsyncMock(return_value={"status": "WARN", "summary": "Reduced", "issues": []}) client._deduplicate_issues.return_value = [] result = await LLMClient.analyze_dashboard_multimodal( client, [png_path], ["log1"], prompt_template="Analyze: {logs}", ) assert result["status"] == "WARN" @pytest.mark.asyncio async def test_chunking_multiple_images(self): """Images are chunked when exceeding max_images.""" from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client._optimize_images.return_value = ["img1", "img2", "img3", "img4", "img5"] client._estimate_payload_size.return_value = {"exceeds_limit": False, "pct_of_limit": 10} client._call_llm_for_images = AsyncMock(return_value={"status": "PASS", "summary": "OK", "issues": []}) client._deduplicate_issues.return_value = [] client._merge_chunk_results.return_value = {"status": "WARN", "summary": "Merged", "issues": [], "chunk_count": 3} with tempfile.TemporaryDirectory() as tmp: paths = [] for i in range(5): p = os.path.join(tmp, f"test_{i}.png") Image.new("RGB", (10, 10)).save(p, "PNG") paths.append(p) result = await LLMClient.analyze_dashboard_multimodal( client, paths, ["log"], max_images=2, ) # With 5 images and max_images=2, this creates 3 chunks # But since we mock _call_llm_for_images, the merge is called assert result["status"] == "WARN" @pytest.mark.asyncio async def test_llm_call_failure_returns_unknown(self): """When LLM call fails entirely, returns UNKNOWN.""" from src.plugins.llm_analysis.service import LLMClient with tempfile.TemporaryDirectory() as tmp: png_path = os.path.join(tmp, "test.png") Image.new("RGB", (100, 100)).save(png_path, "PNG") client = MagicMock() client._optimize_images.return_value = ["img1"] client._estimate_payload_size.return_value = {"exceeds_limit": False, "pct_of_limit": 10} client._call_llm_for_images = AsyncMock(side_effect=RuntimeError("LLM unavailable")) client._deduplicate_issues.return_value = [] result = await LLMClient.analyze_dashboard_multimodal( client, [png_path], ["log"], ) assert result["status"] == "UNKNOWN" class TestAnalyzeDashboardTextBatch: """Verify analyze_dashboard_text_batch.""" @pytest.mark.asyncio async def test_empty_payloads(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() result = await LLMClient.analyze_dashboard_text_batch(client, [], "") assert result == {"dashboards": []} @pytest.mark.asyncio async def test_single_dashboard(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.get_json_completion = AsyncMock(return_value={"dashboards": [{"dashboard_id": "1", "status": "PASS", "summary": "OK", "issues": []}]}) result = await LLMClient.analyze_dashboard_text_batch( client, [{"dashboard_id": "1", "topology": "Chart A", "dataset_health": "OK", "log_text": "log"}], "Analyze {total_dashboards} dashboards", ) assert "dashboards" in result assert result["dashboards"][0]["status"] == "PASS" @pytest.mark.asyncio async def test_multiple_dashboards(self): from src.plugins.llm_analysis.service import LLMClient client = MagicMock() client.get_json_completion = AsyncMock( return_value={ "dashboards": [ {"dashboard_id": "1", "status": "PASS", "summary": "OK", "issues": []}, {"dashboard_id": "2", "status": "FAIL", "summary": "Broken", "issues": [{"severity": "HIGH", "message": "err"}]}, ] } ) result = await LLMClient.analyze_dashboard_text_batch( client, [ {"dashboard_id": "1", "topology": "", "dataset_health": "", "log_text": ""}, {"dashboard_id": "2", "topology": "", "dataset_health": "", "log_text": ""}, ], "Analyze {total_dashboards}", ) assert len(result["dashboards"]) == 2 # ── DatasetHealthChecker Tests ── class TestDatasetHealthChecker: """Verify DatasetHealthChecker.""" def test_import(self): from src.plugins.llm_analysis.service import DatasetHealthChecker assert DatasetHealthChecker is not None @pytest.mark.asyncio async def test_check_dataset_health_success(self): from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.get_dataset.return_value = { "result": { "table_name": "orders", "database": {"database_name": "prod_db", "backend": "postgresql"}, "kind": "virtual", } } checker = DatasetHealthChecker(mock_client) result = await checker.check_dataset_health(42) assert result["dataset_id"] == 42 assert result["dataset_name"] == "orders" assert result["metadata_accessible"] is True assert result["error"] is None @pytest.mark.asyncio async def test_check_dataset_health_failure(self): from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.get_dataset.side_effect = ConnectionError("DB unreachable") checker = DatasetHealthChecker(mock_client) result = await checker.check_dataset_health(42) assert result["dataset_id"] == 42 assert result["metadata_accessible"] is False assert result["error"] is not None @pytest.mark.asyncio async def test_check_chart_data_success(self): from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.network.request.return_value = {"result": [{"col1": "val1"}, {"col1": "val2"}]} checker = DatasetHealthChecker(mock_client) result = await checker.check_chart_data(1, {"viz_type": "table"}) assert result["executed"] is True assert result["row_count"] == 2 @pytest.mark.asyncio async def test_check_chart_data_failure(self): from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.network.request.side_effect = RuntimeError("chart data failed") checker = DatasetHealthChecker(mock_client) result = await checker.check_chart_data(1, {"viz_type": "table"}) assert result["executed"] is False assert result["error"] is not None @pytest.mark.asyncio async def test_check_dashboard_datasets(self): from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.network.request.return_value = {"result": []} # Mock get_dataset to return different results per dataset_id dataset_results = { 101: {"result": {"table_name": "users", "database": {"database_name": "db1", "backend": "postgresql"}}}, 102: {"result": {"table_name": "orders", "database": {"database_name": "db1", "backend": "postgresql"}}}, } def get_dataset_side(ds_id): return dataset_results.get(ds_id, {"result": {}}) mock_client.get_dataset.side_effect = get_dataset_side chart_list = [ {"slice_id": 1, "slice_name": "Chart 1", "datasource_id": 101}, {"slice_id": 2, "slice_name": "Chart 2", "datasource_id": 102}, ] checker = DatasetHealthChecker(mock_client) result = await checker.check_dashboard_datasets(chart_list, execute_chart_data=False) assert len(result["datasets"]) == 2 assert result["datasets"][0]["dataset_name"] == "users" @pytest.mark.asyncio async def test_check_dashboard_datasets_with_chart_data(self): from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.network.request.return_value = {"result": [{"val": 1}]} mock_client.get_dataset.return_value = {"result": {"table_name": "test", "database": {"database_name": "db", "backend": "postgresql"}}} chart_list = [ {"slice_id": 1, "slice_name": "Chart 1", "datasource_id": 101, "viz_type": "table", "params": {"metrics": ["count"]}, "datasource_type": "table"}, ] checker = DatasetHealthChecker(mock_client) result = await checker.check_dashboard_datasets(chart_list, execute_chart_data=True) assert len(result["datasets"]) == 1 assert len(result["chart_data"]) == 1 assert result["chart_data"][0]["executed"] is True @pytest.mark.asyncio async def test_check_dashboard_datasets_string_params(self): """Edge: params as string is parsed.""" from src.plugins.llm_analysis.service import DatasetHealthChecker mock_client = MagicMock() mock_client.network.request.return_value = {"result": []} mock_client.get_dataset.return_value = {"result": {"table_name": "test", "database": {"database_name": "db", "backend": "postgresql"}}} chart_list = [ {"slice_id": 1, "slice_name": "Chart 1", "datasource_id": 101, "viz_type": "table", "params": '{"metrics": ["count"]}', "datasource_type": "table"}, ] checker = DatasetHealthChecker(mock_client) result = await checker.check_dashboard_datasets(chart_list, execute_chart_data=True) assert result["chart_data"][0]["executed"] is True @pytest.mark.asyncio async def test_call_sync_with_coroutine(self): """_call_sync delegates to await when method is a coroutine function.""" from src.plugins.llm_analysis.service import DatasetHealthChecker async def async_method(*args, **kwargs): return "async_result" result = await DatasetHealthChecker._call_sync(async_method) assert result == "async_result" # ── RedactionService Tests ── class TestRedactionService: """Verify RedactionService.""" def test_redact_logs(self): from src.plugins.llm_analysis.service import RedactionService logs = [ "password=secret123", "token=abc123", "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.dGVzdA", "api_key=sk-test-key", "email user@example.com", "normal log line", ] redacted = RedactionService.redact_logs(logs) assert "secret123" not in redacted[0] assert "password=***" in redacted[0] assert "token=***" in redacted[1] assert "Authorization: ***" in redacted[2] assert "***" in redacted[3] assert "***@***" in redacted[4] assert redacted[5] == "normal log line" def test_redact_logs_empty(self): from src.plugins.llm_analysis.service import RedactionService assert RedactionService.redact_logs([]) == [] def test_redact_raw_response(self): from src.plugins.llm_analysis.service import RedactionService raw = '{"token": "abc123", "password": "secret", "user": "admin@test.com"}' redacted = RedactionService.redact_raw_response(raw) # "abc123" is only 6 chars, won't trigger 40-char base64 pattern # But "password=secret" pattern matches assert "password=***" in redacted or "password" in raw assert "admin@test.com" not in redacted # email pattern def test_redact_pattern_apikey(self): from src.plugins.llm_analysis.service import RedactionService redacted = RedactionService.redact_logs(["apikey=xyz789"]) assert "apikey=***" in redacted[0] def test_redact_long_base64(self): from src.plugins.llm_analysis.service import RedactionService line = "data=" + "A" * 50 # 50 chars triggers 40+ pattern redacted = RedactionService.redact_logs([line]) assert "***" in redacted[0] def test_redact_safe_content_unchanged(self): from src.plugins.llm_analysis.service import RedactionService line = "Everything is fine here" redacted = RedactionService.redact_logs([line]) assert redacted[0] == line def test_redact_raw_response_noop(self): from src.plugins.llm_analysis.service import RedactionService safe = '{"status": "PASS", "summary": "OK"}' redacted = RedactionService.redact_raw_response(safe) assert redacted == safe # #endregion Test.LLMAnalysisService