305 lines
14 KiB
Python
305 lines
14 KiB
Python
# #region Test.Maintenance.ChartManager [C:3] [TYPE Module] [SEMANTICS test, maintenance, chart, banner, edge]
|
|
# @BRIEF Tests for maintenance chart manager — ensure_banner_chart, process_dashboards, process_states.
|
|
# @RELATION BINDS_TO -> [MaintenanceChartManager]
|
|
|
|
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 (
|
|
MaintenanceDashboardBanner,
|
|
MaintenanceDashboardBannerStatus,
|
|
MaintenanceDashboardState,
|
|
MaintenanceDashboardStateStatus,
|
|
MaintenanceEvent,
|
|
MaintenanceEventStatus,
|
|
MaintenanceSettings,
|
|
DashboardScope,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db():
|
|
db = MagicMock()
|
|
db.query.return_value.filter.return_value.first.return_value = None
|
|
db.query.return_value.filter.return_value.count.return_value = 0
|
|
db.query.return_value.filter.return_value.all.return_value = []
|
|
return db
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_superset():
|
|
client = AsyncMock()
|
|
client.get_chart.return_value = {"id": 123, "result": {"viz_type": "markdown"}}
|
|
client.create_markdown_chart.return_value = 456
|
|
client.update_dashboard_layout.return_value = {"result": "ok"}
|
|
client.update_banner_on_dashboard.return_value = {"result": "ok"}
|
|
client.remove_chart_from_layout.return_value = {"result": "ok"}
|
|
client.delete_chart.return_value = {"result": "ok"}
|
|
return client
|
|
|
|
|
|
class TestEnsureBannerChart:
|
|
@pytest.mark.asyncio
|
|
async def test_creates_new_banner(self, mock_db, mock_superset):
|
|
"""No existing banner → creates chart, updates layout, persists banner."""
|
|
from src.services.maintenance._chart_manager import ensure_banner_chart
|
|
|
|
result = await ensure_banner_chart(1, "env1", mock_superset, mock_db)
|
|
assert result.chart_id == 456
|
|
assert result.status == MaintenanceDashboardBannerStatus.ACTIVE
|
|
mock_superset.create_markdown_chart.assert_called_once_with(1, "*Maintenance banner placeholder*")
|
|
mock_superset.update_dashboard_layout.assert_called_once_with(1, 456)
|
|
mock_db.add.assert_called_once()
|
|
mock_db.flush.assert_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reuses_existing_banner(self, mock_db, mock_superset):
|
|
"""Existing active banner → verifies chart in Superset and reuses."""
|
|
from src.services.maintenance._chart_manager import ensure_banner_chart
|
|
|
|
existing = MaintenanceDashboardBanner(
|
|
id="b1", dashboard_id=1, environment_id="env1",
|
|
chart_id=123, status=MaintenanceDashboardBannerStatus.ACTIVE, banner_text=""
|
|
)
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = existing
|
|
mock_superset.get_chart.return_value = {"id": 123}
|
|
|
|
result = await ensure_banner_chart(1, "env1", mock_superset, mock_db)
|
|
assert result is existing
|
|
mock_superset.get_chart.assert_called_once_with(123)
|
|
mock_superset.create_markdown_chart.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_banner_removed_and_recreated(self, mock_db, mock_superset):
|
|
"""Existing but stale banner → marked REMOVED, new one created."""
|
|
from src.services.maintenance._chart_manager import ensure_banner_chart
|
|
|
|
existing = MaintenanceDashboardBanner(
|
|
id="b1", dashboard_id=1, environment_id="env1",
|
|
chart_id=123, status=MaintenanceDashboardBannerStatus.ACTIVE, banner_text=""
|
|
)
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = existing
|
|
mock_superset.get_chart.side_effect = Exception("Chart not found")
|
|
|
|
result = await ensure_banner_chart(1, "env1", mock_superset, mock_db)
|
|
assert existing.status == MaintenanceDashboardBannerStatus.REMOVED
|
|
assert result.chart_id == 456 # new chart
|
|
mock_superset.create_markdown_chart.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_chart_failure(self, mock_db, mock_superset):
|
|
"""Failure to create chart → exception propagates."""
|
|
from src.services.maintenance._chart_manager import ensure_banner_chart
|
|
|
|
mock_superset.create_markdown_chart.side_effect = Exception("API error")
|
|
with pytest.raises(Exception, match="API error"):
|
|
await ensure_banner_chart(1, "env1", mock_superset, mock_db)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_layout_update_failure_cleans_up_chart(self, mock_db, mock_superset):
|
|
"""Failure to update layout → deletes chart and raises."""
|
|
from src.services.maintenance._chart_manager import ensure_banner_chart
|
|
|
|
mock_superset.update_dashboard_layout.side_effect = Exception("Layout error")
|
|
with pytest.raises(Exception, match="Layout error"):
|
|
await ensure_banner_chart(1, "env1", mock_superset, mock_db)
|
|
mock_superset.delete_chart.assert_called_once_with(456)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_layout_update_failure_cleanup_ignores_delete_error(self, mock_db, mock_superset):
|
|
"""Delete chart failure during cleanup does not mask original error."""
|
|
from src.services.maintenance._chart_manager import ensure_banner_chart
|
|
|
|
mock_superset.update_dashboard_layout.side_effect = Exception("Layout error")
|
|
mock_superset.delete_chart.side_effect = Exception("Delete also failed")
|
|
with pytest.raises(Exception, match="Layout error"):
|
|
await ensure_banner_chart(1, "env1", mock_superset, mock_db)
|
|
|
|
|
|
class TestProcessDashboardsForStart:
|
|
@pytest.mark.asyncio
|
|
async def test_successful_processing(self, mock_db, mock_superset):
|
|
"""All dashboards processed successfully."""
|
|
from src.services.maintenance._chart_manager import _process_dashboards_for_start
|
|
|
|
mock_superset.get_dashboards.return_value = (1, [{"id": 1, "dashboard_title": "Test"}])
|
|
|
|
successful, failed, banners = await _process_dashboards_for_start(
|
|
[1], "evt1", "env1", "Template: {message}", "UTC", mock_superset, mock_db
|
|
)
|
|
assert len(successful) == 1
|
|
assert len(failed) == 0
|
|
assert len(banners) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_banner_update_failure_handled(self, mock_db, mock_superset):
|
|
"""Failed banner update → records as APPLY_FAILED."""
|
|
from src.services.maintenance._chart_manager import _process_dashboards_for_start
|
|
|
|
mock_superset.update_banner_on_dashboard.side_effect = Exception("Update failed")
|
|
mock_superset.get_dashboards.return_value = (1, [{"id": 1, "dashboard_title": "Test"}])
|
|
|
|
successful, failed, banners = await _process_dashboards_for_start(
|
|
[1], "evt1", "env1", "Template", "UTC", mock_superset, mock_db
|
|
)
|
|
assert len(successful) == 0
|
|
assert len(failed) == 1
|
|
assert len(banners) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_top_level_exception_handling(self, mock_db, mock_superset):
|
|
"""Exception outside banner creation → recorded as failed."""
|
|
from src.services.maintenance._chart_manager import _process_dashboards_for_start
|
|
|
|
mock_superset.create_markdown_chart.side_effect = Exception("Chart creation failed")
|
|
|
|
successful, failed, banners = await _process_dashboards_for_start(
|
|
[1], "evt1", "env1", "Template", "UTC", mock_superset, mock_db
|
|
)
|
|
assert len(successful) == 0
|
|
assert len(failed) == 1
|
|
|
|
|
|
class TestProcessStatesForEnd:
|
|
@pytest.mark.asyncio
|
|
async def test_no_banner_id(self, mock_db, mock_superset):
|
|
"""State with no banner_id → marked REMOVED."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
state.banner_id = None
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 1
|
|
assert state.status == MaintenanceDashboardStateStatus.REMOVED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_events_still_active_rebuilds(self, mock_db, mock_superset):
|
|
"""Other active events exist → rebuild banner."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", dashboard_id=1, chart_id=123,
|
|
status=MaintenanceDashboardBannerStatus.ACTIVE, banner_text="old"
|
|
)
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, banner_id="b1",
|
|
status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
|
|
# Mock query: other active events exist
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.count.return_value = 1
|
|
mock_db.query.return_value.filter.return_value.first.return_value = banner
|
|
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 1
|
|
assert state.status == MaintenanceDashboardStateStatus.REMOVED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_other_events_removes_banner(self, mock_db, mock_superset):
|
|
"""No other active events → remove banner fully."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", dashboard_id=1, chart_id=123,
|
|
status=MaintenanceDashboardBannerStatus.ACTIVE, banner_text="old"
|
|
)
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, banner_id="b1",
|
|
status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.count.return_value = 0
|
|
mock_db.query.return_value.filter.return_value.first.return_value = banner
|
|
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 1
|
|
assert banner.status == MaintenanceDashboardBannerStatus.REMOVED
|
|
assert state.status == MaintenanceDashboardStateStatus.REMOVED
|
|
mock_superset.remove_chart_from_layout.assert_called_once_with(1, 123)
|
|
mock_superset.delete_chart.assert_called_once_with(123)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_chart_failure(self, mock_db, mock_superset):
|
|
"""Failure to remove chart from layout → REMOVAL_FAILED."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", dashboard_id=1, chart_id=123,
|
|
status=MaintenanceDashboardBannerStatus.ACTIVE, banner_text="old"
|
|
)
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, banner_id="b1",
|
|
status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.count.return_value = 0
|
|
mock_db.query.return_value.filter.return_value.first.return_value = banner
|
|
mock_superset.remove_chart_from_layout.side_effect = Exception("Remove failed")
|
|
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 0
|
|
assert len(failed) == 1
|
|
assert state.status == MaintenanceDashboardStateStatus.REMOVAL_FAILED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_chart_failure(self, mock_db, mock_superset):
|
|
"""Failure to delete chart → REMOVAL_FAILED."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", dashboard_id=1, chart_id=123,
|
|
status=MaintenanceDashboardBannerStatus.ACTIVE, banner_text="old"
|
|
)
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, banner_id="b1",
|
|
status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.count.return_value = 0
|
|
mock_db.query.return_value.filter.return_value.first.return_value = banner
|
|
mock_superset.delete_chart.side_effect = Exception("Delete failed")
|
|
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 0
|
|
assert len(failed) == 1
|
|
assert state.status == MaintenanceDashboardStateStatus.REMOVAL_FAILED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_banner_record(self, mock_db, mock_superset):
|
|
"""State with banner_id but no banner record → REMOVED."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, banner_id="nonexistent",
|
|
status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.count.return_value = 0
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 1
|
|
assert state.status == MaintenanceDashboardStateStatus.REMOVED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_top_level_exception(self, mock_db, mock_superset):
|
|
"""Exception at top level → REMOVAL_FAILED."""
|
|
from src.services.maintenance._chart_manager import _process_states_for_end
|
|
|
|
state = MaintenanceDashboardState(
|
|
event_id="evt1", dashboard_id=1, banner_id="b1",
|
|
status=MaintenanceDashboardStateStatus.ACTIVE
|
|
)
|
|
mock_db.query.side_effect = Exception("DB error")
|
|
|
|
removed, failed = await _process_states_for_end([state], "evt1", mock_superset, mock_db)
|
|
assert removed == 0
|
|
assert len(failed) == 1
|
|
# #endregion Test.Maintenance.ChartManager
|