439 lines
16 KiB
Python
439 lines
16 KiB
Python
# [DEF:backend.src.api.routes.__tests__.test_dashboards:Module]
|
|
# @TIER: STANDARD
|
|
# @PURPOSE: Unit tests for Dashboards API endpoints
|
|
# @LAYER: API
|
|
# @RELATION: TESTS -> backend.src.api.routes.dashboards
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
from datetime import datetime, timezone
|
|
from fastapi.testclient import TestClient
|
|
from src.app import app
|
|
from src.api.routes.dashboards import DashboardsResponse
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
# [DEF:test_get_dashboards_success:Function]
|
|
# @TEST: GET /api/dashboards returns 200 and valid schema
|
|
# @PRE: env_id exists
|
|
# @POST: Response matches DashboardsResponse schema
|
|
def test_get_dashboards_success():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.get_resource_service") as mock_service, \
|
|
patch("src.api.routes.dashboards.get_task_manager") as mock_task_mgr, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
# Mock environment
|
|
mock_env = MagicMock()
|
|
mock_env.id = "prod"
|
|
mock_config.return_value.get_environments.return_value = [mock_env]
|
|
|
|
# Mock task manager
|
|
mock_task_mgr.return_value.get_all_tasks.return_value = []
|
|
|
|
# Mock resource service response
|
|
async def mock_get_dashboards(env, tasks):
|
|
return [
|
|
{
|
|
"id": 1,
|
|
"title": "Sales Report",
|
|
"slug": "sales",
|
|
"git_status": {"branch": "main", "sync_status": "OK"},
|
|
"last_task": {"task_id": "task-1", "status": "SUCCESS"}
|
|
}
|
|
]
|
|
mock_service.return_value.get_dashboards_with_status = AsyncMock(
|
|
side_effect=mock_get_dashboards
|
|
)
|
|
|
|
# Mock permission
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.get("/api/dashboards?env_id=prod")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "dashboards" in data
|
|
assert "total" in data
|
|
assert "page" in data
|
|
|
|
|
|
# [/DEF:test_get_dashboards_success:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboards_with_search:Function]
|
|
# @TEST: GET /api/dashboards filters by search term
|
|
# @PRE: search parameter provided
|
|
# @POST: Only matching dashboards returned
|
|
def test_get_dashboards_with_search():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.get_resource_service") as mock_service, \
|
|
patch("src.api.routes.dashboards.get_task_manager") as mock_task_mgr, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
# Mock environment
|
|
mock_env = MagicMock()
|
|
mock_env.id = "prod"
|
|
mock_config.return_value.get_environments.return_value = [mock_env]
|
|
|
|
mock_task_mgr.return_value.get_all_tasks.return_value = []
|
|
|
|
async def mock_get_dashboards(env, tasks):
|
|
return [
|
|
{"id": 1, "title": "Sales Report", "slug": "sales"},
|
|
{"id": 2, "title": "Marketing Dashboard", "slug": "marketing"}
|
|
]
|
|
mock_service.return_value.get_dashboards_with_status = AsyncMock(
|
|
side_effect=mock_get_dashboards
|
|
)
|
|
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.get("/api/dashboards?env_id=prod&search=sales")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# Filtered by search term
|
|
|
|
|
|
# [/DEF:test_get_dashboards_with_search:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboards_env_not_found:Function]
|
|
# @TEST: GET /api/dashboards returns 404 if env_id missing
|
|
# @PRE: env_id does not exist
|
|
# @POST: Returns 404 error
|
|
def test_get_dashboards_env_not_found():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
mock_config.return_value.get_environments.return_value = []
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.get("/api/dashboards?env_id=nonexistent")
|
|
|
|
assert response.status_code == 404
|
|
assert "Environment not found" in response.json()["detail"]
|
|
|
|
|
|
# [/DEF:test_get_dashboards_env_not_found:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboards_invalid_pagination:Function]
|
|
# @TEST: GET /api/dashboards returns 400 for invalid page/page_size
|
|
# @PRE: page < 1 or page_size > 100
|
|
# @POST: Returns 400 error
|
|
def test_get_dashboards_invalid_pagination():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
mock_env = MagicMock()
|
|
mock_env.id = "prod"
|
|
mock_config.return_value.get_environments.return_value = [mock_env]
|
|
mock_perm.return_value = lambda: True
|
|
|
|
# Invalid page
|
|
response = client.get("/api/dashboards?env_id=prod&page=0")
|
|
assert response.status_code == 400
|
|
assert "Page must be >= 1" in response.json()["detail"]
|
|
|
|
# Invalid page_size
|
|
response = client.get("/api/dashboards?env_id=prod&page_size=101")
|
|
assert response.status_code == 400
|
|
assert "Page size must be between 1 and 100" in response.json()["detail"]
|
|
|
|
|
|
# [/DEF:test_get_dashboards_invalid_pagination:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboard_detail_success:Function]
|
|
# @TEST: GET /api/dashboards/{id} returns dashboard detail with charts and datasets
|
|
def test_get_dashboard_detail_success():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm, \
|
|
patch("src.api.routes.dashboards.SupersetClient") as mock_client_cls:
|
|
|
|
mock_env = MagicMock()
|
|
mock_env.id = "prod"
|
|
mock_config.return_value.get_environments.return_value = [mock_env]
|
|
mock_perm.return_value = lambda: True
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.get_dashboard_detail.return_value = {
|
|
"id": 42,
|
|
"title": "Revenue Dashboard",
|
|
"slug": "revenue-dashboard",
|
|
"url": "/superset/dashboard/42/",
|
|
"description": "Overview",
|
|
"last_modified": "2026-02-20T10:00:00+00:00",
|
|
"published": True,
|
|
"charts": [
|
|
{
|
|
"id": 100,
|
|
"title": "Revenue by Month",
|
|
"viz_type": "line",
|
|
"dataset_id": 7,
|
|
"last_modified": "2026-02-19T10:00:00+00:00",
|
|
"overview": "line"
|
|
}
|
|
],
|
|
"datasets": [
|
|
{
|
|
"id": 7,
|
|
"table_name": "fact_revenue",
|
|
"schema": "mart",
|
|
"database": "Analytics",
|
|
"last_modified": "2026-02-18T10:00:00+00:00",
|
|
"overview": "mart.fact_revenue"
|
|
}
|
|
],
|
|
"chart_count": 1,
|
|
"dataset_count": 1
|
|
}
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
response = client.get("/api/dashboards/42?env_id=prod")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["id"] == 42
|
|
assert payload["chart_count"] == 1
|
|
assert payload["dataset_count"] == 1
|
|
# [/DEF:test_get_dashboard_detail_success:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboard_detail_env_not_found:Function]
|
|
# @TEST: GET /api/dashboards/{id} returns 404 for missing environment
|
|
def test_get_dashboard_detail_env_not_found():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
mock_config.return_value.get_environments.return_value = []
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.get("/api/dashboards/42?env_id=missing")
|
|
|
|
assert response.status_code == 404
|
|
assert "Environment not found" in response.json()["detail"]
|
|
# [/DEF:test_get_dashboard_detail_env_not_found:Function]
|
|
|
|
|
|
# [DEF:test_migrate_dashboards_success:Function]
|
|
# @TEST: POST /api/dashboards/migrate creates migration task
|
|
# @PRE: Valid source_env_id, target_env_id, dashboard_ids
|
|
# @POST: Returns task_id
|
|
def test_migrate_dashboards_success():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.get_task_manager") as mock_task_mgr, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
# Mock environments
|
|
mock_source = MagicMock()
|
|
mock_source.id = "source"
|
|
mock_target = MagicMock()
|
|
mock_target.id = "target"
|
|
mock_config.return_value.get_environments.return_value = [mock_source, mock_target]
|
|
|
|
# Mock task manager
|
|
mock_task = MagicMock()
|
|
mock_task.id = "task-migrate-123"
|
|
mock_task_mgr.return_value.create_task = AsyncMock(return_value=mock_task)
|
|
|
|
# Mock permission
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.post(
|
|
"/api/dashboards/migrate",
|
|
json={
|
|
"source_env_id": "source",
|
|
"target_env_id": "target",
|
|
"dashboard_ids": [1, 2, 3],
|
|
"db_mappings": {"old_db": "new_db"}
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "task_id" in data
|
|
|
|
|
|
# [/DEF:test_migrate_dashboards_success:Function]
|
|
|
|
|
|
# [DEF:test_migrate_dashboards_no_ids:Function]
|
|
# @TEST: POST /api/dashboards/migrate returns 400 for empty dashboard_ids
|
|
# @PRE: dashboard_ids is empty
|
|
# @POST: Returns 400 error
|
|
def test_migrate_dashboards_no_ids():
|
|
with patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.post(
|
|
"/api/dashboards/migrate",
|
|
json={
|
|
"source_env_id": "source",
|
|
"target_env_id": "target",
|
|
"dashboard_ids": []
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "At least one dashboard ID must be provided" in response.json()["detail"]
|
|
|
|
|
|
# [/DEF:test_migrate_dashboards_no_ids:Function]
|
|
|
|
|
|
# [DEF:test_backup_dashboards_success:Function]
|
|
# @TEST: POST /api/dashboards/backup creates backup task
|
|
# @PRE: Valid env_id, dashboard_ids
|
|
# @POST: Returns task_id
|
|
def test_backup_dashboards_success():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.get_task_manager") as mock_task_mgr, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
# Mock environment
|
|
mock_env = MagicMock()
|
|
mock_env.id = "prod"
|
|
mock_config.return_value.get_environments.return_value = [mock_env]
|
|
|
|
# Mock task manager
|
|
mock_task = MagicMock()
|
|
mock_task.id = "task-backup-456"
|
|
mock_task_mgr.return_value.create_task = AsyncMock(return_value=mock_task)
|
|
|
|
# Mock permission
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.post(
|
|
"/api/dashboards/backup",
|
|
json={
|
|
"env_id": "prod",
|
|
"dashboard_ids": [1, 2, 3],
|
|
"schedule": "0 0 * * *"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "task_id" in data
|
|
|
|
|
|
# [/DEF:test_backup_dashboards_success:Function]
|
|
|
|
|
|
# [DEF:test_get_database_mappings_success:Function]
|
|
# @TEST: GET /api/dashboards/db-mappings returns mapping suggestions
|
|
# @PRE: Valid source_env_id, target_env_id
|
|
# @POST: Returns list of database mappings
|
|
def test_get_database_mappings_success():
|
|
with patch("src.api.routes.dashboards.get_mapping_service") as mock_service, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
|
|
# Mock mapping service
|
|
mock_service.return_value.get_suggestions = AsyncMock(return_value=[
|
|
{
|
|
"source_db": "old_sales",
|
|
"target_db": "new_sales",
|
|
"source_db_uuid": "uuid-1",
|
|
"target_db_uuid": "uuid-2",
|
|
"confidence": 0.95
|
|
}
|
|
])
|
|
|
|
# Mock permission
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.get("/api/dashboards/db-mappings?source_env_id=prod&target_env_id=staging")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "mappings" in data
|
|
|
|
|
|
# [/DEF:test_get_database_mappings_success:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboard_tasks_history_filters_success:Function]
|
|
# @TEST: GET /api/dashboards/{id}/tasks returns backup and llm tasks for dashboard
|
|
def test_get_dashboard_tasks_history_filters_success():
|
|
with patch("src.api.routes.dashboards.get_task_manager") as mock_task_mgr, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm:
|
|
now = datetime.now(timezone.utc)
|
|
|
|
llm_task = MagicMock()
|
|
llm_task.id = "task-llm-1"
|
|
llm_task.plugin_id = "llm_dashboard_validation"
|
|
llm_task.status = "SUCCESS"
|
|
llm_task.started_at = now
|
|
llm_task.finished_at = now
|
|
llm_task.params = {"dashboard_id": "42", "environment_id": "prod"}
|
|
llm_task.result = {"summary": "LLM validation complete"}
|
|
|
|
backup_task = MagicMock()
|
|
backup_task.id = "task-backup-1"
|
|
backup_task.plugin_id = "superset-backup"
|
|
backup_task.status = "RUNNING"
|
|
backup_task.started_at = now
|
|
backup_task.finished_at = None
|
|
backup_task.params = {"env": "prod", "dashboards": [42]}
|
|
backup_task.result = {}
|
|
|
|
other_task = MagicMock()
|
|
other_task.id = "task-other"
|
|
other_task.plugin_id = "superset-backup"
|
|
other_task.status = "SUCCESS"
|
|
other_task.started_at = now
|
|
other_task.finished_at = now
|
|
other_task.params = {"env": "prod", "dashboards": [777]}
|
|
other_task.result = {}
|
|
|
|
mock_task_mgr.return_value.get_all_tasks.return_value = [other_task, llm_task, backup_task]
|
|
mock_perm.return_value = lambda: True
|
|
|
|
response = client.get("/api/dashboards/42/tasks?env_id=prod&limit=10")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["dashboard_id"] == 42
|
|
assert len(data["items"]) == 2
|
|
assert {item["plugin_id"] for item in data["items"]} == {"llm_dashboard_validation", "superset-backup"}
|
|
# [/DEF:test_get_dashboard_tasks_history_filters_success:Function]
|
|
|
|
|
|
# [DEF:test_get_dashboard_thumbnail_success:Function]
|
|
# @TEST: GET /api/dashboards/{id}/thumbnail proxies image bytes from Superset
|
|
def test_get_dashboard_thumbnail_success():
|
|
with patch("src.api.routes.dashboards.get_config_manager") as mock_config, \
|
|
patch("src.api.routes.dashboards.has_permission") as mock_perm, \
|
|
patch("src.api.routes.dashboards.SupersetClient") as mock_client_cls:
|
|
mock_env = MagicMock()
|
|
mock_env.id = "prod"
|
|
mock_config.return_value.get_environments.return_value = [mock_env]
|
|
mock_perm.return_value = lambda: True
|
|
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b"fake-image-bytes"
|
|
mock_response.headers = {"Content-Type": "image/png"}
|
|
|
|
def _network_request(method, endpoint, **kwargs):
|
|
if method == "POST":
|
|
return {"image_url": "/api/v1/dashboard/42/screenshot/abc123/"}
|
|
return mock_response
|
|
|
|
mock_client.network.request.side_effect = _network_request
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
response = client.get("/api/dashboards/42/thumbnail?env_id=prod")
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == b"fake-image-bytes"
|
|
assert response.headers["content-type"].startswith("image/png")
|
|
# [/DEF:test_get_dashboard_thumbnail_success:Function]
|
|
|
|
|
|
# [/DEF:backend.src.api.routes.__tests__.test_dashboards:Module]
|