busya
ce0369ae5c
Subagents delivered tests across all uncovered backend modules: Schemas (100%): agent, auth, health, profile, settings, validation Services (98-100%): auth, profile, health, llm, mapping, resource, security, git, superset_lookup, sql_table_extractor, rbac API routes (new): auth, admin, health, environments, plugins, dashboards (helpers, projection, actions, listing), git (config, deps, env, helpers) Clean Release (100%): DTO, facade, policy_engine, stages, repos, preparation, source_isolation, compliance Git services: base, remote_providers Agent module: app, run, middleware, langgraph_setup Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching Reports: normalizer, report_service, type_profiles Notifications: service, providers Also: - .gitignore: add .coverage, *.cover, coverage-* dirs - src/schemas/auth.py: fix AD group DN regex (comma in CN=...) - Remove co-located src/services/__tests__/ (caused pytest module collision)
480 lines
17 KiB
Python
480 lines
17 KiB
Python
# #region Test.Schemas.Validation [C:2] [TYPE Module] [SEMANTICS test,schema,validation,task,run]
|
|
# @BRIEF Tests for schemas/validation.py — SourceInput, SourceResponse, ValidationTaskCreate,
|
|
# ValidationTaskUpdate, ValidationTaskResponse, ValidationTaskListResponse,
|
|
# TriggerRunResponse, RunResponse, RunListResponse, RecordResponse,
|
|
# RecordListResponse, RunDetailResponse.
|
|
# @RELATION BINDS_TO -> [ValidationSchemas]
|
|
# @TEST_EDGE: missing_field -> optional fields default correctly
|
|
# @TEST_EDGE: invalid_type -> Literal constraints reject bad values
|
|
# @TEST_EDGE: serialization -> round-trip via model_dump/model_validate
|
|
|
|
from datetime import datetime, timezone
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from src.schemas.validation import (
|
|
RecordListResponse,
|
|
RecordResponse,
|
|
RunDetailResponse,
|
|
RunListResponse,
|
|
RunResponse,
|
|
SourceInput,
|
|
SourceResponse,
|
|
TriggerRunResponse,
|
|
ValidationTaskCreate,
|
|
ValidationTaskListResponse,
|
|
ValidationTaskResponse,
|
|
ValidationTaskUpdate,
|
|
)
|
|
|
|
|
|
class TestSourceInput:
|
|
"""SourceInput — Literal type and required value."""
|
|
|
|
def test_dashboard_id(self):
|
|
si = SourceInput(type="dashboard_id", value="42")
|
|
assert si.type == "dashboard_id"
|
|
assert si.value == "42"
|
|
|
|
def test_dashboard_url(self):
|
|
si = SourceInput(type="dashboard_url", value="http://superset/dash/42")
|
|
assert si.type == "dashboard_url"
|
|
|
|
def test_invalid_type_raises(self):
|
|
with pytest.raises(ValidationError):
|
|
SourceInput(type="invalid", value="x")
|
|
|
|
def test_serialize_roundtrip(self):
|
|
si = SourceInput(type="dashboard_id", value="42")
|
|
data = si.model_dump()
|
|
restored = SourceInput.model_validate(data)
|
|
assert restored.type == "dashboard_id"
|
|
assert restored.value == "42"
|
|
|
|
|
|
class TestSourceResponse:
|
|
"""SourceResponse — validation source with metadata and from_attributes."""
|
|
|
|
def test_minimal(self):
|
|
sr = SourceResponse(id="s-1", type="dashboard_id", value="42")
|
|
assert sr.status == "valid"
|
|
assert sr.title is None
|
|
assert sr.resolved_dashboard_id is None
|
|
assert sr.last_error is None
|
|
assert sr.last_checked_at is None
|
|
assert sr.created_at is None
|
|
|
|
def test_with_errors(self):
|
|
sr = SourceResponse(id="s-1", type="dashboard_url", value="http://x", status="error", last_error="Invalid")
|
|
assert sr.status == "error"
|
|
assert sr.last_error == "Invalid"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
sr = SourceResponse(id="s-1", type="dashboard_id", value="42", title="Sales")
|
|
data = sr.model_dump()
|
|
restored = SourceResponse.model_validate(data)
|
|
assert restored.title == "Sales"
|
|
|
|
def test_from_attributes(self):
|
|
class FakeORM:
|
|
id = "s-1"
|
|
type = "dashboard_id"
|
|
value = "42"
|
|
status = "valid"
|
|
title = "Sales"
|
|
resolved_dashboard_id = "42"
|
|
last_error = None
|
|
last_checked_at = None
|
|
created_at = None
|
|
sr = SourceResponse.model_validate(FakeORM())
|
|
assert sr.id == "s-1"
|
|
assert sr.resolved_dashboard_id == "42"
|
|
|
|
|
|
class TestValidationTaskCreate:
|
|
"""ValidationTaskCreate — create task with v1/v2 fields."""
|
|
|
|
def test_minimal(self):
|
|
task = ValidationTaskCreate(
|
|
name="Test Task",
|
|
environment_id="env-1",
|
|
provider_id="prov-1",
|
|
)
|
|
assert task.description is None
|
|
assert task.dashboard_ids == []
|
|
assert task.is_active is True
|
|
assert task.screenshot_enabled is True
|
|
assert task.logs_enabled is True
|
|
assert task.execute_chart_data is False
|
|
assert task.llm_batch_size == 1
|
|
assert task.policy_dashboard_concurrency_limit == 3
|
|
assert task.sources is None
|
|
|
|
def test_with_all_fields(self):
|
|
sources = [SourceInput(type="dashboard_id", value="42")]
|
|
task = ValidationTaskCreate(
|
|
name="Full Task",
|
|
description="Full description",
|
|
environment_id="env-1",
|
|
dashboard_ids=["42", "99"],
|
|
provider_id="prov-1",
|
|
schedule_days=[0, 2, 4],
|
|
window_start="09:00",
|
|
window_end="18:00",
|
|
notify_owners=False,
|
|
custom_channels=["slack:#alerts"],
|
|
alert_condition="WARN_AND_FAIL",
|
|
is_active=False,
|
|
screenshot_enabled=False,
|
|
logs_enabled=False,
|
|
execute_chart_data=True,
|
|
llm_batch_size=5,
|
|
policy_dashboard_concurrency_limit=10,
|
|
prompt_template="Custom template",
|
|
sources=sources,
|
|
)
|
|
assert task.name == "Full Task"
|
|
assert task.schedule_days == [0, 2, 4]
|
|
assert task.execute_chart_data is True
|
|
assert task.llm_batch_size == 5
|
|
assert task.sources == sources
|
|
|
|
def test_llm_batch_size_ge_one(self):
|
|
with pytest.raises(ValidationError):
|
|
ValidationTaskCreate(name="T", environment_id="e1", provider_id="p1", llm_batch_size=0)
|
|
|
|
def test_concurrency_limit_ge_one(self):
|
|
with pytest.raises(ValidationError):
|
|
ValidationTaskCreate(name="T", environment_id="e1", provider_id="p1", policy_dashboard_concurrency_limit=0)
|
|
|
|
def test_serialize_roundtrip(self):
|
|
task = ValidationTaskCreate(name="T", environment_id="e1", provider_id="p1")
|
|
data = task.model_dump()
|
|
restored = ValidationTaskCreate.model_validate(data)
|
|
assert restored.name == "T"
|
|
assert restored.is_active is True
|
|
|
|
|
|
class TestValidationTaskUpdate:
|
|
"""ValidationTaskUpdate — all fields optional for partial update."""
|
|
|
|
def test_empty(self):
|
|
u = ValidationTaskUpdate()
|
|
assert u.name is None
|
|
assert u.description is None
|
|
assert u.environment_id is None
|
|
assert u.dashboard_ids is None
|
|
assert u.provider_id is None
|
|
assert u.schedule_days is None
|
|
assert u.is_active is None
|
|
assert u.sources is None
|
|
|
|
def test_partial(self):
|
|
u = ValidationTaskUpdate(name="Renamed", is_active=False)
|
|
assert u.name == "Renamed"
|
|
assert u.is_active is False
|
|
|
|
def test_serialize_roundtrip(self):
|
|
u = ValidationTaskUpdate(name="New", screenshot_enabled=False)
|
|
data = u.model_dump()
|
|
restored = ValidationTaskUpdate.model_validate(data)
|
|
assert restored.name == "New"
|
|
assert restored.screenshot_enabled is False
|
|
|
|
|
|
class TestValidationTaskResponse:
|
|
"""ValidationTaskResponse — full task response with from_attributes."""
|
|
|
|
def test_minimal(self):
|
|
ts = datetime.now(timezone.utc)
|
|
task = ValidationTaskResponse(
|
|
id="task-1",
|
|
name="My Task",
|
|
environment_id="env-1",
|
|
is_active=True,
|
|
dashboard_ids=["42"],
|
|
created_at=ts,
|
|
)
|
|
assert task.description is None
|
|
assert task.provider_id is None
|
|
assert task.last_run_status is None
|
|
assert task.screenshot_enabled is True
|
|
assert task.llm_batch_size == 1
|
|
assert task.sources == []
|
|
|
|
def test_with_all_fields(self):
|
|
ts = datetime.now(timezone.utc)
|
|
sources = [SourceResponse(id="s-1", type="dashboard_id", value="42")]
|
|
task = ValidationTaskResponse(
|
|
id="task-1", name="Full", environment_id="env-1", is_active=True,
|
|
dashboard_ids=["42"], provider_id="prov-1",
|
|
schedule_days=[1], window_start="09:00", window_end="18:00",
|
|
notify_owners=False, custom_channels=["email"], alert_condition="ALWAYS",
|
|
created_at=ts, updated_at=ts,
|
|
last_run_status="PASS", last_run_at=ts,
|
|
last_run_summary={"pass": 5, "fail": 0},
|
|
screenshot_enabled=False, logs_enabled=True, execute_chart_data=True,
|
|
llm_batch_size=3, policy_dashboard_concurrency_limit=5,
|
|
prompt_template="custom", sources=sources,
|
|
)
|
|
assert task.last_run_status == "PASS"
|
|
assert task.last_run_summary == {"pass": 5, "fail": 0}
|
|
assert len(task.sources) == 1
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
task = ValidationTaskResponse(id="t-1", name="T", environment_id="e1", is_active=True, dashboard_ids=[], created_at=ts)
|
|
data = task.model_dump()
|
|
restored = ValidationTaskResponse.model_validate(data)
|
|
assert restored.name == "T"
|
|
|
|
def test_from_attributes(self):
|
|
ts = datetime.now(timezone.utc)
|
|
class FakeORM:
|
|
id = "t-1"
|
|
name = "Task"
|
|
description = "Desc"
|
|
environment_id = "e1"
|
|
is_active = True
|
|
dashboard_ids = ["42"]
|
|
provider_id = "p1"
|
|
schedule_days = [1]
|
|
window_start = "09:00"
|
|
window_end = "18:00"
|
|
notify_owners = True
|
|
custom_channels = None
|
|
alert_condition = "FAIL_ONLY"
|
|
created_at = ts
|
|
updated_at = ts
|
|
last_run_status = "PASS"
|
|
last_run_at = ts
|
|
last_run_summary = None
|
|
screenshot_enabled = True
|
|
logs_enabled = True
|
|
execute_chart_data = False
|
|
llm_batch_size = 1
|
|
policy_dashboard_concurrency_limit = 3
|
|
prompt_template = None
|
|
sources = []
|
|
task = ValidationTaskResponse.model_validate(FakeORM())
|
|
assert task.name == "Task"
|
|
assert task.provider_id == "p1"
|
|
assert task.screenshot_enabled is True
|
|
|
|
|
|
class TestValidationTaskListResponse:
|
|
"""ValidationTaskListResponse — paginated list."""
|
|
|
|
def test_defaults(self):
|
|
resp = ValidationTaskListResponse()
|
|
assert resp.items == []
|
|
assert resp.total == 0
|
|
assert resp.page == 1
|
|
assert resp.page_size == 20
|
|
|
|
def test_with_items(self):
|
|
ts = datetime.now(timezone.utc)
|
|
items = [ValidationTaskResponse(id="t-1", name="T", environment_id="e1", is_active=True, dashboard_ids=[], created_at=ts)]
|
|
resp = ValidationTaskListResponse(items=items, total=1, page=1, page_size=10)
|
|
assert len(resp.items) == 1
|
|
assert resp.total == 1
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = ValidationTaskListResponse(total=5, page=2, page_size=10)
|
|
data = resp.model_dump()
|
|
restored = ValidationTaskListResponse.model_validate(data)
|
|
assert restored.total == 5
|
|
assert restored.page == 2
|
|
|
|
|
|
class TestTriggerRunResponse:
|
|
"""TriggerRunResponse — spawned task response."""
|
|
|
|
def test_minimal(self):
|
|
resp = TriggerRunResponse(task_id="t-1", spawned_task_id="st-1")
|
|
assert resp.run_id is None
|
|
assert resp.status == "PENDING"
|
|
assert resp.message == "Validation task spawned"
|
|
|
|
def test_with_run_id(self):
|
|
resp = TriggerRunResponse(task_id="t-1", spawned_task_id="st-1", run_id="run-1", status="RUNNING")
|
|
assert resp.run_id == "run-1"
|
|
assert resp.status == "RUNNING"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = TriggerRunResponse(task_id="t-1", spawned_task_id="st-1")
|
|
data = resp.model_dump()
|
|
restored = TriggerRunResponse.model_validate(data)
|
|
assert restored.task_id == "t-1"
|
|
assert restored.spawned_task_id == "st-1"
|
|
|
|
|
|
class TestRunResponse:
|
|
"""RunResponse — v2 ValidationRun display."""
|
|
|
|
def test_minimal(self):
|
|
resp = RunResponse(id="run-1", policy_id="pol-1", status="PENDING")
|
|
assert resp.task_id is None
|
|
assert resp.trigger == "manual"
|
|
assert resp.started_at is None
|
|
assert resp.finished_at is None
|
|
assert resp.dashboard_count == 0
|
|
assert resp.created_at is None
|
|
|
|
def test_with_all_fields(self):
|
|
ts = datetime.now(timezone.utc)
|
|
resp = RunResponse(
|
|
id="run-1", policy_id="pol-1", task_id="t-1", status="RUNNING",
|
|
trigger="scheduled", started_at=ts, finished_at=None,
|
|
dashboard_count=5, pass_count=3, warn_count=1, fail_count=1,
|
|
unknown_count=0, created_at=ts,
|
|
)
|
|
assert resp.trigger == "scheduled"
|
|
assert resp.pass_count == 3
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = RunResponse(id="r-1", policy_id="p-1", status="COMPLETED")
|
|
data = resp.model_dump()
|
|
restored = RunResponse.model_validate(data)
|
|
assert restored.status == "COMPLETED"
|
|
|
|
def test_from_attributes(self):
|
|
class FakeORM:
|
|
id = "r-1"
|
|
policy_id = "p-1"
|
|
task_id = "t-1"
|
|
status = "RUNNING"
|
|
trigger = "manual"
|
|
started_at = None
|
|
finished_at = None
|
|
dashboard_count = 10
|
|
pass_count = 8
|
|
warn_count = 1
|
|
fail_count = 1
|
|
unknown_count = 0
|
|
created_at = None
|
|
resp = RunResponse.model_validate(FakeORM())
|
|
assert resp.dashboard_count == 10
|
|
|
|
|
|
class TestRunListResponse:
|
|
"""RunListResponse — paginated run list."""
|
|
|
|
def test_defaults(self):
|
|
resp = RunListResponse()
|
|
assert resp.items == []
|
|
assert resp.total == 0
|
|
assert resp.page == 1
|
|
assert resp.page_size == 20
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = RunListResponse(total=3, page=1, page_size=10)
|
|
data = resp.model_dump()
|
|
restored = RunListResponse.model_validate(data)
|
|
assert restored.total == 3
|
|
|
|
|
|
class TestRecordResponse:
|
|
"""RecordResponse — validation record with lots of optional fields."""
|
|
|
|
def test_minimal(self):
|
|
ts = datetime.now(timezone.utc)
|
|
resp = RecordResponse(id="rec-1", dashboard_id="42", status="PASS", timestamp=ts)
|
|
assert resp.policy_id is None
|
|
assert resp.run_id is None
|
|
assert resp.dashboard_title is None
|
|
assert resp.summary == ""
|
|
assert resp.issues == []
|
|
assert resp.screenshot_paths == []
|
|
|
|
def test_with_all_fields(self):
|
|
ts = datetime.now(timezone.utc)
|
|
resp = RecordResponse(
|
|
id="rec-1", policy_id="pol-1", run_id="run-1", dashboard_id="42",
|
|
dashboard_title="Sales", environment_id="env-1", status="FAIL",
|
|
summary="Error", timestamp=ts, execution_path="screenshot",
|
|
screenshot_path="/screens/1.png", screenshot_paths=["/screens/1.png"],
|
|
issues=[{"severity": "FAIL", "message": "broken"}],
|
|
raw_response='{"ok": true}', dataset_health={"status": "OK"},
|
|
chart_data_results=[{"chart": "c1"}],
|
|
tab_screenshots=[{"tab": "tab1"}],
|
|
logs_sent_to_llm=["log1"], token_usage={"prompt": 100},
|
|
timings={"total": 5.0},
|
|
)
|
|
assert resp.dashboard_title == "Sales"
|
|
assert len(resp.issues) == 1
|
|
assert resp.timings == {"total": 5.0}
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
resp = RecordResponse(id="rec-1", dashboard_id="42", status="PASS", timestamp=ts)
|
|
data = resp.model_dump()
|
|
restored = RecordResponse.model_validate(data)
|
|
assert restored.dashboard_id == "42"
|
|
|
|
def test_from_attributes(self):
|
|
ts = datetime.now(timezone.utc)
|
|
class FakeORM:
|
|
id = "rec-1"
|
|
policy_id = "p-1"
|
|
run_id = "r-1"
|
|
dashboard_id = "42"
|
|
dashboard_title = "Sales"
|
|
environment_id = "e1"
|
|
status = "WARN"
|
|
summary = "Warning"
|
|
timestamp = ts
|
|
execution_path = None
|
|
screenshot_path = None
|
|
screenshot_paths = []
|
|
issues = []
|
|
raw_response = None
|
|
dataset_health = None
|
|
chart_data_results = []
|
|
tab_screenshots = []
|
|
logs_sent_to_llm = []
|
|
token_usage = None
|
|
timings = None
|
|
resp = RecordResponse.model_validate(FakeORM())
|
|
assert resp.dashboard_title == "Sales"
|
|
assert resp.status == "WARN"
|
|
|
|
|
|
class TestRecordListResponse:
|
|
"""RecordListResponse — paginated record list."""
|
|
|
|
def test_defaults(self):
|
|
resp = RecordListResponse()
|
|
assert resp.items == []
|
|
assert resp.total == 0
|
|
assert resp.page == 1
|
|
assert resp.page_size == 20
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = RecordListResponse(total=10, page=1, page_size=25)
|
|
data = resp.model_dump()
|
|
restored = RecordListResponse.model_validate(data)
|
|
assert restored.total == 10
|
|
|
|
|
|
class TestRunDetailResponse:
|
|
"""RunDetailResponse — run with records."""
|
|
|
|
def test_minimal(self):
|
|
run = RunResponse(id="r-1", policy_id="p-1", status="COMPLETED")
|
|
resp = RunDetailResponse(run=run)
|
|
assert resp.records == []
|
|
|
|
def test_with_records(self):
|
|
run = RunResponse(id="r-1", policy_id="p-1", status="COMPLETED")
|
|
resp = RunDetailResponse(run=run, records=[{"id": "rec-1"}, {"id": "rec-2"}])
|
|
assert len(resp.records) == 2
|
|
|
|
def test_serialize_roundtrip(self):
|
|
run = RunResponse(id="r-1", policy_id="p-1", status="COMPLETED")
|
|
resp = RunDetailResponse(run=run)
|
|
data = resp.model_dump()
|
|
restored = RunDetailResponse.model_validate(data)
|
|
assert restored.run.id == "r-1"
|
|
# #endregion Test.Schemas.Validation
|