busya
d2b53c2a79
Core implementation of the v2 LLM dashboard validation pipeline: - LLM plugin with Path A (screenshots) and Path B (logs-only) execution - Validation task management (CRUD, schedule, run) - WebSocket task progress with Python 3.13 asyncio fix - Cross-task runs listing (GET /validation-tasks/runs/all) - RecordResponse schema for validation records - JSON prompt helper, per-dashboard status aggregation - Prompt templates with docs/git-commit/validation presets - Migration: v2 validation models + description column - Tests: plugin persistence, prompt templates, batch, payload, url
514 lines
24 KiB
Python
514 lines
24 KiB
Python
# #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
|