241 lines
10 KiB
Python
241 lines
10 KiB
Python
# #region Test.Api.ValidationTasks.Edge [C:3] [TYPE Module] [SEMANTICS test,validation,tasks,edge,coverage]
|
|
# @BRIEF Edge-case tests for validation task API — covers uncovered branches from validation_tasks.py.
|
|
# @RELATION BINDS_TO -> [ValidationTasksRoutes]
|
|
# @TEST_EDGE: parse_url_unexpected_error -> 502 on unexpected Superset error
|
|
# @TEST_EDGE: parse_url_dashboard_resolve_fail -> fallback when dashboard title resolution fails
|
|
# @TEST_EDGE: delete_task_with_runs -> scheduler.remove_validation_job called with delete_runs
|
|
# @TEST_EDGE: parse_dt_invalid_input -> _parse_dt returns None for invalid strings
|
|
# @TEST_EDGE: task_service_dep -> _get_task_service builds ValidationTaskService correctly
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-tests")
|
|
os.environ.setdefault("DEV_MODE", "true")
|
|
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
_src = str(Path(__file__).resolve().parent.parent.parent / "src")
|
|
if _src not in sys.path:
|
|
sys.path.insert(0, _src)
|
|
|
|
|
|
def _make_client(overrides: dict | None = None) -> TestClient:
|
|
from src.api.routes.validation_tasks import router, _get_task_service
|
|
from src.core.database import get_db
|
|
from src.dependencies import get_config_manager, get_current_user, get_scheduler_service, get_task_manager
|
|
from src.schemas.auth import User, RoleSchema
|
|
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/validation-tasks")
|
|
|
|
mock_user = User(
|
|
id="admin-1", username="admin", email="admin@x.com",
|
|
auth_source="LOCAL",
|
|
created_at=__import__("datetime").datetime.now(),
|
|
roles=[RoleSchema(id="r1", name="Admin", description="", permissions=[])],
|
|
)
|
|
|
|
app.dependency_overrides[_get_task_service] = lambda: MagicMock()
|
|
app.dependency_overrides[get_db] = lambda: MagicMock()
|
|
app.dependency_overrides[get_current_user] = lambda: mock_user
|
|
app.dependency_overrides[get_config_manager] = lambda: MagicMock()
|
|
app.dependency_overrides[get_scheduler_service] = lambda: MagicMock()
|
|
app.dependency_overrides[get_task_manager] = lambda: MagicMock()
|
|
if overrides:
|
|
for dep, fn in overrides.items():
|
|
app.dependency_overrides[dep] = fn
|
|
return TestClient(app)
|
|
|
|
|
|
# ── _parse_dt edge cases ──────────────────────────────────────
|
|
|
|
|
|
class TestParseDt:
|
|
"""Direct test of _parse_dt utility function."""
|
|
|
|
# #region test_parse_dt_none [C:2] [TYPE Function]
|
|
def test_parse_dt_none(self):
|
|
from src.api.routes.validation_tasks import _parse_dt
|
|
assert _parse_dt(None) is None
|
|
# #endregion test_parse_dt_none
|
|
|
|
# #region test_parse_dt_valid [C:2] [TYPE Function]
|
|
def test_parse_dt_valid(self):
|
|
from src.api.routes.validation_tasks import _parse_dt
|
|
result = _parse_dt("2024-01-15T10:30:00")
|
|
assert result is not None
|
|
assert result.year == 2024
|
|
# #endregion test_parse_dt_valid
|
|
|
|
# #region test_parse_dt_invalid [C:2] [TYPE Function]
|
|
# @BRIEF Invalid string -> returns None.
|
|
def test_parse_dt_invalid(self):
|
|
from src.api.routes.validation_tasks import _parse_dt
|
|
assert _parse_dt("not-a-date") is None
|
|
# #endregion test_parse_dt_invalid
|
|
|
|
|
|
# ── _get_task_service — dependency injector ────────────────────
|
|
|
|
|
|
class TestGetTaskService:
|
|
"""_get_task_service dependency injector builds ValidationTaskService."""
|
|
|
|
# #region test_get_task_service_constructs [C:2] [TYPE Function]
|
|
# @BRIEF Dependency returns a ValidationTaskService instance.
|
|
def test_get_task_service_constructs(self):
|
|
from src.api.routes.validation_tasks import _get_task_service
|
|
from src.core.database import get_db
|
|
from src.dependencies import get_config_manager, get_current_user
|
|
|
|
mock_db = MagicMock()
|
|
mock_config = MagicMock()
|
|
mock_user = MagicMock()
|
|
mock_user.username = "testuser"
|
|
|
|
app = FastAPI()
|
|
app.dependency_overrides[get_db] = lambda: mock_db
|
|
app.dependency_overrides[get_config_manager] = lambda: mock_config
|
|
app.dependency_overrides[get_current_user] = lambda: mock_user
|
|
|
|
# Use TestClient to trigger the dependency injection
|
|
client = TestClient(app)
|
|
|
|
# We just verify the function signature works
|
|
from src.services.validation_service import ValidationTaskService
|
|
svc = _get_task_service(
|
|
db=mock_db,
|
|
config_manager=mock_config,
|
|
current_user=mock_user,
|
|
)
|
|
assert isinstance(svc, ValidationTaskService)
|
|
# #endregion test_get_task_service_constructs
|
|
|
|
|
|
# ── delete_task with delete_runs ──────────────────────────────
|
|
|
|
|
|
class TestDeleteTaskEdge:
|
|
"""DELETE /validation-tasks/{policy_id} — edge paths."""
|
|
|
|
# #region test_delete_task_with_runs [C:2] [TYPE Function]
|
|
# @BRIEF delete_runs=true passes parameter to service.
|
|
def test_delete_task_with_runs(self):
|
|
mock_svc = MagicMock()
|
|
mock_scheduler = MagicMock()
|
|
|
|
from src.api.routes.validation_tasks import _get_task_service
|
|
from src.dependencies import get_scheduler_service
|
|
client = _make_client({
|
|
_get_task_service: lambda: mock_svc,
|
|
get_scheduler_service: lambda: mock_scheduler,
|
|
})
|
|
|
|
resp = client.delete("/validation-tasks/task-1?delete_runs=true")
|
|
assert resp.status_code == 204
|
|
mock_svc.delete_task.assert_called_with("task-1", delete_runs=True)
|
|
mock_scheduler.remove_validation_job.assert_called_with("task-1")
|
|
# #endregion test_delete_task_with_runs
|
|
|
|
# #region test_delete_task_not_found [C:2] [TYPE Function]
|
|
# @BRIEF Service raises ValueError -> 404.
|
|
def test_delete_task_not_found(self):
|
|
mock_svc = MagicMock()
|
|
mock_svc.delete_task.side_effect = ValueError("Task not found")
|
|
|
|
from src.api.routes.validation_tasks import _get_task_service
|
|
client = _make_client({_get_task_service: lambda: mock_svc})
|
|
|
|
resp = client.delete("/validation-tasks/unknown")
|
|
assert resp.status_code == 404
|
|
# #endregion test_delete_task_not_found
|
|
|
|
|
|
# ── parse_superset_url — edge cases ───────────────────────────
|
|
|
|
|
|
class TestParseSupersetUrlEdge:
|
|
"""POST /validation-tasks/parse-url — edge paths."""
|
|
|
|
PARSE_PAYLOAD = {
|
|
"url": "https://superset.example.com/superset/dashboard/1/",
|
|
"environment_id": "env-1",
|
|
}
|
|
|
|
# #region test_parse_url_dashboard_resolve_fail [C:2] [TYPE Function]
|
|
# @BRIEF Dashboard ID resolved but API call fails -> falls back to str(dashboard_id).
|
|
def test_parse_url_dashboard_resolve_fail(self):
|
|
mock_config = MagicMock()
|
|
mock_config.get_environment.return_value = MagicMock()
|
|
|
|
mock_extractor = MagicMock()
|
|
mock_parsed = MagicMock()
|
|
mock_parsed.dashboard_id = 1
|
|
mock_parsed.chart_id = None
|
|
mock_parsed.query_state = {"native_filters": None, "activeTabs": None, "anchor": None}
|
|
mock_parsed.partial_recovery = None
|
|
mock_parsed.unresolved_references = None
|
|
mock_extractor.parse_superset_link = AsyncMock(return_value=mock_parsed)
|
|
mock_client = MagicMock()
|
|
mock_client.get_dashboard_detail = AsyncMock(side_effect=RuntimeError("API down"))
|
|
mock_extractor.client = mock_client
|
|
|
|
from src.dependencies import get_config_manager
|
|
with patch("src.core.utils.superset_context_extractor.SupersetContextExtractor", return_value=mock_extractor):
|
|
client = _make_client({get_config_manager: lambda: mock_config})
|
|
resp = client.post("/validation-tasks/parse-url", json=self.PARSE_PAYLOAD)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["dashboard_id"] == "1"
|
|
assert data["title"] == "1" # fallback str(dashboard_id)
|
|
# #endregion test_parse_url_dashboard_resolve_fail
|
|
|
|
# #region test_parse_url_unexpected_error [C:2] [TYPE Function]
|
|
# @BRIEF Unexpected exception during parse -> 502.
|
|
def test_parse_url_unexpected_error(self):
|
|
mock_config = MagicMock()
|
|
mock_config.get_environment.return_value = MagicMock()
|
|
|
|
mock_extractor = MagicMock()
|
|
mock_extractor.parse_superset_link = AsyncMock(side_effect=RuntimeError("Unexpected network failure"))
|
|
|
|
from src.dependencies import get_config_manager
|
|
with patch("src.core.utils.superset_context_extractor.SupersetContextExtractor", return_value=mock_extractor):
|
|
client = _make_client({get_config_manager: lambda: mock_config})
|
|
resp = client.post("/validation-tasks/parse-url", json=self.PARSE_PAYLOAD)
|
|
assert resp.status_code == 502
|
|
# #endregion test_parse_url_unexpected_error
|
|
|
|
|
|
# ── trigger_run — edge cases ──────────────────────────────────
|
|
|
|
|
|
class TestTriggerRunEdge:
|
|
"""POST /validation-tasks/{policy_id}/run — edge paths."""
|
|
|
|
# #region test_trigger_run_with_no_run_id [C:2] [TYPE Function]
|
|
# @BRIEF Service returns result without run_id -> run_id is None.
|
|
def test_trigger_run_with_no_run_id(self):
|
|
mock_svc = MagicMock()
|
|
mock_svc.trigger_run = AsyncMock(return_value={"task_id": "t-1"})
|
|
|
|
from src.api.routes.validation_tasks import _get_task_service
|
|
client = _make_client({_get_task_service: lambda: mock_svc})
|
|
|
|
resp = client.post("/validation-tasks/task-1/run")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["spawned_task_id"] == "t-1"
|
|
assert data["run_id"] is None
|
|
# #endregion test_trigger_run_with_no_run_id
|
|
# #endregion Test.Api.ValidationTasks.Edge
|