267 lines
12 KiB
Python
267 lines
12 KiB
Python
# #region Test.Maintenance.DashboardScanner [C:3] [TYPE Module] [SEMANTICS test, maintenance, dashboard, scanner, filter]
|
|
# @BRIEF Tests for dashboard scanner — find_affected_dashboards, _get_linked_dashboards, _apply_dashboard_filters, _resolve_dashboard_title.
|
|
# @RELATION BINDS_TO -> [MaintenanceDashboardScanner]
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.models.maintenance import DashboardScope, MaintenanceSettings
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_superset():
|
|
client = AsyncMock()
|
|
client.get_datasets.return_value = (0, [])
|
|
client.get_dashboards.return_value = (0, [])
|
|
client.get_dataset_detail.return_value = {"linked_dashboards": []}
|
|
return client
|
|
|
|
|
|
class TestFindAffectedDashboards:
|
|
@pytest.mark.asyncio
|
|
async def test_empty_tables(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
result = await find_affected_dashboards([], mock_superset)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_after_normalize(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
result = await find_affected_dashboards(["", " "], mock_superset)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_superset_fetch_error(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.side_effect = Exception("API error")
|
|
with pytest.raises(Exception, match="API error"):
|
|
await find_affected_dashboards(["public.table1"], mock_superset)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_table_based_match(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.return_value = (1, [
|
|
{"id": 1, "schema": "public", "table_name": "table1", "sql": None, "is_sqllab_view": False}
|
|
])
|
|
mock_superset.get_dataset_detail.return_value = {
|
|
"linked_dashboards": [{"id": 10, "title": "Dashboard 10"}]
|
|
}
|
|
result = await find_affected_dashboards(["public.table1"], mock_superset)
|
|
assert 10 in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_virtual_sql_dataset_match(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.return_value = (1, [
|
|
{"id": 2, "schema": "", "table_name": "", "sql": "SELECT * FROM raw.orders", "is_sqllab_view": True}
|
|
])
|
|
mock_superset.get_dataset_detail.return_value = {
|
|
"linked_dashboards": [{"id": 20, "title": "Dashboard 20"}]
|
|
}
|
|
with patch("src.services.maintenance._dashboard_scanner.extract_tables_from_sql", return_value={"raw.orders"}):
|
|
result = await find_affected_dashboards(["raw.orders"], mock_superset)
|
|
assert 20 in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_virtual_sql_no_match(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.return_value = (1, [
|
|
{"id": 2, "schema": "", "table_name": "", "sql": "SELECT * FROM raw.orders", "is_sqllab_view": True}
|
|
])
|
|
mock_superset.get_dataset_detail.return_value = {
|
|
"linked_dashboards": [{"id": 20}]
|
|
}
|
|
with patch("src.services.maintenance._dashboard_scanner.extract_tables_from_sql", return_value=set()):
|
|
result = await find_affected_dashboards(["raw.orders"], mock_superset)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_linked_dashboards_api_failure(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.return_value = (1, [
|
|
{"id": 1, "schema": "public", "table_name": "table1", "sql": None, "is_sqllab_view": False}
|
|
])
|
|
mock_superset.get_dataset_detail.side_effect = Exception("API error")
|
|
result = await find_affected_dashboards(["public.table1"], mock_superset)
|
|
assert result == [] # no linked dashboards due to error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_linked_dashboards(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.return_value = (1, [
|
|
{"id": 1, "schema": "public", "table_name": "table1", "sql": None, "is_sqllab_view": False}
|
|
])
|
|
mock_superset.get_dataset_detail.return_value = {"linked_dashboards": []}
|
|
result = await find_affected_dashboards(["public.table1"], mock_superset)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_with_settings_filtering(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import find_affected_dashboards
|
|
mock_superset.get_datasets.return_value = (1, [
|
|
{"id": 1, "schema": "public", "table_name": "table1", "sql": None, "is_sqllab_view": False}
|
|
])
|
|
mock_superset.get_dataset_detail.return_value = {
|
|
"linked_dashboards": [{"id": 10, "title": "Dashboard 10"}]
|
|
}
|
|
settings = MaintenanceSettings(
|
|
id="default",
|
|
dashboard_scope=DashboardScope.ALL,
|
|
excluded_dashboard_ids=[],
|
|
forced_dashboard_ids=[],
|
|
)
|
|
result = await find_affected_dashboards(["public.table1"], mock_superset, settings)
|
|
assert 10 in result
|
|
|
|
|
|
class TestGetLinkedDashboards:
|
|
@pytest.mark.asyncio
|
|
async def test_no_dataset_id(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _get_linked_dashboards
|
|
result = await _get_linked_dashboards(mock_superset, None)
|
|
assert result == set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exception_handled(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _get_linked_dashboards
|
|
mock_superset.get_dataset_detail.side_effect = Exception("API error")
|
|
result = await _get_linked_dashboards(mock_superset, 1)
|
|
assert result == set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _get_linked_dashboards
|
|
mock_superset.get_dataset_detail.return_value = {
|
|
"linked_dashboards": [{"id": 10}, {"id": 20}]
|
|
}
|
|
result = await _get_linked_dashboards(mock_superset, 1)
|
|
assert result == {10, 20}
|
|
|
|
|
|
class TestApplyDashboardFilters:
|
|
@pytest.mark.asyncio
|
|
async def test_no_settings(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
result = await _apply_dashboard_filters([1, 2], mock_superset, None)
|
|
assert result == [1, 2]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_published_only_filter(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
mock_superset.get_dashboards.return_value = (2, [
|
|
{"id": 1, "published": True},
|
|
{"id": 2, "published": False},
|
|
])
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.PUBLISHED_ONLY,
|
|
)
|
|
result = await _apply_dashboard_filters([1, 2], mock_superset, settings)
|
|
assert result == [1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_draft_only_filter(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
mock_superset.get_dashboards.return_value = (2, [
|
|
{"id": 1, "published": True},
|
|
{"id": 2, "published": False},
|
|
])
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.DRAFT_ONLY,
|
|
)
|
|
result = await _apply_dashboard_filters([1, 2], mock_superset, settings)
|
|
assert result == [2]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_excluded_filter(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.ALL,
|
|
excluded_dashboard_ids=[2],
|
|
)
|
|
result = await _apply_dashboard_filters([1, 2], mock_superset, settings)
|
|
assert result == [1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forced_overrides_excluded(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.ALL,
|
|
excluded_dashboard_ids=[2], forced_dashboard_ids=[2],
|
|
)
|
|
result = await _apply_dashboard_filters([1, 2], mock_superset, settings)
|
|
assert 2 in result # forced overrides excluded
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forced_adds_missing(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.ALL,
|
|
forced_dashboard_ids=[99],
|
|
)
|
|
result = await _apply_dashboard_filters([1], mock_superset, settings)
|
|
assert 99 in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scope_filtering_failure_fallback(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
mock_superset.get_dashboards.side_effect = Exception("API error")
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.PUBLISHED_ONLY,
|
|
)
|
|
result = await _apply_dashboard_filters([1], mock_superset, settings)
|
|
# Falls back to unfiltered list when API fails
|
|
assert result == [1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scope_dashboard_not_in_list(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _apply_dashboard_filters
|
|
mock_superset.get_dashboards.return_value = (1, [
|
|
{"id": 10, "published": True},
|
|
])
|
|
settings = MaintenanceSettings(
|
|
id="default", dashboard_scope=DashboardScope.PUBLISHED_ONLY,
|
|
)
|
|
result = await _apply_dashboard_filters([1], mock_superset, settings)
|
|
assert result == []
|
|
|
|
|
|
class TestResolveDashboardTitle:
|
|
@pytest.mark.asyncio
|
|
async def test_found_by_id(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _resolve_dashboard_title
|
|
mock_superset.get_dashboards.return_value = (2, [
|
|
{"id": 1, "dashboard_title": "My Dashboard"},
|
|
{"id": 2, "title": "Other"},
|
|
])
|
|
result = await _resolve_dashboard_title(1, mock_superset)
|
|
assert result == "My Dashboard"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_found_by_title_fallback(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _resolve_dashboard_title
|
|
mock_superset.get_dashboards.return_value = (1, [
|
|
{"id": 2, "title": "Fallback Title"},
|
|
])
|
|
result = await _resolve_dashboard_title(2, mock_superset)
|
|
assert result == "Fallback Title"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_not_found_returns_id(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _resolve_dashboard_title
|
|
mock_superset.get_dashboards.return_value = (0, [])
|
|
result = await _resolve_dashboard_title(99, mock_superset)
|
|
assert result == "99"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_exception_returns_id(self, mock_superset):
|
|
from src.services.maintenance._dashboard_scanner import _resolve_dashboard_title
|
|
mock_superset.get_dashboards.side_effect = Exception("API error")
|
|
result = await _resolve_dashboard_title(99, mock_superset)
|
|
assert result == "99"
|
|
# #endregion Test.Maintenance.DashboardScanner
|