# [DEF:TestDashboardsApi:Module] # @RELATION: VERIFIES ->[src.api.routes.dashboards] # @COMPLEXITY: 3 # @PURPOSE: Comprehensive contract-driven tests for Dashboard Hub API # @LAYER: Domain (Tests) # @SEMANTICS: tests, dashboards, api, contract, remediation import pytest from fastapi.testclient import TestClient from unittest.mock import MagicMock, patch, AsyncMock from datetime import datetime, timezone from src.app import app from src.api.routes.dashboards import ( DashboardsResponse, DashboardDetailResponse, DashboardTaskHistoryResponse, DatabaseMappingsResponse, ) from src.dependencies import ( get_current_user, has_permission, get_config_manager, get_task_manager, get_resource_service, get_mapping_service, ) # Global mock user mock_user = MagicMock() mock_user.username = "testuser" mock_user.roles = [] admin_role = MagicMock() admin_role.name = "Admin" mock_user.roles.append(admin_role) @pytest.fixture(autouse=True) def mock_deps(): config_manager = MagicMock() task_manager = MagicMock() resource_service = MagicMock() mapping_service = MagicMock() app.dependency_overrides[get_config_manager] = lambda: config_manager app.dependency_overrides[get_task_manager] = lambda: task_manager app.dependency_overrides[get_resource_service] = lambda: resource_service app.dependency_overrides[get_mapping_service] = lambda: mapping_service app.dependency_overrides[get_current_user] = lambda: mock_user # Overrides for specific permission checks app.dependency_overrides[has_permission("plugin:migration", "READ")] = ( lambda: mock_user ) app.dependency_overrides[has_permission("plugin:migration", "EXECUTE")] = ( lambda: mock_user ) app.dependency_overrides[has_permission("plugin:backup", "EXECUTE")] = ( lambda: mock_user ) app.dependency_overrides[has_permission("tasks", "READ")] = lambda: mock_user app.dependency_overrides[has_permission("dashboards", "READ")] = lambda: mock_user yield { "config": config_manager, "task": task_manager, "resource": resource_service, "mapping": mapping_service, } app.dependency_overrides.clear() client = TestClient(app) # --- 1. get_dashboards tests --- # [DEF:test_get_dashboards_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboards_success(mock_deps): """Uses @TEST_FIXTURE: dashboard_list_happy data.""" mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_deps["task"].get_all_tasks.return_value = [] # @TEST_FIXTURE: dashboard_list_happy -> {"id": 1, "title": "Main Revenue"} mock_deps["resource"].get_dashboards_with_status = AsyncMock( return_value=[ { "id": 1, "title": "Main Revenue", "slug": "main-revenue", "git_status": {"branch": "main", "sync_status": "OK"}, } ] ) response = client.get("/api/dashboards?env_id=prod&page=1&page_size=10") assert response.status_code == 200 data = response.json() # exhaustive @POST assertions assert "dashboards" in data assert len(data["dashboards"]) == 1 # @TEST_FIXTURE: expected_count: 1 assert data["dashboards"][0]["title"] == "Main Revenue" assert data["total"] == 1 assert data["page"] == 1 assert data["page_size"] == 10 assert data["total_pages"] == 1 # schema validation DashboardsResponse(**data) # [/DEF:test_get_dashboards_success:Function] # [DEF:test_get_dashboards_with_search:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboards_with_search(mock_deps): mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_deps["task"].get_all_tasks.return_value = [] mock_deps["resource"].get_dashboards_with_status = AsyncMock( return_value=[ {"id": 1, "title": "Sales Report", "slug": "sales"}, {"id": 2, "title": "Marketing", "slug": "marketing"}, ] ) response = client.get("/api/dashboards?env_id=prod&search=sales") assert response.status_code == 200 data = response.json() assert len(data["dashboards"]) == 1 assert data["dashboards"][0]["title"] == "Sales Report" # [/DEF:test_get_dashboards_with_search:Function] # [DEF:test_get_dashboards_empty:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboards_empty(mock_deps): """@TEST_EDGE: empty_dashboards -> {env_id: 'empty_env', expected_total: 0}""" mock_env = MagicMock() mock_env.id = "empty_env" mock_deps["config"].get_environments.return_value = [mock_env] mock_deps["task"].get_all_tasks.return_value = [] mock_deps["resource"].get_dashboards_with_status = AsyncMock(return_value=[]) response = client.get("/api/dashboards?env_id=empty_env") assert response.status_code == 200 data = response.json() assert data["total"] == 0 assert len(data["dashboards"]) == 0 assert data["total_pages"] == 1 DashboardsResponse(**data) # [/DEF:test_get_dashboards_empty:Function] # [DEF:test_get_dashboards_superset_failure:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboards_superset_failure(mock_deps): """@TEST_EDGE: external_superset_failure -> {env_id: 'bad_conn', status: 503}""" mock_env = MagicMock() mock_env.id = "bad_conn" mock_deps["config"].get_environments.return_value = [mock_env] mock_deps["task"].get_all_tasks.return_value = [] mock_deps["resource"].get_dashboards_with_status = AsyncMock( side_effect=Exception("Connection refused") ) response = client.get("/api/dashboards?env_id=bad_conn") assert response.status_code == 503 assert "Failed to fetch dashboards" in response.json()["detail"] # [/DEF:test_get_dashboards_superset_failure:Function] # [DEF:test_get_dashboards_env_not_found:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboards_env_not_found(mock_deps): mock_deps["config"].get_environments.return_value = [] 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] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboards_invalid_pagination(mock_deps): mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] # page < 1 assert client.get("/api/dashboards?env_id=prod&page=0").status_code == 400 assert client.get("/api/dashboards?env_id=prod&page=-1").status_code == 400 # page_size < 1 assert client.get("/api/dashboards?env_id=prod&page_size=0").status_code == 400 # page_size > 100 assert client.get("/api/dashboards?env_id=prod&page_size=101").status_code == 400 # --- 2. get_database_mappings tests --- # [/DEF:test_get_dashboards_invalid_pagination:Function] # [DEF:test_get_database_mappings_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_database_mappings_success(mock_deps): mock_s = MagicMock() mock_s.id = "s" mock_t = MagicMock() mock_t.id = "t" mock_deps["config"].get_environments.return_value = [mock_s, mock_t] mock_deps["mapping"].get_suggestions = AsyncMock( return_value=[{"source_db": "src", "target_db": "dst", "confidence": 0.9}] ) response = client.get("/api/dashboards/db-mappings?source_env_id=s&target_env_id=t") assert response.status_code == 200 data = response.json() assert len(data["mappings"]) == 1 assert data["mappings"][0]["confidence"] == 0.9 DatabaseMappingsResponse(**data) # [/DEF:test_get_database_mappings_success:Function] # [DEF:test_get_database_mappings_env_not_found:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_database_mappings_env_not_found(mock_deps): mock_deps["config"].get_environments.return_value = [] response = client.get( "/api/dashboards/db-mappings?source_env_id=ghost&target_env_id=t" ) assert response.status_code == 404 # --- 3. get_dashboard_detail tests --- # [/DEF:test_get_database_mappings_env_not_found:Function] # [DEF:test_get_dashboard_detail_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_detail_success(mock_deps): with patch("src.api.routes.dashboards.SupersetClient") as mock_client_cls: mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_client = MagicMock() detail_payload = { "id": 42, "title": "Detail", "charts": [], "datasets": [], "chart_count": 0, "dataset_count": 0, } mock_client.get_dashboard_detail.return_value = detail_payload mock_client_cls.return_value = mock_client response = client.get("/api/dashboards/42?env_id=prod") assert response.status_code == 200 data = response.json() assert data["id"] == 42 DashboardDetailResponse(**data) # [/DEF:test_get_dashboard_detail_success:Function] # [DEF:test_get_dashboard_detail_env_not_found:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_detail_env_not_found(mock_deps): mock_deps["config"].get_environments.return_value = [] response = client.get("/api/dashboards/42?env_id=missing") assert response.status_code == 404 # --- 4. get_dashboard_tasks_history tests --- # [/DEF:test_get_dashboard_detail_env_not_found:Function] # [DEF:test_get_dashboard_tasks_history_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_tasks_history_success(mock_deps): now = datetime.now(timezone.utc) task1 = MagicMock( id="t1", plugin_id="superset-backup", status="SUCCESS", started_at=now, finished_at=None, params={"env": "prod", "dashboards": [42]}, result={}, ) mock_deps["task"].get_all_tasks.return_value = [task1] response = client.get("/api/dashboards/42/tasks?env_id=prod") assert response.status_code == 200 data = response.json() assert data["dashboard_id"] == 42 assert len(data["items"]) == 1 DashboardTaskHistoryResponse(**data) # [/DEF:test_get_dashboard_tasks_history_success:Function] # [DEF:test_get_dashboard_tasks_history_sorting:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_tasks_history_sorting(mock_deps): """@POST: Response contains sorted task history (newest first).""" from datetime import timedelta now = datetime.now(timezone.utc) older = now - timedelta(hours=2) newest = now task_old = MagicMock( id="t-old", plugin_id="superset-backup", status="SUCCESS", started_at=older, finished_at=None, params={"env": "prod", "dashboards": [42]}, result={}, ) task_new = MagicMock( id="t-new", plugin_id="superset-backup", status="RUNNING", started_at=newest, finished_at=None, params={"env": "prod", "dashboards": [42]}, result={}, ) # Provide in wrong order to verify the endpoint sorts mock_deps["task"].get_all_tasks.return_value = [task_old, task_new] response = client.get("/api/dashboards/42/tasks?env_id=prod") assert response.status_code == 200 data = response.json() assert len(data["items"]) == 2 # Newest first assert data["items"][0]["id"] == "t-new" assert data["items"][1]["id"] == "t-old" # --- 5. get_dashboard_thumbnail tests --- # [/DEF:test_get_dashboard_tasks_history_sorting:Function] # [DEF:test_get_dashboard_thumbnail_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_thumbnail_success(mock_deps): with patch("src.api.routes.dashboards.SupersetClient") as mock_client_cls: mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_client = MagicMock() mock_response = MagicMock( status_code=200, content=b"img", headers={"Content-Type": "image/png"} ) mock_client.network.request.side_effect = ( lambda method, endpoint, **kw: {"image_url": "url"} if method == "POST" else mock_response ) 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"img" # [/DEF:test_get_dashboard_thumbnail_success:Function] # [DEF:test_get_dashboard_thumbnail_env_not_found:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_thumbnail_env_not_found(mock_deps): mock_deps["config"].get_environments.return_value = [] response = client.get("/api/dashboards/42/thumbnail?env_id=missing") assert response.status_code == 404 # [/DEF:test_get_dashboard_thumbnail_env_not_found:Function] # [DEF:test_get_dashboard_thumbnail_202:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_get_dashboard_thumbnail_202(mock_deps): """@POST: Returns 202 when thumbnail is being prepared by Superset.""" with patch("src.api.routes.dashboards.SupersetClient") as mock_client_cls: mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_client = MagicMock() # POST cache_dashboard_screenshot returns image_url mock_client.network.request.side_effect = [ {"image_url": "/api/v1/dashboard/42/thumbnail/abc123/"}, # POST MagicMock( status_code=202, json=lambda: {"message": "Thumbnail is being generated"}, headers={"Content-Type": "application/json"}, ), # GET thumbnail -> 202 ] mock_client_cls.return_value = mock_client response = client.get("/api/dashboards/42/thumbnail?env_id=prod") assert response.status_code == 202 assert "Thumbnail is being generated" in response.json()["message"] # --- 6. migrate_dashboards tests --- # [/DEF:test_get_dashboard_thumbnail_202:Function] # [DEF:test_migrate_dashboards_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_migrate_dashboards_success(mock_deps): mock_s = MagicMock() mock_s.id = "s" mock_t = MagicMock() mock_t.id = "t" mock_deps["config"].get_environments.return_value = [mock_s, mock_t] mock_deps["task"].create_task = AsyncMock(return_value=MagicMock(id="task-123")) response = client.post( "/api/dashboards/migrate", json={"source_env_id": "s", "target_env_id": "t", "dashboard_ids": [1]}, ) assert response.status_code == 200 assert response.json()["task_id"] == "task-123" # [/DEF:test_migrate_dashboards_success:Function] # [DEF:test_migrate_dashboards_pre_checks:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_migrate_dashboards_pre_checks(mock_deps): # Missing IDs response = client.post( "/api/dashboards/migrate", json={"source_env_id": "s", "target_env_id": "t", "dashboard_ids": []}, ) assert response.status_code == 400 assert "At least one dashboard ID must be provided" in response.json()["detail"] # [/DEF:test_migrate_dashboards_pre_checks:Function] # [DEF:test_migrate_dashboards_env_not_found:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_migrate_dashboards_env_not_found(mock_deps): """@PRE: source_env_id and target_env_id are valid environment IDs.""" mock_deps["config"].get_environments.return_value = [] response = client.post( "/api/dashboards/migrate", json={"source_env_id": "ghost", "target_env_id": "t", "dashboard_ids": [1]}, ) assert response.status_code == 404 assert "Source environment not found" in response.json()["detail"] # --- 7. backup_dashboards tests --- # [/DEF:test_migrate_dashboards_env_not_found:Function] # [DEF:test_backup_dashboards_success:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_backup_dashboards_success(mock_deps): mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_deps["task"].create_task = AsyncMock(return_value=MagicMock(id="backup-123")) response = client.post( "/api/dashboards/backup", json={"env_id": "prod", "dashboard_ids": [1]} ) assert response.status_code == 200 assert response.json()["task_id"] == "backup-123" # [/DEF:test_backup_dashboards_success:Function] # [DEF:test_backup_dashboards_pre_checks:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_backup_dashboards_pre_checks(mock_deps): response = client.post( "/api/dashboards/backup", json={"env_id": "prod", "dashboard_ids": []} ) assert response.status_code == 400 # [/DEF:test_backup_dashboards_pre_checks:Function] # [DEF:test_backup_dashboards_env_not_found:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_backup_dashboards_env_not_found(mock_deps): """@PRE: env_id is a valid environment ID.""" mock_deps["config"].get_environments.return_value = [] response = client.post( "/api/dashboards/backup", json={"env_id": "ghost", "dashboard_ids": [1]} ) assert response.status_code == 404 assert "Environment not found" in response.json()["detail"] # [/DEF:test_backup_dashboards_env_not_found:Function] # [DEF:test_backup_dashboards_with_schedule:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_backup_dashboards_with_schedule(mock_deps): """@POST: If schedule is provided, a scheduled task is created.""" mock_env = MagicMock() mock_env.id = "prod" mock_deps["config"].get_environments.return_value = [mock_env] mock_deps["task"].create_task = AsyncMock(return_value=MagicMock(id="sched-456")) response = client.post( "/api/dashboards/backup", json={"env_id": "prod", "dashboard_ids": [1], "schedule": "0 0 * * *"}, ) assert response.status_code == 200 assert response.json()["task_id"] == "sched-456" # Verify schedule was propagated to create_task call_kwargs = mock_deps["task"].create_task.call_args task_params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {}) assert task_params["schedule"] == "0 0 * * *" # --- 8. Internal logic: _task_matches_dashboard --- # [/DEF:test_backup_dashboards_with_schedule:Function] from src.api.routes.dashboards import _task_matches_dashboard # [DEF:test_task_matches_dashboard_logic:Function] # @RELATION: BINDS_TO ->[TestDashboardsApi] def test_task_matches_dashboard_logic(): task = MagicMock( plugin_id="superset-backup", params={"dashboards": [42], "env": "prod"} ) assert _task_matches_dashboard(task, 42, "prod") is True assert _task_matches_dashboard(task, 43, "prod") is False assert _task_matches_dashboard(task, 42, "dev") is False llm_task = MagicMock( plugin_id="llm_dashboard_validation", params={"dashboard_id": 42, "environment_id": "prod"}, ) assert _task_matches_dashboard(llm_task, 42, "prod") is True assert _task_matches_dashboard(llm_task, 42, None) is True # [/DEF:test_task_matches_dashboard_logic:Function] # [/DEF:TestDashboardsApi:Module]