271 lines
12 KiB
Python
271 lines
12 KiB
Python
# #region Test.Maintenance.BannerRenderer [C:3] [TYPE Module] [SEMANTICS test, maintenance, banner, renderer, coverage]
|
|
# @BRIEF Tests for banner renderer — build_banner_text, _build_banner_text_for_dashboard, rebuild_banner.
|
|
# @RELATION BINDS_TO -> [MaintenanceBannerRenderer]
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.models.maintenance import (
|
|
MaintenanceDashboardBanner,
|
|
MaintenanceDashboardBannerStatus,
|
|
MaintenanceDashboardState,
|
|
MaintenanceDashboardStateStatus,
|
|
MaintenanceEvent,
|
|
MaintenanceEventStatus,
|
|
MaintenanceSettings,
|
|
)
|
|
|
|
|
|
class TestBuildBannerText:
|
|
@pytest.fixture
|
|
def single_event(self):
|
|
return [{"start_time": "2026-06-15T10:00:00Z", "end_time": "2026-06-15T12:00:00Z", "message": "Scheduled maintenance"}]
|
|
|
|
def test_empty_events(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
assert build_banner_text([], "Template") == ""
|
|
|
|
def test_single_event_substitution(self, single_event):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
result = build_banner_text(single_event, "Maintenance: {message} ({start_time} - {end_time})")
|
|
assert "Scheduled maintenance" in result
|
|
assert "2026-06-15 10:00" in result
|
|
assert "2026-06-15 12:00" in result
|
|
|
|
def test_single_event_with_unknown_variables_removed(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [{"start_time": None, "end_time": None, "message": "Test"}]
|
|
result = build_banner_text(events, "Msg: {message} ({unknown_var})")
|
|
assert "Msg: Test" in result
|
|
assert "{unknown_var}" not in result
|
|
|
|
def test_multi_event_combined(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [
|
|
{"start_time": "2026-06-15T10:00:00Z", "end_time": "2026-06-15T12:00:00Z", "message": "First event"},
|
|
{"start_time": "2026-06-16T10:00:00Z", "end_time": "2026-06-16T12:00:00Z", "message": "Second event"},
|
|
]
|
|
result = build_banner_text(events, "Event: {message}")
|
|
assert "First event" in result
|
|
assert "Second event" in result
|
|
assert "---" in result # separator
|
|
|
|
def test_multi_event_with_single_event_path(self):
|
|
"""Multiple events path but len(events)==1 comparison after join."""
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [{"start_time": "2026-06-15T10:00:00Z", "end_time": None, "message": "Test"}]
|
|
result = build_banner_text(events, "Msg: {message} ({start_time})")
|
|
assert "Test" in result
|
|
assert "2026-06-15 10:00" in result
|
|
|
|
def test_html_escaping(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [{"start_time": None, "end_time": None, "message": "<script>alert('xss')</script>"}]
|
|
result = build_banner_text(events, "{message}")
|
|
assert "<script>" in result
|
|
assert "<script>" not in result
|
|
|
|
def test_missing_start_time_defaults(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [{"start_time": None, "end_time": None, "message": "Test"}]
|
|
result = build_banner_text(events, "From {start_time} to {end_time}")
|
|
assert "уточняется" in result
|
|
|
|
def test_invalid_datetime_passthrough(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [{"start_time": "not-a-date", "end_time": None, "message": "Test"}]
|
|
result = build_banner_text(events, "{start_time}")
|
|
assert "not-a-date" in result
|
|
|
|
def test_empty_message_escaping(self):
|
|
from src.services.maintenance._banner_renderer import build_banner_text
|
|
events = [{"start_time": None, "end_time": None, "message": ""}]
|
|
result = build_banner_text(events, "{message}")
|
|
assert result is not None
|
|
|
|
|
|
class TestBuildBannerTextForDashboard:
|
|
def test_no_active_states(self):
|
|
from src.services.maintenance._banner_renderer import _build_banner_text_for_dashboard
|
|
db = MagicMock()
|
|
db.query.return_value.filter.return_value.filter.return_value.all.return_value = []
|
|
result = _build_banner_text_for_dashboard("b1", db, "Template", "UTC")
|
|
assert result == ""
|
|
|
|
def test_active_states_found(self):
|
|
from src.services.maintenance._banner_renderer import _build_banner_text_for_dashboard
|
|
db = MagicMock()
|
|
state = MaintenanceDashboardState(event_id="evt1", dashboard_id=1, banner_id="b1")
|
|
ev = MaintenanceEvent(
|
|
id="evt1",
|
|
start_time=datetime.now(UTC),
|
|
end_time=datetime.now(UTC),
|
|
message="Test event",
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
)
|
|
db.query.return_value.filter.return_value.filter.return_value.all.return_value = [state]
|
|
db.query.return_value.filter.return_value.first.return_value = ev
|
|
result = _build_banner_text_for_dashboard("b1", db, "Msg: {message}", "UTC")
|
|
assert "Test event" in result
|
|
|
|
def test_event_not_in_active_status(self):
|
|
from src.services.maintenance._banner_renderer import _build_banner_text_for_dashboard
|
|
db = MagicMock()
|
|
state = MaintenanceDashboardState(event_id="evt1", dashboard_id=1, banner_id="b1")
|
|
ev = MaintenanceEvent(
|
|
id="evt1", status=MaintenanceEventStatus.CHECKING,
|
|
)
|
|
db.query.return_value.filter.return_value.filter.return_value.all.return_value = [state]
|
|
db.query.return_value.filter.return_value.first.return_value = ev
|
|
result = _build_banner_text_for_dashboard("b1", db, "Template", "UTC")
|
|
assert result == ""
|
|
|
|
def test_exception_returns_empty(self):
|
|
from src.services.maintenance._banner_renderer import _build_banner_text_for_dashboard
|
|
db = MagicMock()
|
|
db.query.side_effect = Exception("DB error")
|
|
result = _build_banner_text_for_dashboard("b1", db, "Template", "UTC")
|
|
assert result == ""
|
|
|
|
def test_event_without_start_end(self):
|
|
from src.services.maintenance._banner_renderer import _build_banner_text_for_dashboard
|
|
db = MagicMock()
|
|
state = MaintenanceDashboardState(event_id="evt1", dashboard_id=1, banner_id="b1")
|
|
ev = MaintenanceEvent(
|
|
id="evt1", start_time=None, end_time=None, message="No times",
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
)
|
|
db.query.return_value.filter.return_value.filter.return_value.all.return_value = [state]
|
|
db.query.return_value.filter.return_value.first.return_value = ev
|
|
result = _build_banner_text_for_dashboard("b1", db, "{message}", "UTC")
|
|
assert "No times" in result
|
|
|
|
|
|
class TestRebuildBanner:
|
|
@pytest.mark.asyncio
|
|
async def test_banner_not_found(self):
|
|
from src.services.maintenance._banner_renderer import rebuild_banner
|
|
db = MagicMock()
|
|
db.query.return_value.filter.return_value.first.return_value = None
|
|
result = await rebuild_banner("nonexistent", db, AsyncMock())
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_banner_not_active(self):
|
|
from src.services.maintenance._banner_renderer import rebuild_banner
|
|
db = MagicMock()
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", status=MaintenanceDashboardBannerStatus.REMOVED,
|
|
chart_id=123, dashboard_id=1, environment_id="e1", banner_text=""
|
|
)
|
|
db.query.return_value.filter.return_value.first.return_value = banner
|
|
result = await rebuild_banner("b1", db, AsyncMock())
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_text_no_chart_id(self):
|
|
from src.services.maintenance._banner_renderer import rebuild_banner
|
|
db = MagicMock()
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", status=MaintenanceDashboardBannerStatus.ACTIVE,
|
|
chart_id=None, dashboard_id=1, environment_id="e1", banner_text=""
|
|
)
|
|
settings = MaintenanceSettings(id="default", banner_template="Tpl", display_timezone="UTC")
|
|
db.query.return_value.filter.return_value.first.return_value = banner
|
|
# First query returns settings, second returns banner
|
|
def query_side_effect(model):
|
|
if model == MaintenanceSettings:
|
|
m = MagicMock()
|
|
m.first.return_value = settings
|
|
return m
|
|
m = MagicMock()
|
|
m.first.return_value = None
|
|
return m
|
|
db.query.side_effect = query_side_effect
|
|
result = await rebuild_banner("b1", db, AsyncMock())
|
|
assert result is True # marked REMOVED since no active events
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_rebuild(self):
|
|
from src.services.maintenance._banner_renderer import rebuild_banner
|
|
db = MagicMock()
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", status=MaintenanceDashboardBannerStatus.ACTIVE,
|
|
chart_id=123, dashboard_id=1, environment_id="e1", banner_text=""
|
|
)
|
|
settings = MaintenanceSettings(id="default", banner_template="Msg: {message}", display_timezone="UTC")
|
|
state = MaintenanceDashboardState(event_id="evt1", dashboard_id=1, banner_id="b1")
|
|
ev = MaintenanceEvent(id="evt1", message="Hello", status=MaintenanceEventStatus.ACTIVE)
|
|
|
|
# Complex mock chain for queries
|
|
call_count = [0]
|
|
def query_side_effect(model):
|
|
call_count[0] += 1
|
|
m = MagicMock()
|
|
if model == MaintenanceDashboardBanner:
|
|
m.filter.return_value.first.return_value = banner
|
|
elif model == MaintenanceSettings:
|
|
m.filter.return_value.first.return_value = settings
|
|
elif model == MaintenanceDashboardState:
|
|
m2 = MagicMock()
|
|
m2.filter.return_value.filter.return_value.all.return_value = [state]
|
|
m.filter.return_value = m2
|
|
elif model == MaintenanceEvent:
|
|
m.filter.return_value.first.return_value = ev
|
|
else:
|
|
m.first.return_value = None
|
|
return m
|
|
db.query.side_effect = query_side_effect
|
|
|
|
mock_superset = AsyncMock()
|
|
mock_superset.update_banner_on_dashboard.return_value = {"result": "ok"}
|
|
|
|
result = await rebuild_banner("b1", db, mock_superset)
|
|
assert result is True
|
|
mock_superset.update_banner_on_dashboard.assert_called_once()
|
|
assert "Hello" in (banner.banner_text or "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_failure(self):
|
|
from src.services.maintenance._banner_renderer import rebuild_banner
|
|
db = MagicMock()
|
|
banner = MaintenanceDashboardBanner(
|
|
id="b1", status=MaintenanceDashboardBannerStatus.ACTIVE,
|
|
chart_id=123, dashboard_id=1, environment_id="e1", banner_text=""
|
|
)
|
|
settings = MaintenanceSettings(id="default", banner_template="Msg: {message}", display_timezone="UTC")
|
|
state = MaintenanceDashboardState(event_id="evt1", dashboard_id=1, banner_id="b1")
|
|
ev = MaintenanceEvent(id="evt1", message="Hello", status=MaintenanceEventStatus.ACTIVE)
|
|
|
|
call_count = [0]
|
|
def query_side_effect(model):
|
|
call_count[0] += 1
|
|
m = MagicMock()
|
|
if model == MaintenanceDashboardBanner:
|
|
m.filter.return_value.first.return_value = banner
|
|
elif model == MaintenanceSettings:
|
|
m.filter.return_value.first.return_value = settings
|
|
elif model == MaintenanceDashboardState:
|
|
m2 = MagicMock()
|
|
m2.filter.return_value.filter.return_value.all.return_value = [state]
|
|
m.filter.return_value = m2
|
|
elif model == MaintenanceEvent:
|
|
m.filter.return_value.first.return_value = ev
|
|
else:
|
|
m.first.return_value = None
|
|
return m
|
|
db.query.side_effect = query_side_effect
|
|
|
|
mock_superset = AsyncMock()
|
|
mock_superset.update_banner_on_dashboard.side_effect = Exception("Update failed")
|
|
|
|
result = await rebuild_banner("b1", db, mock_superset)
|
|
assert result is False
|
|
# #endregion Test.Maintenance.BannerRenderer
|