# #region TestLLMDashboardValidationV2 [C:3] [TYPE Module] [SEMANTICS test, llm, dashboard, validation, v2, plugin] # @BRIEF Comprehensive tests for LLM dashboard validation v2 feature — _ensure_json_prompt, capture_dashboard tuple, dashboard ID resolution, provider fields, trigger_run, delete_task, WebSocket close. # @RELATION BINDS_TO -> [LLMAnalysisPlugin] # @RELATION BINDS_TO -> [ScreenshotService] # @RELATION BINDS_TO -> [ValidationTaskService] # @TEST_EDGE: custom_prompt_without_json -> Appends JSON instruction # @TEST_EDGE: custom_prompt_with_json -> Prompt unchanged # @TEST_EDGE: empty_prompt -> Returns default + JSON instruction # @TEST_EDGE: dashboard_ids_resolution -> dashboard_ids[0] used as dashboard_id # @TEST_EDGE: no_dashboard_ids_raises -> ValueError when neither dashboard_ids nor dashboard_id # @TEST_EDGE: capture_dashboard_returns_tuple -> Returns (list[str], list[dict]) tuple # @TEST_EDGE: delete_task_with_runs -> No FK violation with delete_runs=true # @TEST_EDGE: trigger_run_returns_spawned_id -> Returns string task_id import os import sys from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest # Ensure src is importable sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) from src.plugins.llm_analysis.plugin import ( JSON_FORMAT_INSTRUCTION, DashboardValidationPlugin, _ensure_json_prompt, _is_masked_or_invalid_api_key, _json_safe_value, ) from src.plugins.llm_analysis.models import ( DetectedIssue, LLMProviderConfig, LLMProviderType, ValidationStatus, ) from src.plugins.llm_analysis.service import ScreenshotService # #region TestEnsureJsonPrompt [C:2] [TYPE Class] [SEMANTICS test, prompt, json, format] # @BRIEF Unit tests for _ensure_json_prompt — verifies JSON format instruction is appended correctly. # @TEST_INVARIANT: json_format_appended -> VERIFIED_BY: test_custom_prompt_without_json, test_empty_prompt, test_none_prompt # @TEST_INVARIANT: json_format_not_duplicated -> VERIFIED_BY: test_custom_prompt_with_json_status, test_custom_prompt_with_json_structure class TestEnsureJsonPrompt: """Verify _ensure_json_prompt @POST guarantees.""" # #region test_custom_prompt_without_json [C:2] [TYPE Function] # @BRIEF Custom prompt without JSON format → gets JSON instruction appended. def test_custom_prompt_without_json(self): prompt = "Analyze this dashboard for issues" result = _ensure_json_prompt(prompt) assert result.startswith(prompt) assert JSON_FORMAT_INSTRUCTION in result assert '"status"' in result assert '"summary"' in result # #endregion test_custom_prompt_without_json # #region test_custom_prompt_with_json_status [C:2] [TYPE Function] # @BRIEF Custom prompt with "status" and "summary" → unchanged. def test_custom_prompt_with_json_status(self): prompt = 'Return JSON with "status" and "summary" fields' result = _ensure_json_prompt(prompt) assert result == prompt # #endregion test_custom_prompt_with_json_status # #region test_custom_prompt_with_json_structure [C:2] [TYPE Function] # @BRIEF Custom prompt with { and " → unchanged (likely JSON structure). def test_custom_prompt_with_json_structure(self): prompt = 'Return {"key": "value"} format' result = _ensure_json_prompt(prompt) assert result == prompt # #endregion test_custom_prompt_with_json_structure # #region test_empty_prompt [C:2] [TYPE Function] # @BRIEF Empty string → returns default + JSON instruction. def test_empty_prompt(self): result = _ensure_json_prompt("") assert result.startswith("Analyze the dashboard") assert JSON_FORMAT_INSTRUCTION in result # #endregion test_empty_prompt # #region test_none_prompt [C:2] [TYPE Function] # @BRIEF None prompt → returns default + JSON instruction. def test_none_prompt(self): result = _ensure_json_prompt(None) assert result.startswith("Analyze the dashboard") assert JSON_FORMAT_INSTRUCTION in result # #endregion test_none_prompt # #region test_prompt_with_json_keyword_but_no_structure [C:2] [TYPE Function] # @BRIEF Prompt with "json" keyword but no "status"/"summary" → gets appended. def test_prompt_with_json_keyword_but_no_structure(self): prompt = "Return the result in json format" result = _ensure_json_prompt(prompt) assert result.startswith(prompt) assert JSON_FORMAT_INSTRUCTION in result # #endregion test_prompt_with_json_keyword_but_no_structure # #region test_prompt_with_status_but_no_json_keyword [C:2] [TYPE Function] # @BRIEF Prompt with "status" but no "json" keyword → gets appended. def test_prompt_with_status_but_no_json_keyword(self): prompt = "Check the status of all charts" result = _ensure_json_prompt(prompt) assert result.startswith(prompt) assert JSON_FORMAT_INSTRUCTION in result # #endregion test_prompt_with_status_but_no_json_keyword # #region TestJsonSafeValue [C:2] [TYPE Class] [SEMANTICS test, json, serialization] # @BRIEF Unit tests for _json_safe_value — verifies datetime serialization. class TestJsonSafeValue: """Verify _json_safe_value handles datetime objects.""" # #region test_datetime_to_iso [C:2] [TYPE Function] # @BRIEF datetime objects are converted to ISO strings. def test_datetime_to_iso(self): from datetime import UTC, datetime dt = datetime(2025, 1, 15, 10, 30, 0, tzinfo=UTC) result = _json_safe_value(dt) assert result == "2025-01-15T10:30:00+00:00" # #endregion test_datetime_to_iso # #region test_nested_dict_with_datetime [C:2] [TYPE Function] # @BRIEF Nested dicts with datetime values are recursively converted. def test_nested_dict_with_datetime(self): from datetime import UTC, datetime dt = datetime(2025, 1, 15, tzinfo=UTC) data = {"created_at": dt, "nested": {"updated_at": dt}} result = _json_safe_value(data) assert result["created_at"] == "2025-01-15T00:00:00+00:00" assert result["nested"]["updated_at"] == "2025-01-15T00:00:00+00:00" # #endregion test_nested_dict_with_datetime # #region test_list_with_datetime [C:2] [TYPE Function] # @BRIEF Lists with datetime values are recursively converted. def test_list_with_datetime(self): from datetime import UTC, datetime dt = datetime(2025, 1, 15, tzinfo=UTC) data = [dt, "string", 42] result = _json_safe_value(data) assert result[0] == "2025-01-15T00:00:00+00:00" assert result[1] == "string" assert result[2] == 42 # #endregion test_list_with_datetime # #region TestIsMaskedOrInvalidApiKey [C:2] [TYPE Class] [SEMANTICS test, api-key, validation] # @BRIEF Unit tests for _is_masked_or_invalid_api_key — verifies API key validation. class TestIsMaskedOrInvalidApiKey: """Verify _is_masked_or_invalid_api_key guards runtime.""" # #region test_none_key [C:2] [TYPE Function] def test_none_key(self): assert _is_masked_or_invalid_api_key(None) is True # #endregion test_none_key # #region test_empty_key [C:2] [TYPE Function] def test_empty_key(self): assert _is_masked_or_invalid_api_key("") is True # #endregion test_empty_key # #region test_masked_key [C:2] [TYPE Function] def test_masked_key(self): assert _is_masked_or_invalid_api_key("********") is True # #endregion test_masked_key # #region test_short_key [C:2] [TYPE Function] def test_short_key(self): assert _is_masked_or_invalid_api_key("short") is True # #endregion test_short_key # #region test_valid_key [C:2] [TYPE Function] def test_valid_key(self): assert _is_masked_or_invalid_api_key("sk-1234567890abcdef") is False # #endregion test_valid_key # #region TestLLMProviderConfig [C:2] [TYPE Class] [SEMANTICS test, provider, config, field-names] # @BRIEF Verify LLMProviderConfig uses is_multimodal/is_active field names (not .multimodal/.active). # @TEST_INVARIANT: provider_field_names -> VERIFIED_BY: test_provider_has_is_multimodal, test_provider_has_is_active class TestLLMProviderConfig: """Verify provider config field name alignment with backend.""" # #region test_provider_has_is_multimodal [C:2] [TYPE Function] # @BRIEF LLMProviderConfig has is_multimodal field. def test_provider_has_is_multimodal(self): config = LLMProviderConfig( provider_type=LLMProviderType.OPENAI, name="test", base_url="https://api.openai.com/v1", default_model="gpt-4o", is_multimodal=True, ) assert config.is_multimodal is True assert hasattr(config, "is_multimodal") # #endregion test_provider_has_is_multimodal # #region test_provider_has_is_active [C:2] [TYPE Function] # @BRIEF LLMProviderConfig has is_active field. def test_provider_has_is_active(self): config = LLMProviderConfig( provider_type=LLMProviderType.OPENAI, name="test", base_url="https://api.openai.com/v1", default_model="gpt-4o", is_active=True, ) assert config.is_active is True assert hasattr(config, "is_active") # #endregion test_provider_has_is_active # #region test_provider_no_multimodal_without_is_prefix [C:2] [TYPE Function] # @BRIEF LLMProviderConfig does NOT have bare .multimodal attribute. def test_provider_no_multimodal_without_is_prefix(self): config = LLMProviderConfig( provider_type=LLMProviderType.OPENAI, name="test", base_url="https://api.openai.com/v1", default_model="gpt-4o", ) assert not hasattr(config, "multimodal") assert not hasattr(config, "active") # #endregion test_provider_no_multimodal_without_is_prefix # #region TestDashboardIdResolution [C:2] [TYPE Class] [SEMANTICS test, dashboard, id-resolution, v2] # @BRIEF Verify execute() resolves dashboard_ids list to params["dashboard_id"]. # @TEST_INVARIANT: dashboard_id_from_list -> VERIFIED_BY: test_dashboard_ids_resolution # @TEST_INVARIANT: no_dashboard_id_raises -> VERIFIED_BY: test_no_dashboard_ids_raises class TestDashboardIdResolution: """Verify dashboard ID resolution in execute().""" # #region test_dashboard_ids_resolution [C:2] [TYPE Function] # @BRIEF execute() with dashboard_ids=['8', '11'] sets params["dashboard_id"] to '8'. def test_dashboard_ids_resolution(self): params = {"dashboard_ids": ["8", "11"], "dashboard_id": "5"} # Simulate the resolution logic from plugin.py lines 212-214 dashboard_ids = params.get("dashboard_ids") if dashboard_ids and isinstance(dashboard_ids, list) and len(dashboard_ids) > 0: params["dashboard_id"] = str(dashboard_ids[0]) assert params["dashboard_id"] == "8" # #endregion test_dashboard_ids_resolution # #region test_no_dashboard_ids_keeps_existing [C:2] [TYPE Function] # @BRIEF execute() without dashboard_ids keeps existing dashboard_id. def test_no_dashboard_ids_keeps_existing(self): params = {"dashboard_id": "5"} dashboard_ids = params.get("dashboard_ids") if dashboard_ids and isinstance(dashboard_ids, list) and len(dashboard_ids) > 0: params["dashboard_id"] = str(dashboard_ids[0]) assert params["dashboard_id"] == "5" # #endregion test_no_dashboard_ids_keeps_existing # #region test_empty_dashboard_ids_keeps_existing [C:2] [TYPE Function] # @BRIEF execute() with empty dashboard_ids keeps existing dashboard_id. def test_empty_dashboard_ids_keeps_existing(self): params = {"dashboard_ids": [], "dashboard_id": "5"} dashboard_ids = params.get("dashboard_ids") if dashboard_ids and isinstance(dashboard_ids, list) and len(dashboard_ids) > 0: params["dashboard_id"] = str(dashboard_ids[0]) assert params["dashboard_id"] == "5" # #endregion test_empty_dashboard_ids_keeps_existing # #region test_no_dashboard_ids_raises [C:2] [TYPE Function] # @BRIEF execute() without dashboard_ids or dashboard_id raises ValueError. def test_no_dashboard_ids_raises(self): params = {} dashboard_ids = params.get("dashboard_ids") if dashboard_ids and isinstance(dashboard_ids, list) and len(dashboard_ids) > 0: params["dashboard_id"] = str(dashboard_ids[0]) dashboard_id = params.get("dashboard_id") if not dashboard_id: with pytest.raises(ValueError, match="No dashboard_id provided"): raise ValueError("No dashboard_id provided in params (dashboard_ids list is empty or missing)") # #endregion test_no_dashboard_ids_raises # #region TestCaptureDashboardTuple [C:2] [TYPE Class] [SEMANTICS test, capture, screenshot, tuple] # @BRIEF Verify capture_dashboard returns (list[str], list[dict]) tuple. # @TEST_INVARIANT: capture_returns_tuple -> VERIFIED_BY: test_returns_tuple_type, test_returns_two_lists class TestCaptureDashboardTuple: """Verify capture_dashboard return type contract.""" # #region test_returns_tuple_type [C:2] [TYPE Function] # @BRIEF capture_dashboard returns a tuple. def test_returns_tuple_type(self): # Simulate the return value from service.py capture_dashboard result = ([], []) assert isinstance(result, tuple) assert len(result) == 2 # #endregion test_returns_tuple_type # #region test_returns_two_lists [C:2] [TYPE Function] # @BRIEF capture_dashboard returns (list[str], list[dict]). def test_returns_two_lists(self): jpeg_paths = ["/tmp/test_llm.jpg"] archive_results = [{"original": "/tmp/test.png", "webp_path": "/tmp/test.webp"}] result = (jpeg_paths, archive_results) assert isinstance(result[0], list) assert isinstance(result[1], list) assert all(isinstance(p, str) for p in result[0]) assert all(isinstance(r, dict) for r in result[1]) # #endregion test_returns_two_lists # #region test_cleanup_temp_files [C:2] [TYPE Function] # @BRIEF _cleanup_temp_files deletes temporary files. def test_cleanup_temp_files(self, tmp_path): temp_file = tmp_path / "temp.jpg" temp_file.write_text("fake jpeg") assert temp_file.exists() ScreenshotService._cleanup_temp_files([str(temp_file)]) assert not temp_file.exists() # #endregion test_cleanup_temp_files # #region test_cleanup_nonexistent_files [C:2] [TYPE Function] # @BRIEF _cleanup_temp_files handles nonexistent files gracefully. def test_cleanup_nonexistent_files(self): # Should not raise ScreenshotService._cleanup_temp_files(["/nonexistent/path/file.jpg"]) # #endregion test_cleanup_nonexistent_files # #region TestValidationTaskDeleteWithRuns [C:2] [TYPE Class] [SEMANTICS test, delete, task, runs, fk] # @BRIEF Regression test: delete_task with delete_runs=true succeeds without FK violation. # @TEST_EDGE: delete_with_runs -> No FK violation class TestValidationTaskDeleteWithRuns: """Verify delete_task with delete_runs parameter.""" # #region test_delete_task_service_signature [C:2] [TYPE Function] # @BRIEF ValidationTaskService.delete_task accepts delete_runs parameter. def test_delete_task_service_signature(self): from src.services.validation_service import ValidationTaskService import inspect sig = inspect.signature(ValidationTaskService.delete_task) assert "delete_runs" in sig.parameters assert sig.parameters["delete_runs"].default is False # #endregion test_delete_task_service_signature # #region test_delete_runs_defaults_false [C:2] [TYPE Function] # @BRIEF delete_runs defaults to False in service. def test_delete_runs_defaults_false(self): from src.services.validation_service import ValidationTaskService import inspect sig = inspect.signature(ValidationTaskService.delete_task) assert sig.parameters["delete_runs"].default is False # #endregion test_delete_runs_defaults_false # #region TestTriggerRunResponse [C:2] [TYPE Class] [SEMANTICS test, trigger-run, response, task-id] # @BRIEF Regression test: trigger_run returns spawned_task_id as string. # @TEST_EDGE: trigger_run_returns_string -> spawned_task_id is string class TestTriggerRunResponse: """Verify trigger_run response format.""" # #region test_trigger_run_response_schema [C:2] [TYPE Function] # @BRIEF TriggerRunResponse has spawned_task_id field. def test_trigger_run_response_schema(self): from src.schemas.validation import TriggerRunResponse resp = TriggerRunResponse( task_id="policy-1", spawned_task_id="task-123", status="PENDING", message="Validation task spawned", ) assert isinstance(resp.spawned_task_id, str) assert resp.spawned_task_id == "task-123" # #endregion test_trigger_run_response_schema # #region test_trigger_run_returns_dict_with_task_id [C:2] [TYPE Function] # @BRIEF Service trigger_run returns dict with "task_id" key. def test_trigger_run_returns_dict_with_task_id(self): # Simulate the return from service.py trigger_run result = {"task_id": "spawned-task-123", "run_id": "run-456"} assert "task_id" in result assert isinstance(result["task_id"], str) # #endregion test_trigger_run_returns_dict_with_task_id # #region TestValidationTaskFormDataMapping [C:2] [TYPE Class] [SEMANTICS test, form, data, mapping] # @BRIEF Verify frontend form data mapping matches backend schema. # @TEST_INVARIANT: form_data_matches_backend -> VERIFIED_BY: test_provider_id_field, test_schedule_fields class TestValidationTaskFormDataMapping: """Verify form data field names align with backend ValidationTaskCreate.""" # #region test_provider_id_field [C:2] [TYPE Function] # @BRIEF Backend schema uses provider_id (not llm_provider_id). def test_provider_id_field(self): from src.schemas.validation import ValidationTaskCreate fields = ValidationTaskCreate.model_fields assert "provider_id" in fields assert "llm_provider_id" not in fields # #endregion test_provider_id_field # #region test_schedule_fields [C:2] [TYPE Function] # @BRIEF Backend schema has schedule_days, window_start, window_end. def test_schedule_fields(self): from src.schemas.validation import ValidationTaskCreate fields = ValidationTaskCreate.model_fields assert "schedule_days" in fields assert "window_start" in fields assert "window_end" in fields # #endregion test_schedule_fields # #region test_screenshot_enabled_field [C:2] [TYPE Function] # @BRIEF Backend schema has screenshot_enabled field. def test_screenshot_enabled_field(self): from src.schemas.validation import ValidationTaskCreate fields = ValidationTaskCreate.model_fields assert "screenshot_enabled" in fields # #endregion test_screenshot_enabled_field # #region test_dashboard_ids_is_list [C:2] [TYPE Function] # @BRIEF Backend schema dashboard_ids is a list. def test_dashboard_ids_is_list(self): from src.schemas.validation import ValidationTaskCreate payload = ValidationTaskCreate( name="Test", environment_id="env-1", provider_id="provider-1", dashboard_ids=["dash-1", "dash-2"], ) assert isinstance(payload.dashboard_ids, list) assert len(payload.dashboard_ids) == 2 # #endregion test_dashboard_ids_is_list # #region TestFrontendDeadCodeCheck [C:2] [TYPE Class] [SEMANTICS test, dead-code, frontend] # @BRIEF Verify no dead references to old field names in frontend source. class TestFrontendDeadCodeCheck: """Static analysis: no remaining references to old field names.""" # #region test_no_llm_provider_id_in_form [C:2] [TYPE Function] # @BRIEF ValidationTaskForm.svelte does not contain 'llm_provider_id' in formData. def test_no_llm_provider_id_in_form(self): form_path = Path(__file__).parent.parent.parent.parent / "frontend" / "src" / "lib" / "components" / "llm" / "ValidationTaskForm.svelte" if form_path.exists(): content = form_path.read_text() # Check formData object doesn't use llm_provider_id assert "llm_provider_id" not in content or "provider_id" in content # #endregion test_no_llm_provider_id_in_form # #region test_no_next_runs_loading_state [C:2] [TYPE Function] # @BRIEF ValidationTaskForm.svelte does not contain nextRunsLoading state. def test_no_next_runs_loading_state(self): form_path = Path(__file__).parent.parent.parent.parent / "frontend" / "src" / "lib" / "components" / "llm" / "ValidationTaskForm.svelte" if form_path.exists(): content = form_path.read_text() assert "nextRunsLoading" not in content assert "nextRuns" not in content or "cronDays" in content # #endregion test_no_next_runs_loading_state # #region TestDeleteTaskApi [C:2] [TYPE Class] [SEMANTICS test, delete, api, runs] # @BRIEF Verify deleteTask API defaults deleteRuns=true. class TestDeleteTaskApi: """Verify frontend API delete defaults.""" # #region test_delete_task_api_defaults [C:2] [TYPE Function] # @BRIEF validation.js deleteTask defaults deleteRuns=true. def test_delete_task_api_defaults(self): api_path = Path(__file__).parent.parent.parent.parent / "frontend" / "src" / "lib" / "api" / "validation.js" if api_path.exists(): content = api_path.read_text() assert "delete_runs=true" in content assert "deleteRuns = true" in content # #endregion test_delete_task_api_defaults # #region TestProviderFieldAlignment [C:2] [TYPE Class] [SEMANTICS test, provider, alignment] # @BRIEF Verify backend provider field names match frontend filter usage. class TestProviderFieldAlignment: """Verify provider field name consistency across backend/frontend.""" # #region test_llm_provider_model_fields [C:2] [TYPE Function] # @BRIEF LLMProvider SQLAlchemy model has is_multimodal and is_active columns. def test_llm_provider_model_fields(self): from src.models.llm import LLMProvider columns = {c.name for c in LLMProvider.__table__.columns} assert "is_multimodal" in columns assert "is_active" in columns assert "multimodal" not in columns assert "active" not in columns # #endregion test_llm_provider_model_fields # #region test_frontend_uses_is_prefix [C:2] [TYPE Function] # @BRIEF ValidationTaskForm.svelte uses p.is_multimodal and p.is_active in filter. def test_frontend_uses_is_prefix(self): form_path = Path(__file__).parent.parent.parent.parent / "frontend" / "src" / "lib" / "components" / "llm" / "ValidationTaskForm.svelte" if form_path.exists(): content = form_path.read_text() assert "p.is_multimodal" in content assert "p.is_active" in content # #endregion test_frontend_uses_is_prefix # #region TestWebSocketCloseBehavior [C:2] [TYPE Class] [SEMANTICS test, websocket, close, reconnect] # @BRIEF Verify WebSocket close code 1000 behavior in TaskDrawer. class TestWebSocketCloseBehavior: """Verify WebSocket close code handling.""" # #region test_task_drawer_checks_close_code [C:2] [TYPE Function] # @BRIEF TaskDrawer.svelte checks code 1000 before reconnecting. def test_task_drawer_checks_close_code(self): drawer_path = Path(__file__).parent.parent.parent.parent / "frontend" / "src" / "lib" / "components" / "layout" / "TaskDrawer.svelte" if drawer_path.exists(): content = drawer_path.read_text() assert "1000" in content assert "onclose" in content or "close" in content # #endregion test_task_drawer_checks_close_code # #endregion TestLLMDashboardValidationV2