busya
e1dcf7cf90
- decompose _routes.py (947 lines) into a 27-line facade + 5 handler modules (events/preview/start/end/settings), all <=400; extract _chart_layout from _chart_manager (441->381+78); contract IDs preserved verbatim, 9 routes registered in original order
- close RBAC FR-015 invariant gap: tests/api/test_maintenance_routes_rbac.py — 403 denied + 401 unauth on all 9 endpoints with exact guards (root cause of the blind spot: conftest MaintenanceRouteEnv overrode permission closures with lambda: None)
- repair 4 UX contract drifts: EventsTable empty-state + expandedEventIds invariant aligned to implementation (backend terminal-events expansion is intended per MaintenanceEventStateStatuses @POST); Badge phantom loading state removed (INV_9); SettingsPanel fields disabled during Saving implemented per contract
- add 34 L2 component tests (EventsTable/Badge/SettingsPanel) covering declared @UX_STATE/@UX_TEST/@UX_RECOVERY
- split oversized test files (709/686/624 -> all <=520, collected counts identical)
- test hygiene: RootTransaction is_active guard removes SAWarning in shared postgres fixture; AsyncMock create_task coroutine leak fixed in scheduler tests (zero RuntimeWarnings)
- examples/maintenance actualized against current API: optional environment_id with PROD fan-out (batch response), 422 no-PROD-target, GET events/{id}/dashboards, settings field list
Verified: isolated worktree (HEAD + this diff) 729 backend tests passed; frontend 109 passed; ruff clean; anchors balanced; index rebuilt (0 warnings)
412 lines
16 KiB
Python
412 lines
16 KiB
Python
# #region Test.Maintenance.ApiEventsSettings [C:3] [TYPE Module] [SEMANTICS test,maintenance,api,events,settings,fastapi,pytest]
|
|
# @BRIEF Contract tests for maintenance event/read/settings endpoints — split from
|
|
# test_maintenance_api.py. Tests GET /events (+ auto-expiry), GET /events/{id}/dashboards,
|
|
# GET /dashboard-banners, GET /settings, PUT /settings.
|
|
# @LAYER Test
|
|
# @RELATION BINDS_TO -> [Api.Routes.MaintenanceRoutesModule]
|
|
# @TEST_CONTRACT: GET /api/maintenance/events -> 200 {active, completed}
|
|
# @TEST_CONTRACT: GET /api/maintenance/{id}/dashboards -> 200 [historical dashboard ids]
|
|
# @TEST_CONTRACT: GET /api/maintenance/dashboard-banners -> 200 [banners]
|
|
# @TEST_CONTRACT: GET /api/maintenance/settings -> 200 settings
|
|
# @TEST_CONTRACT: PUT /api/maintenance/settings -> 200 updated settings
|
|
# @TEST_EDGE: expired_active_event -> auto-end task dispatched with operation=end params
|
|
# @TEST_EDGE: future_or_null_end_time -> no auto-end task created
|
|
# @TEST_EDGE: terminal_event_history -> completed events retain removed dashboard states in counts
|
|
from datetime import UTC, datetime, timedelta
|
|
import pytest
|
|
|
|
# ── Patch GitService at module level to prevent /app/storage/repositories error ──
|
|
# Only patch git_service (5-line shim), NOT git._base (233 lines — breaks other tests).
|
|
# See test_api_key_routes.py for identical pattern.
|
|
import sys
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
_SAVED_GIT_SERVICE = sys.modules.get('src.services.git_service')
|
|
|
|
_mock_git_svc_cls = MagicMock()
|
|
_mock_git_svc_cls.return_value = MagicMock()
|
|
sys.modules['src.services.git_service'] = MagicMock(GitService=_mock_git_svc_cls)
|
|
|
|
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def _restore_git_modules():
|
|
"""Restore real git_service module after this module's tests finish."""
|
|
yield
|
|
if _SAVED_GIT_SERVICE is not None:
|
|
sys.modules['src.services.git_service'] = _SAVED_GIT_SERVICE
|
|
else:
|
|
sys.modules.pop('src.services.git_service', None)
|
|
|
|
|
|
# ── Fixtures ──────────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
"""Create a TestClient with all dependencies mocked."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.app import app
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def auth_mock(client, mock_db):
|
|
"""Create a valid API key with maintenance permissions and mock JWT user."""
|
|
from src.app import app
|
|
from src.core.auth.api_key import generate_api_key
|
|
from src.dependencies import get_current_user
|
|
from src.models.api_key import APIKey
|
|
|
|
# Create API key for mutation endpoints
|
|
raw, prefix, key_hash = generate_api_key()
|
|
api_key = APIKey(
|
|
key_hash=key_hash,
|
|
prefix=prefix,
|
|
name="Test API Key",
|
|
permissions=["maintenance:start", "maintenance:end", "maintenance:end_all"],
|
|
active=True,
|
|
)
|
|
mock_db.add(api_key)
|
|
mock_db.commit()
|
|
|
|
client.headers.update({"X-API-Key": raw})
|
|
|
|
# Mock JWT user for read-only endpoints
|
|
mock_role = MagicMock()
|
|
mock_role.name = "Admin"
|
|
mock_role.permissions = []
|
|
|
|
mock_user = MagicMock()
|
|
mock_user.username = "test-admin"
|
|
mock_user.roles = [mock_role]
|
|
|
|
async def _mock_get_current_user():
|
|
return mock_user
|
|
|
|
app.dependency_overrides[get_current_user] = _mock_get_current_user
|
|
|
|
yield
|
|
|
|
client.headers.pop("X-API-Key", None)
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db():
|
|
"""Patch get_db dependency to return a clean in-memory session.
|
|
|
|
Uses StaticPool so all Session() instances share the same connection,
|
|
preventing SQLite threading issues where each :memory: connection
|
|
creates a separate database.
|
|
Creates all tables on the shared Base.metadata.
|
|
"""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from src.dependencies import get_db
|
|
from src.models.maintenance import DashboardScope, MaintenanceSettings
|
|
from src.models.mapping import Base
|
|
|
|
# StaticPool ensures all sessions share the same in-memory connection
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False},
|
|
echo=False,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
# Create default settings
|
|
settings = MaintenanceSettings(
|
|
id="default",
|
|
target_environment_id="test-env",
|
|
display_timezone="UTC",
|
|
banner_template="Test: {message} ({start_time}-{end_time})",
|
|
dashboard_scope=DashboardScope.PUBLISHED_ONLY,
|
|
excluded_dashboard_ids=[],
|
|
forced_dashboard_ids=[],
|
|
)
|
|
session.add(settings)
|
|
session.commit()
|
|
|
|
def _get_db_override():
|
|
db = Session()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
from src.app import app
|
|
app.dependency_overrides[get_db] = _get_db_override
|
|
yield session
|
|
app.dependency_overrides.pop(get_db, None)
|
|
session.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_task_manager():
|
|
"""Patch get_task_manager to return a mock."""
|
|
from src.dependencies import get_task_manager
|
|
|
|
mock_tm = MagicMock()
|
|
mock_task = MagicMock()
|
|
mock_task.id = "test-task-id-123"
|
|
mock_tm.create_task = AsyncMock(return_value=mock_task)
|
|
mock_tm.broadcast_maintenance_event = AsyncMock()
|
|
|
|
from src.app import app
|
|
app.dependency_overrides[get_task_manager] = lambda: mock_tm
|
|
yield mock_tm
|
|
app.dependency_overrides.pop(get_task_manager, None)
|
|
|
|
|
|
# ── GET /api/maintenance/events ───────────────────────────────
|
|
|
|
class TestEventsEndpoint:
|
|
"""Contract tests for GET /api/maintenance/events."""
|
|
|
|
# #region Test.MaintenanceApi.TestListEvents [C:2] [TYPE Function]
|
|
# @BRIEF Returns active and completed events.
|
|
def test_list_events(self, client, mock_db, mock_task_manager):
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
active = MaintenanceEvent(
|
|
tables=["raw.active"],
|
|
start_time=datetime.now(UTC),
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
environment_id="test-env",
|
|
)
|
|
mock_db.add(active)
|
|
|
|
completed = MaintenanceEvent(
|
|
tables=["raw.done"],
|
|
start_time=datetime.now(UTC),
|
|
status=MaintenanceEventStatus.COMPLETED,
|
|
environment_id="test-env",
|
|
)
|
|
mock_db.add(completed)
|
|
mock_db.commit()
|
|
|
|
response = client.get("/api/maintenance/events")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "active" in data
|
|
assert "completed" in data
|
|
# #endregion Test.MaintenanceApi.TestListEvents
|
|
|
|
# #region Test.MaintenanceApi.TestHistoricalDashboardCounts [C:2] [TYPE Function]
|
|
# @BRIEF Terminal events retain every linked dashboard state while active events expose only live banners.
|
|
def test_historical_dashboard_counts_include_removed_states(self, client, mock_db, mock_task_manager, monkeypatch):
|
|
from src.api.routes.maintenance import _routes_events
|
|
from src.models.maintenance import (
|
|
MaintenanceDashboardState,
|
|
MaintenanceDashboardStateStatus,
|
|
MaintenanceEvent,
|
|
MaintenanceEventStatus,
|
|
)
|
|
|
|
active = MaintenanceEvent(
|
|
tables=["raw.active"], start_time=datetime.now(UTC), status=MaintenanceEventStatus.ACTIVE, environment_id="test-env"
|
|
)
|
|
completed = MaintenanceEvent(
|
|
tables=["raw.completed"], start_time=datetime.now(UTC), status=MaintenanceEventStatus.COMPLETED, environment_id="test-env"
|
|
)
|
|
mock_db.add_all([active, completed])
|
|
mock_db.flush()
|
|
mock_db.add_all([
|
|
MaintenanceDashboardState(event_id=active.id, dashboard_id=10, status=MaintenanceDashboardStateStatus.ACTIVE),
|
|
MaintenanceDashboardState(event_id=active.id, dashboard_id=11, status=MaintenanceDashboardStateStatus.REMOVED),
|
|
MaintenanceDashboardState(event_id=completed.id, dashboard_id=20, status=MaintenanceDashboardStateStatus.REMOVED),
|
|
MaintenanceDashboardState(event_id=completed.id, dashboard_id=21, status=MaintenanceDashboardStateStatus.REMOVAL_FAILED),
|
|
])
|
|
mock_db.commit()
|
|
monkeypatch.setattr(_routes_events, "get_config_manager", lambda: (_ for _ in ()).throw(AssertionError("list must not call Superset")))
|
|
|
|
response = client.get("/api/maintenance/events")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["active"][0]["affected_count"] == 1
|
|
assert payload["completed"][0]["affected_count"] == 2
|
|
assert payload["completed"][0]["dashboards"] == []
|
|
# #endregion Test.MaintenanceApi.TestHistoricalDashboardCounts
|
|
|
|
# #region Test.MaintenanceApi.TestEventDashboardDetails [C:2] [TYPE Function]
|
|
# @BRIEF Expanding a terminal event returns its historical dashboard IDs on demand.
|
|
def test_event_dashboard_details_include_terminal_states(self, client, mock_db, mock_task_manager, monkeypatch):
|
|
from src.api.routes.maintenance import _routes_events
|
|
from src.models.maintenance import (
|
|
MaintenanceDashboardState,
|
|
MaintenanceDashboardStateStatus,
|
|
MaintenanceEvent,
|
|
MaintenanceEventStatus,
|
|
)
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.completed"], start_time=datetime.now(UTC), status=MaintenanceEventStatus.COMPLETED, environment_id="test-env"
|
|
)
|
|
mock_db.add(event)
|
|
mock_db.flush()
|
|
mock_db.add_all([
|
|
MaintenanceDashboardState(event_id=event.id, dashboard_id=20, status=MaintenanceDashboardStateStatus.REMOVED),
|
|
MaintenanceDashboardState(event_id=event.id, dashboard_id=21, status=MaintenanceDashboardStateStatus.REMOVAL_FAILED),
|
|
])
|
|
mock_db.commit()
|
|
monkeypatch.setattr(_routes_events, "get_config_manager", lambda: MagicMock(get_environments=lambda: []))
|
|
|
|
response = client.get(f"/api/maintenance/events/{event.id}/dashboards")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == [{"id": 20, "title": "20"}, {"id": 21, "title": "21"}]
|
|
# #endregion Test.MaintenanceApi.TestEventDashboardDetails
|
|
|
|
# #region Test.MaintenanceApi.TestExpiredEventAutoEnds [C:2] [TYPE Function]
|
|
# @BRIEF Event past end_time with ACTIVE status → create_task called with auto-end params.
|
|
def test_expired_event_auto_ends(self, client, mock_db, mock_task_manager):
|
|
from datetime import timedelta
|
|
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(UTC) - timedelta(hours=2),
|
|
end_time=datetime.now(UTC) - timedelta(hours=1),
|
|
message="Expired maintenance",
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
environment_id="test-env",
|
|
# Auto-expiry honors the documented invariant: only auto_end=True events
|
|
# are auto-ended; end_time alone stays informational.
|
|
auto_end=True,
|
|
)
|
|
mock_db.add(event)
|
|
mock_db.commit()
|
|
|
|
response = client.get("/api/maintenance/events")
|
|
assert response.status_code == 200
|
|
|
|
mock_task_manager.create_task.assert_called_once_with(
|
|
plugin_id="maintenance_banner_apply",
|
|
params={
|
|
"operation": "end",
|
|
"event_id": event.id,
|
|
"environment_id": "test-env",
|
|
"user": "system",
|
|
},
|
|
)
|
|
# #endregion Test.MaintenanceApi.TestExpiredEventAutoEnds
|
|
|
|
# #region Test.MaintenanceApi.TestFutureEndTimeSkipped [C:2] [TYPE Function]
|
|
# @BRIEF Event with future end_time → no task created, skipped by < now filter.
|
|
def test_future_end_time_skipped(self, client, mock_db, mock_task_manager):
|
|
from datetime import timedelta
|
|
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(UTC) + timedelta(hours=1),
|
|
end_time=datetime.now(UTC) + timedelta(hours=3),
|
|
message="Future maintenance",
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
environment_id="test-env",
|
|
)
|
|
mock_db.add(event)
|
|
mock_db.commit()
|
|
|
|
response = client.get("/api/maintenance/events")
|
|
assert response.status_code == 200
|
|
|
|
mock_task_manager.create_task.assert_not_called()
|
|
# #endregion Test.MaintenanceApi.TestFutureEndTimeSkipped
|
|
|
|
# #region Test.MaintenanceApi.TestNoEndTimeSkipped [C:2] [TYPE Function]
|
|
# @BRIEF Event with end_time=None → no task created, skipped by isnot(None) filter.
|
|
def test_no_end_time_skipped(self, client, mock_db, mock_task_manager):
|
|
from datetime import timedelta
|
|
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(UTC) - timedelta(hours=2),
|
|
end_time=None,
|
|
message="No end time",
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
environment_id="test-env",
|
|
)
|
|
mock_db.add(event)
|
|
mock_db.commit()
|
|
|
|
response = client.get("/api/maintenance/events")
|
|
assert response.status_code == 200
|
|
|
|
mock_task_manager.create_task.assert_not_called()
|
|
# #endregion Test.MaintenanceApi.TestNoEndTimeSkipped
|
|
|
|
|
|
# ── GET /api/maintenance/dashboard-banners ────────────────────
|
|
|
|
class TestDashboardBannersEndpoint:
|
|
"""Contract tests for GET /api/maintenance/dashboard-banners."""
|
|
|
|
# #region Test.MaintenanceApi.TestListBanners [C:2] [TYPE Function]
|
|
# @BRIEF Returns list of active banners.
|
|
def test_list_banners(self, client, mock_db, mock_task_manager):
|
|
from src.models.maintenance import (
|
|
MaintenanceDashboardBanner,
|
|
MaintenanceDashboardBannerStatus,
|
|
)
|
|
|
|
banner = MaintenanceDashboardBanner(
|
|
environment_id="test-env",
|
|
dashboard_id=101,
|
|
chart_id=12345,
|
|
banner_text="Test",
|
|
status=MaintenanceDashboardBannerStatus.ACTIVE,
|
|
)
|
|
mock_db.add(banner)
|
|
mock_db.commit()
|
|
|
|
response = client.get("/api/maintenance/dashboard-banners")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
# #endregion Test.MaintenanceApi.TestListBanners
|
|
|
|
|
|
# ── GET /api/maintenance/settings ─────────────────────────────
|
|
|
|
class TestSettingsEndpoint:
|
|
"""Contract tests for GET/PUT /api/maintenance/settings."""
|
|
|
|
# #region Test.MaintenanceApi.TestGetSettings [C:2] [TYPE Function]
|
|
# @BRIEF GET returns settings.
|
|
def test_get_settings(self, client, mock_db, mock_task_manager):
|
|
response = client.get("/api/maintenance/settings")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "target_environment_id" in data
|
|
assert data["target_environment_id"] == "test-env"
|
|
assert data["default_message"]
|
|
# #endregion Test.MaintenanceApi.TestGetSettings
|
|
|
|
# #region Test.MaintenanceApi.TestPutSettings [C:2] [TYPE Function]
|
|
# @BRIEF PUT updates settings.
|
|
def test_put_settings(self, client, mock_db, mock_task_manager):
|
|
response = client.put(
|
|
"/api/maintenance/settings",
|
|
json={
|
|
"display_timezone": "Europe/Moscow",
|
|
"default_message": "Выполняется обновление витрин.",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["display_timezone"] == "Europe/Moscow"
|
|
assert data["default_message"] == "Выполняется обновление витрин."
|
|
# #endregion Test.MaintenanceApi.TestPutSettings
|
|
# #endregion Test.Maintenance.ApiEventsSettings
|