# #region Test.LLMAnalysisPlugin [C:3] [TYPE Module] [SEMANTICS test, llm, plugin, validation, documentation] # @BRIEF Comprehensive tests for LLMAnalysisPlugin module — helpers, DashboardValidationPlugin, DocumentationPlugin. # @RELATION BINDS_TO -> [LLMAnalysisPlugin] # @TEST_EDGE: masked_api_key -> Detected as invalid # @TEST_EDGE: json_safe_datetime -> datetime converted to ISO string # @TEST_EDGE: ensure_json_prompt -> Appends JSON format when missing # @TEST_EDGE: update_run_status -> Aggregates record statuses correctly from datetime import UTC, datetime, timedelta import json from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock import pytest # ── Module-level Function Tests ── class TestIsMaskedOrInvalidApiKey: """Verify _is_masked_or_invalid_api_key.""" def test_none_is_invalid(self): from src.plugins.llm_analysis.plugin import _is_masked_or_invalid_api_key assert _is_masked_or_invalid_api_key(None) is True def test_empty_is_invalid(self): from src.plugins.llm_analysis.plugin import _is_masked_or_invalid_api_key assert _is_masked_or_invalid_api_key("") is True def test_masked_is_invalid(self): from src.plugins.llm_analysis.plugin import _is_masked_or_invalid_api_key assert _is_masked_or_invalid_api_key("********") is True assert _is_masked_or_invalid_api_key("EMPTY_OR_NONE") is True def test_short_is_invalid(self): from src.plugins.llm_analysis.plugin import _is_masked_or_invalid_api_key assert _is_masked_or_invalid_api_key("short") is True # len < 16 def test_valid_key(self): from src.plugins.llm_analysis.plugin import _is_masked_or_invalid_api_key assert _is_masked_or_invalid_api_key("sk-valid-key-that-is-long-enough") is False def test_exactly_16_chars(self): from src.plugins.llm_analysis.plugin import _is_masked_or_invalid_api_key assert _is_masked_or_invalid_api_key("1234567890123456") is False # len == 16, valid class TestJsonSafeValue: """Verify _json_safe_value.""" def test_datetime_converted(self): from src.plugins.llm_analysis.plugin import _json_safe_value dt = datetime(2024, 1, 15, 10, 30, 0, tzinfo=UTC) result = _json_safe_value(dt) assert result == "2024-01-15T10:30:00+00:00" def test_nested_dict(self): from src.plugins.llm_analysis.plugin import _json_safe_value dt = datetime(2024, 6, 1, 12, 0, 0, tzinfo=UTC) data = {"name": "test", "created": dt, "items": [{"updated": dt}]} result = _json_safe_value(data) assert result["created"] == "2024-06-01T12:00:00+00:00" assert result["items"][0]["updated"] == "2024-06-01T12:00:00+00:00" def test_plain_value(self): from src.plugins.llm_analysis.plugin import _json_safe_value assert _json_safe_value("hello") == "hello" assert _json_safe_value(42) == 42 assert _json_safe_value(None) is None def test_empty_structures(self): from src.plugins.llm_analysis.plugin import _json_safe_value assert _json_safe_value({}) == {} assert _json_safe_value([]) == [] class TestEnsureJsonPrompt: """Verify _ensure_json_prompt.""" def test_none_returns_default(self): from src.plugins.llm_analysis.plugin import _ensure_json_prompt result = _ensure_json_prompt(None) assert result.startswith("Analyze the dashboard") assert '"status"' in result def test_empty_returns_default(self): from src.plugins.llm_analysis.plugin import _ensure_json_prompt result = _ensure_json_prompt("") assert result.startswith("Analyze the dashboard") # empty string is falsy assert '"status"' in result def test_already_has_json_format(self): from src.plugins.llm_analysis.plugin import _ensure_json_prompt prompt = 'Analyze this dashboard and return JSON with "status" and "summary"' result = _ensure_json_prompt(prompt) assert result == prompt # unchanged def test_missing_json_appended(self): from src.plugins.llm_analysis.plugin import _ensure_json_prompt prompt = "Tell me about this dashboard" result = _ensure_json_prompt(prompt) assert 'Respond ONLY with valid JSON' in result def test_has_json_keyword_but_no_structure(self): """Has 'json' keyword but no '"status"' or '"summary"'. JSON instruction appended.""" from src.plugins.llm_analysis.plugin import _ensure_json_prompt prompt = "Return the data in json format" result = _ensure_json_prompt(prompt) assert 'Respond ONLY with valid JSON' in result def test_has_curly_braces_json(self): """Prompt already has { and " suggesting JSON structure.""" from src.plugins.llm_analysis.plugin import _ensure_json_prompt prompt = '{ "result": "ok" }' result = _ensure_json_prompt(prompt) assert result == prompt class TestUpdateRunStatus: """Verify _update_run_status.""" def test_no_run_id_returns_early(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() _update_run_status(db, None) # no crash db.query.assert_not_called() def test_empty_run_id_returns_early(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() _update_run_status(db, "") # no crash db.query.assert_not_called() def test_no_records_found(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() db.query.return_value.filter.return_value.all.return_value = [] db.query.return_value.filter.return_value.first.return_value = MagicMock() _update_run_status(db, "run-1") # Should log and return without commit db.commit.assert_not_called() def test_no_run_found(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() db.query.return_value.filter.return_value.all.return_value = [MagicMock()] db.query.return_value.filter.return_value.first.return_value = None _update_run_status(db, "run-1") db.commit.assert_not_called() def test_partial_records_not_complete(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() records = [MagicMock(status="PASS"), MagicMock(status="WARN")] run = MagicMock() run.dashboard_count = 5 # expect 5, got 2 db.query.return_value.filter.return_value.all.return_value = records db.query.return_value.filter.return_value.first.return_value = run _update_run_status(db, "run-1") db.commit.assert_not_called() def test_all_records_complete(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() records = [ MagicMock(status="PASS"), MagicMock(status="WARN"), MagicMock(status="FAIL"), MagicMock(status="UNKNOWN"), ] run = MagicMock() run.dashboard_count = 4 db.query.return_value.filter.return_value.all.return_value = records db.query.return_value.filter.return_value.first.return_value = run _update_run_status(db, "run-1") assert run.status == 'completed' assert run.pass_count == 1 assert run.warn_count == 1 assert run.fail_count == 1 assert run.unknown_count == 1 db.commit.assert_called_once() def test_all_pass_updates_correctly(self): from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() records = [MagicMock(status="PASS"), MagicMock(status="PASS")] run = MagicMock() run.dashboard_count = 2 db.query.return_value.filter.return_value.all.return_value = records db.query.return_value.filter.return_value.first.return_value = run _update_run_status(db, "run-1") assert run.pass_count == 2 assert run.fail_count == 0 db.commit.assert_called_once() def test_exception_handled(self): """Exception during update does not propagate.""" from src.plugins.llm_analysis.plugin import _update_run_status db = MagicMock() db.query.side_effect = Exception("DB error") _update_run_status(db, "run-1") # should not raise # ── DashboardValidationPlugin Tests ── class TestDashboardValidationPluginProperties: """Verify DashboardValidationPlugin static properties.""" def test_properties(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() assert plugin.id == "llm_dashboard_validation" assert plugin.name == "Dashboard LLM Validation" assert plugin.description == "Automated dashboard health analysis using LLMs." assert plugin.version == "1.0.0" def test_get_schema(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() schema = plugin.get_schema() assert schema["type"] == "object" assert "dashboard_id" in schema["properties"] assert "environment_id" in schema["properties"] assert "provider_id" in schema["required"] class TestDashboardValidationPluginExecute: """Verify DashboardValidationPlugin.execute — main dispatch.""" def _make_mock_db(self): """Create a mock DB session.""" mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_db.close.return_value = None return mock_db def _make_mock_provider(self, is_multimodal=True): """Create a mock DB provider.""" provider = MagicMock() provider.id = "prov-1" provider.name = "Test Provider" provider.provider_type = "openai" provider.base_url = "https://api.openai.com" provider.default_model = "gpt-4o" provider.is_active = True provider.is_multimodal = is_multimodal provider.max_images = 10 return provider async def _run_execute(self, plugin, params, mock_db, mock_env, mock_provider, api_key="sk-real-key-that-is-valid", context=None, patch_path_a=None, patch_path_b=None): """Helper to run execute with all patching.""" with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_env_obj = mock_env if mock_env else MagicMock() mock_cm.get_environment.return_value = mock_env_obj with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = mock_provider MockProviderService.return_value.get_decrypted_api_key.return_value = api_key if patch_path_a is not None: with patch.object(plugin, '_execute_path_a', new=AsyncMock(return_value=patch_path_a)): return await plugin.execute(params, context=context) elif patch_path_b is not None: with patch.object(plugin, '_execute_path_b', new=AsyncMock(return_value=patch_path_b)): return await plugin.execute(params, context=context) else: return await plugin.execute(params, context=context) @pytest.mark.asyncio async def test_execute_path_a(self): """Happy: Path A (screenshot) succeeds.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider(is_multimodal=True) result = await self._run_execute( plugin, {"dashboard_id": "42", "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": True}, mock_db, MagicMock(), provider, patch_path_a={"dashboard_id": "42", "execution_path": "multimodal", "status": "PASS"}, ) assert result["total"] == 1 @pytest.mark.asyncio async def test_execute_path_b(self): """Happy: Path B (text-only) succeeds.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider(is_multimodal=False) result = await self._run_execute( plugin, {"dashboard_id": "42", "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": False}, mock_db, MagicMock(), provider, patch_path_b={"dashboard_id": "42", "execution_path": "text_only", "status": "WARN"}, ) assert result["total"] == 1 @pytest.mark.asyncio async def test_execute_env_not_found(self): """Negative: env not found raises.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = None with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-valid" with pytest.raises(ValueError, match="env-1 not found"): await plugin.execute({ "dashboard_id": "42", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_execute_provider_not_found(self): """Negative: provider not found raises.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() mock_env = MagicMock() mock_env.id = "env-1" with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = mock_env with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = None with pytest.raises(ValueError, match="prov-1 not found"): await plugin.execute({ "dashboard_id": "42", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_execute_invalid_api_key(self): """Negative: masked/invalid API key raises.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() mock_env = MagicMock() provider = self._make_mock_provider() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = mock_env with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "********" with pytest.raises(ValueError, match="Invalid API key"): await plugin.execute({ "dashboard_id": "42", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_execute_no_dashboard_id(self): """Negative: no dashboard_id raises.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-valid" with pytest.raises(ValueError, match="No dashboard_id"): await plugin.execute({ "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_execute_screenshot_without_multimodal_raises(self): """Negative: screenshot_enabled but provider not multimodal.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() mock_env = MagicMock() provider = self._make_mock_provider(is_multimodal=False) with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = mock_env with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-valid" with pytest.raises(ValueError, match="multimodal"): await plugin.execute({ "dashboard_id": "42", "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": True, }) @pytest.mark.asyncio async def test_execute_with_dashboard_ids_list(self): """Happy: v2 dashboard_ids list processed.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() mock_env = MagicMock() result = await self._run_execute( plugin, {"dashboard_ids": ["42", "43"], "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": False}, mock_db, mock_env, provider, patch_path_b={"dashboard_id": "42", "status": "PASS"}, ) assert result["total"] == 2 @pytest.mark.asyncio async def test_execute_continues_on_dashboard_failure(self): """Negative: one dashboard fails, loop continues to next.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() mock_env = MagicMock() call_count = [0] async def path_b_side(params, ctx, db, prov, key, env, cm, log): call_count[0] += 1 if call_count[0] == 1: raise ValueError("First dashboard failed") return {"dashboard_id": params.get("dashboard_id"), "status": "PASS"} with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = mock_env with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-valid" with patch.object(plugin, '_execute_path_b', new=AsyncMock(side_effect=path_b_side)): with pytest.raises(ValueError, match="First dashboard failed"): await plugin.execute({ "dashboard_ids": ["42", "43"], "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": False, }) assert call_count[0] == 2 @pytest.mark.asyncio async def test_execute_with_context(self): """Edge: TaskContext provided.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() mock_ctx = MagicMock() mock_ctx.logger = MagicMock() mock_ctx.logger.with_source.return_value = MagicMock() mock_ctx.task_id = "task-1" mock_env = MagicMock() result = await self._run_execute( plugin, {"dashboard_ids": ["42"], "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": False}, mock_db, mock_env, provider, context=mock_ctx, patch_path_b={"dashboard_id": "42", "status": "PASS"}, ) assert result["total"] == 1 @pytest.mark.asyncio async def test_execute_url_reparse(self): """Edge: dashboard URL provided triggers URL re-parsing.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-valid" with patch('src.core.utils.superset_context_extractor.SupersetContextExtractor') as MockExtractor: mock_ext = MagicMock() MockExtractor.return_value = mock_ext mock_parsed = MagicMock() mock_parsed.dashboard_id = "99" mock_ext.parse_superset_link.return_value = mock_parsed with patch.object(plugin, '_execute_path_b', new=AsyncMock(return_value={ "dashboard_id": "99", "status": "PASS", })): result = await plugin.execute({ "dashboard_id": "42", "dashboard_url": "http://example.com/superset/dashboard/99/", "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": False, }) assert result["total"] == 1 @pytest.mark.asyncio async def test_execute_url_reparse_failure_sets_source_invalid(self): """Edge: URL re-parse failure sets source_invalid flag.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = self._make_mock_db() provider = self._make_mock_provider() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = provider MockProviderService.return_value.get_decrypted_api_key.return_value = "sk-real-key-that-is-valid" with patch('src.core.utils.superset_context_extractor.SupersetContextExtractor') as MockExtractor: mock_ext = MagicMock() MockExtractor.return_value = mock_ext mock_ext.parse_superset_link.side_effect = ValueError("Bad URL") with patch.object(plugin, '_execute_path_b', new=AsyncMock(return_value={ "dashboard_id": "42", "status": "PASS", })): result = await plugin.execute({ "dashboard_id": "42", "dashboard_url": "bad-url", "environment_id": "env-1", "provider_id": "prov-1", "screenshot_enabled": False, }) assert result["total"] == 1 class TestExecutePathA: """Verify _execute_path_a — screenshot + multimodal LLM flow.""" @pytest.mark.asyncio async def test_full_flow(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.storage.root_path = "/tmp/storage" mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = { "dashboard_id": "42", "environment_id": "env-1", "run_id": "run-42", } # Mock ScreenshotService with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS: mock_ss = MagicMock() MockSS.return_value = mock_ss mock_ss.capture_dashboard = AsyncMock(return_value=(["/tmp/jpeg.jpg"], [{"webp_path": "/tmp/archive.webp"}])) # Mock LLMClient with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.analyze_dashboard_multimodal = AsyncMock(return_value={ "status": "PASS", "summary": "All good", "issues": [], "chunk_count": 1, }) # Mock _fetch_dashboard_logs with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log entry"], {"logs_fetch_duration_ms": 10} ))): # Mock cleanup with patch('src.plugins.llm_analysis.plugin.ScreenshotService._cleanup_temp_files'): # Mock NotificationService with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: mock_notif = MagicMock() MockNotif.return_value = mock_notif mock_notif.dispatch_report = AsyncMock() result = await plugin._execute_path_a( params=params, context=None, db=mock_db, db_provider=mock_provider, api_key="sk-real-key", env=mock_env, config_mgr=mock_cm, log=mock_log, ) assert result["execution_path"] == "multimodal" assert result["status"] == "PASS" assert "timings" in result mock_db.add.assert_called_once() mock_db.commit.assert_called() @pytest.mark.asyncio async def test_no_screenshots_falls_back_to_text(self): """When capture_dashboard returns no screenshots, fallback to text-only.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.storage.root_path = "/tmp/storage" mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = {"dashboard_id": "42", "environment_id": "env-1"} with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS: mock_ss = MagicMock() MockSS.return_value = mock_ss # No screenshots captured mock_ss.capture_dashboard = AsyncMock(return_value=([], [])) with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.analyze_dashboard_text_batch = AsyncMock(return_value={ "dashboards": [{"status": "WARN", "summary": "No screenshots", "issues": []}] }) with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log"], {"logs_fetch_duration_ms": 5} ))): with patch('src.plugins.llm_analysis.plugin.ScreenshotService._cleanup_temp_files'): with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: MockNotif.return_value.dispatch_report = AsyncMock() result = await plugin._execute_path_a( params, None, mock_db, mock_provider, "sk-key", mock_env, mock_cm, mock_log, ) assert result["status"] == "WARN" @pytest.mark.asyncio async def test_notification_failure_does_not_block(self): """Edge: notification failure is logged but does not block return.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.storage.root_path = "/tmp/storage" mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = {"dashboard_id": "42", "environment_id": "env-1"} with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS: MockSS.return_value.capture_dashboard = AsyncMock(return_value=(["/tmp/jpg"], [{"webp_path": "/tmp/webp"}])) with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: MockLLM.return_value.analyze_dashboard_multimodal = AsyncMock(return_value={ "status": "PASS", "summary": "OK", "issues": [], "chunk_count": 1, }) with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log"], {"logs_fetch_duration_ms": 5} ))): with patch('src.plugins.llm_analysis.plugin.ScreenshotService._cleanup_temp_files'): with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: MockNotif.return_value.dispatch_report = AsyncMock(side_effect=RuntimeError("Notif failed")) result = await plugin._execute_path_a( params, None, mock_db, mock_provider, "sk-key", mock_env, mock_cm, mock_log, ) assert result["status"] == "PASS" # notification failure is non-blocking class TestExecutePathB: """Verify _execute_path_b — text-only LLM flow.""" @pytest.mark.asyncio async def test_full_flow(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.llm = {"prompts": {"dashboard_validation_prompt_text": "Analyze: {topology}"}} mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = {"dashboard_id": "42", "environment_id": "env-1", "run_id": "run-99"} # Mock SupersetClient with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC: mock_sc = MagicMock() MockSC.return_value = mock_sc mock_sc.get_dashboard = AsyncMock(return_value={ "result": { "slices": [{"id": 1, "slice_id": 1}], } }) mock_sc.get_chart = AsyncMock(return_value={ "result": { "id": 1, "slice_name": "Chart 1", "datasource_id": 101, "datasource_type": "table", "viz_type": "table", "params": '{"metrics": ["count"]}', } }) # Mock DatasetHealthChecker with patch('src.plugins.llm_analysis.plugin.DatasetHealthChecker') as MockDHC: mock_dhc = MagicMock() MockDHC.return_value = mock_dhc mock_dhc.check_dashboard_datasets = AsyncMock(return_value={ "datasets": [{"dataset_id": 101, "metadata_accessible": True}], "chart_data": [], }) # Mock LLMClient with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.get_json_completion = AsyncMock(return_value={ "status": "PASS", "summary": "Text analysis OK", "issues": [], }) # Mock _fetch_dashboard_logs with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log entry"], {"logs_fetch_duration_ms": 10} ))): with patch('src.plugins.llm_analysis.plugin.build_dashboard_topology', return_value="Tab1 > Chart1"): with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: MockNotif.return_value.dispatch_report = AsyncMock() result = await plugin._execute_path_b( params=params, context=None, db=mock_db, db_provider=mock_provider, api_key="sk-real-key", env=mock_env, config_mgr=mock_cm, log=mock_log, ) assert result["execution_path"] == "text_only" assert result["status"] == "PASS" mock_db.add.assert_called_once() mock_db.commit.assert_called() class TestFetchDashboardLogs: """Verify _execute_path_a — screenshot + multimodal LLM flow.""" @pytest.mark.asyncio async def test_full_flow(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.storage.root_path = "/tmp/storage" mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = { "dashboard_id": "42", "environment_id": "env-1", "run_id": "run-42", } # Mock ScreenshotService with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS: mock_ss = MagicMock() MockSS.return_value = mock_ss mock_ss.capture_dashboard = AsyncMock(return_value=(["/tmp/jpeg.jpg"], [{"webp_path": "/tmp/archive.webp"}])) # Mock LLMClient with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.analyze_dashboard_multimodal = AsyncMock(return_value={ "status": "PASS", "summary": "All good", "issues": [], "chunk_count": 1, }) # Mock _fetch_dashboard_logs with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log entry"], {"logs_fetch_duration_ms": 10} ))): # Mock cleanup with patch('src.plugins.llm_analysis.plugin.ScreenshotService._cleanup_temp_files'): # Mock NotificationService with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: mock_notif = MagicMock() MockNotif.return_value = mock_notif mock_notif.dispatch_report = AsyncMock() result = await plugin._execute_path_a( params=params, context=None, db=mock_db, db_provider=mock_provider, api_key="sk-real-key", env=mock_env, config_mgr=mock_cm, log=mock_log, ) assert result["execution_path"] == "multimodal" assert result["status"] == "PASS" assert "timings" in result mock_db.add.assert_called_once() mock_db.commit.assert_called() @pytest.mark.asyncio async def test_no_screenshots_falls_back_to_text(self): """When capture_dashboard returns no screenshots, fallback to text-only.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.storage.root_path = "/tmp/storage" mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = {"dashboard_id": "42", "environment_id": "env-1"} with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS: mock_ss = MagicMock() MockSS.return_value = mock_ss # No screenshots captured mock_ss.capture_dashboard = AsyncMock(return_value=([], [])) with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.analyze_dashboard_text_batch = AsyncMock(return_value={ "dashboards": [{"status": "WARN", "summary": "No screenshots", "issues": []}] }) with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log"], {"logs_fetch_duration_ms": 5} ))): with patch('src.plugins.llm_analysis.plugin.ScreenshotService._cleanup_temp_files'): with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: MockNotif.return_value.dispatch_report = AsyncMock() result = await plugin._execute_path_a( params, None, mock_db, mock_provider, "sk-key", mock_env, mock_cm, mock_log, ) assert result["status"] == "WARN" @pytest.mark.asyncio async def test_notification_failure_does_not_block(self): """Edge: notification failure is logged but does not block return.""" from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.storage.root_path = "/tmp/storage" mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = {"dashboard_id": "42", "environment_id": "env-1"} with patch('src.plugins.llm_analysis.plugin.ScreenshotService') as MockSS: MockSS.return_value.capture_dashboard = AsyncMock(return_value=(["/tmp/jpg"], [{"webp_path": "/tmp/webp"}])) with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: MockLLM.return_value.analyze_dashboard_multimodal = AsyncMock(return_value={ "status": "PASS", "summary": "OK", "issues": [], "chunk_count": 1, }) with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log"], {"logs_fetch_duration_ms": 5} ))): with patch('src.plugins.llm_analysis.plugin.ScreenshotService._cleanup_temp_files'): with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: MockNotif.return_value.dispatch_report = AsyncMock(side_effect=RuntimeError("Notif failed")) result = await plugin._execute_path_a( params, None, mock_db, mock_provider, "sk-key", mock_env, mock_cm, mock_log, ) assert result["status"] == "PASS" # notification failure is non-blocking class TestExecutePathB: """Verify _execute_path_b — text-only LLM flow.""" @pytest.mark.asyncio async def test_full_flow(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_db = MagicMock() mock_db.add.return_value = None mock_db.commit.return_value = None mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.name = "Test" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.max_images = 10 mock_cm = MagicMock() mock_cfg = MagicMock() mock_cfg.settings.llm = {"prompts": {"dashboard_validation_prompt_text": "Analyze: {topology}"}} mock_cm.get_config.return_value = mock_cfg mock_log = MagicMock() mock_log.with_source.return_value = mock_log params = {"dashboard_id": "42", "environment_id": "env-1", "run_id": "run-99"} # Mock SupersetClient with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC: mock_sc = MagicMock() MockSC.return_value = mock_sc mock_sc.get_dashboard = AsyncMock(return_value={ "result": { "slices": [{"id": 1, "slice_id": 1}], } }) mock_sc.get_chart = AsyncMock(return_value={ "result": { "id": 1, "slice_name": "Chart 1", "datasource_id": 101, "datasource_type": "table", "viz_type": "table", "params": '{"metrics": ["count"]}', } }) # Mock DatasetHealthChecker with patch('src.plugins.llm_analysis.plugin.DatasetHealthChecker') as MockDHC: mock_dhc = MagicMock() MockDHC.return_value = mock_dhc mock_dhc.check_dashboard_datasets = AsyncMock(return_value={ "datasets": [{"dataset_id": 101, "metadata_accessible": True}], "chart_data": [], }) # Mock LLMClient with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.get_json_completion = AsyncMock(return_value={ "status": "PASS", "summary": "Text analysis OK", "issues": [], }) # Mock _fetch_dashboard_logs with patch.object(plugin, '_fetch_dashboard_logs', new=AsyncMock(return_value=( ["log entry"], {"logs_fetch_duration_ms": 10} ))): with patch('src.plugins.llm_analysis.plugin.build_dashboard_topology', return_value="Tab1 > Chart1"): with patch('src.plugins.llm_analysis.plugin.NotificationService') as MockNotif: MockNotif.return_value.dispatch_report = AsyncMock() result = await plugin._execute_path_b( params=params, context=None, db=mock_db, db_provider=mock_provider, api_key="sk-real-key", env=mock_env, config_mgr=mock_cm, log=mock_log, ) assert result["execution_path"] == "text_only" assert result["status"] == "PASS" mock_db.add.assert_called_once() mock_db.commit.assert_called() class TestFetchDashboardLogs: """Verify _fetch_dashboard_logs.""" @pytest.mark.asyncio async def test_success(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_log = MagicMock() mock_log.with_source.return_value = mock_log with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC: mock_sc = MagicMock() MockSC.return_value = mock_sc mock_sc.network.request = AsyncMock(return_value={ "result": [ {"action": "view", "dttm": "2024-01-01T00:00:00", "json": "{}"}, {"action": "save", "dttm": "2024-01-01T01:00:00", "json": '{"foo": "bar"}'}, ] }) logs, timings = await plugin._fetch_dashboard_logs(mock_env, "42", mock_log) assert len(logs) == 2 assert "[2024-01-01T00:00:00] view: {}" in logs assert "logs_fetch_duration_ms" in timings @pytest.mark.asyncio async def test_no_logs_found(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_log = MagicMock() mock_log.with_source.return_value = mock_log with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC: mock_sc = MagicMock() MockSC.return_value = mock_sc mock_sc.network.request = AsyncMock(return_value={"result": []}) logs, timings = await plugin._fetch_dashboard_logs(mock_env, "42", mock_log) assert len(logs) == 1 assert "No recent logs" in logs[0] @pytest.mark.asyncio async def test_network_error(self): from src.plugins.llm_analysis.plugin import DashboardValidationPlugin plugin = DashboardValidationPlugin() mock_env = MagicMock() mock_env.id = "env-1" mock_env.url = "https://superset.example.com" mock_log = MagicMock() mock_log.with_source.return_value = mock_log with patch('src.plugins.llm_analysis.plugin.SupersetClient') as MockSC: mock_sc = MagicMock() MockSC.return_value = mock_sc mock_sc.network.request.side_effect = ConnectionError("API unreachable") logs, timings = await plugin._fetch_dashboard_logs(mock_env, "42", mock_log) assert len(logs) == 1 assert "Error fetching" in logs[0] # ── DocumentationPlugin Tests ── class TestDocumentationPluginProperties: """Verify DocumentationPlugin static properties.""" def test_properties(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() assert plugin.id == "llm_documentation" assert plugin.name == "Dataset LLM Documentation" assert plugin.description == "Automated dataset and column documentation using LLMs." assert plugin.version == "1.0.0" def test_get_schema(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() schema = plugin.get_schema() assert schema["type"] == "object" assert "dataset_id" in schema["properties"] assert "environment_id" in schema["properties"] assert "provider_id" in schema["properties"] assert "dataset_id" in schema["required"] class TestDocumentationPluginExecute: """Verify DocumentationPlugin.execute.""" @pytest.mark.asyncio async def test_full_flow(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() mock_db = MagicMock() mock_db.close.return_value = None with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_env = MagicMock() mock_env.id = "env-1" mock_env.name = "Test Env" mock_cm.get_environment.return_value = mock_env with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: mock_svc = MagicMock() MockProviderService.return_value = mock_svc mock_provider = MagicMock() mock_provider.id = "prov-1" mock_provider.provider_type = "openai" mock_provider.base_url = "https://api.openai.com" mock_provider.default_model = "gpt-4o" mock_provider.is_multimodal = True mock_svc.get_provider.return_value = mock_provider mock_svc.get_decrypted_api_key.return_value = "sk-real-key-that-is-long-enough" with patch('src.core.superset_client.SupersetClient') as MockSC: mock_sc = MagicMock() MockSC.return_value = mock_sc mock_sc.get_dataset = AsyncMock(return_value={ "columns": [ {"id": 1, "column_name": "id", "type": "INTEGER", "description": None}, {"id": 2, "column_name": "name", "type": "VARCHAR", "description": None}, ], "table_name": "users", }) mock_sc.update_dataset = AsyncMock(return_value={"result": "ok"}) with patch('src.plugins.llm_analysis.plugin.LLMClient') as MockLLM: mock_llm = MagicMock() MockLLM.return_value = mock_llm mock_llm.get_json_completion = AsyncMock(return_value={ "dataset_description": "Users table with IDs and names", "column_descriptions": [ {"name": "id", "description": "Primary key"}, {"name": "name", "description": "User display name"}, ], }) mock_cfg = MagicMock() mock_cfg.settings.llm = {"prompts": {"documentation_prompt": "Doc: {dataset_name} {columns_json}"}} mock_cm.get_config.return_value = mock_cfg result = await plugin.execute({ "dataset_id": "1", "environment_id": "env-1", "provider_id": "prov-1", }) assert "dataset_description" in result assert result["dataset_description"] == "Users table with IDs and names" mock_sc.update_dataset.assert_called_once() @pytest.mark.asyncio async def test_env_not_found(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() mock_db = MagicMock() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = None with pytest.raises(ValueError, match="env-1 not found"): await plugin.execute({ "dataset_id": "1", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_provider_not_found(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() mock_db = MagicMock() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = None with pytest.raises(ValueError, match="prov-1 not found"): await plugin.execute({ "dataset_id": "1", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_invalid_api_key(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() mock_db = MagicMock() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: MockProviderService.return_value.get_provider.return_value = MagicMock() MockProviderService.return_value.get_decrypted_api_key.return_value = "********" with pytest.raises(ValueError, match="Invalid API key"): await plugin.execute({ "dataset_id": "1", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_provider_not_found(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() mock_db = MagicMock() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: mock_svc = MagicMock() MockProviderService.return_value = mock_svc mock_svc.get_provider.return_value = None with pytest.raises(ValueError, match="prov-1 not found"): await plugin.execute({ "dataset_id": "1", "environment_id": "env-1", "provider_id": "prov-1", }) @pytest.mark.asyncio async def test_invalid_api_key(self): from src.plugins.llm_analysis.plugin import DocumentationPlugin plugin = DocumentationPlugin() mock_db = MagicMock() with patch('src.plugins.llm_analysis.plugin.SessionLocal', return_value=mock_db): with patch('src.dependencies.get_config_manager') as mock_get_cm: mock_cm = MagicMock() mock_get_cm.return_value = mock_cm mock_cm.get_environment.return_value = MagicMock() with patch('src.plugins.llm_analysis.plugin.LLMProviderService') as MockProviderService: mock_svc = MagicMock() MockProviderService.return_value = mock_svc mock_svc.get_provider.return_value = MagicMock() mock_svc.get_decrypted_api_key.return_value = "********" with pytest.raises(ValueError, match="Invalid API key"): await plugin.execute({ "dataset_id": "1", "environment_id": "env-1", "provider_id": "prov-1", }) # #endregion Test.LLMAnalysisPlugin