busya
31680a1bc9
- Auto-expiry: expired events auto-end on GET /events via task manager - Height: _estimate_markdown_height adapted to unit=8px scale with padding detection - CRITICAL-1: removed dead import remove_chart_from_position (QA finding) - CRITICAL-2: added chart alive check in ensure_banner_chart (QA finding) - CRITICAL-3: fixed test assertions update_markdown_chart → update_banner_on_dashboard - @POST contract: fixed return range [2,12] → [19,200] - Tests: 8 new tests (auto-expiry + layout height)
467 lines
17 KiB
Python
467 lines
17 KiB
Python
# #region test_maintenance_api [C:3] [TYPE TestModule] [SEMANTICS test, maintenance, api, fastapi, pytest]
|
|
# @BRIEF Contract tests for Maintenance Banner API endpoints — T016, T025.
|
|
# Tests POST /start, POST /{id}/end, POST /end-all, GET /events, GET /dashboard-banners,
|
|
# GET /settings, PUT /settings.
|
|
# @LAYER Test
|
|
# @RELATION BINDS_TO -> [MaintenanceRoutesModule]
|
|
# @TEST_CONTRACT: POST /api/maintenance/start -> 202 {task_id, maintenance_id, status}
|
|
# @TEST_CONTRACT: POST /api/maintenance/{id}/end -> 202 {task_id, status}
|
|
# @TEST_CONTRACT: POST /api/maintenance/end-all -> 202 {task_id, status}
|
|
# @TEST_EDGE: missing_tables -> 422 validation error
|
|
# @TEST_EDGE: end_time_before_start_time -> 400 validation error
|
|
# @TEST_EDGE: non_existent_event_end -> 404 not found
|
|
import json
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock
|
|
|
|
import pytest
|
|
from fastapi import status
|
|
|
|
# Patch GitService at module level to prevent /app/storage/repositories error
|
|
# This patches the base module BEFORE any imports can reach it
|
|
import sys
|
|
from unittest.mock import MagicMock
|
|
|
|
_mock_git_svc_cls = MagicMock()
|
|
_mock_git_svc_cls.return_value = MagicMock()
|
|
sys.modules['src.services.git._base'] = MagicMock(GitService=_mock_git_svc_cls)
|
|
sys.modules['src.services.git_service'] = MagicMock(GitService=_mock_git_svc_cls)
|
|
|
|
|
|
# ── 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)
|
|
|
|
|
|
def _mock_auth_overrides():
|
|
"""Apply dependency overrides for authentication.
|
|
|
|
Creates a mock admin user so that all RBAC checks pass.
|
|
The Admin role has full access via has_permission()'s special case.
|
|
"""
|
|
from src.dependencies import get_current_user
|
|
from src.app import app
|
|
|
|
# Create admin user with Admin role (full access)
|
|
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
|
|
|
|
|
|
def _clear_overrides():
|
|
"""Clear dependency overrides."""
|
|
from src.app import app
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def auth_mock():
|
|
"""Auto-apply auth mocks for all tests."""
|
|
_mock_auth_overrides()
|
|
yield
|
|
_clear_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, Session as SASession
|
|
from sqlalchemy.pool import StaticPool
|
|
from src.models.mapping import Base
|
|
from src.dependencies import get_db
|
|
from src.models.maintenance import MaintenanceSettings, DashboardScope
|
|
|
|
# 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)
|
|
|
|
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)
|
|
|
|
|
|
# ── T016: POST /api/maintenance/start ─────────────────────────
|
|
|
|
class TestStartEndpoint:
|
|
"""Contract tests for POST /api/maintenance/start."""
|
|
|
|
# #region test_start_returns_202 [C:2] [TYPE Function]
|
|
# @BRIEF Valid request returns 202 with task_id and maintenance_id.
|
|
def test_start_returns_202(self, client, mock_db, mock_task_manager):
|
|
response = client.post(
|
|
"/api/maintenance/start",
|
|
json={
|
|
"tables": ["raw.sales"],
|
|
"start_time": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
|
|
"end_time": (datetime.now(timezone.utc) + timedelta(hours=3)).isoformat(),
|
|
"message": "Test maintenance",
|
|
},
|
|
)
|
|
assert response.status_code == 202
|
|
data = response.json()
|
|
assert "task_id" in data
|
|
assert "maintenance_id" in data
|
|
assert data["status"] == "pending"
|
|
# #endregion test_start_returns_202
|
|
|
|
# #region test_start_missing_tables [C:2] [TYPE Function]
|
|
# @BRIEF Missing tables field returns 422.
|
|
def test_start_missing_tables(self, client, mock_db, mock_task_manager):
|
|
response = client.post(
|
|
"/api/maintenance/start",
|
|
json={
|
|
"start_time": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
# #endregion test_start_missing_tables
|
|
|
|
# #region test_start_end_time_before_start [C:2] [TYPE Function]
|
|
# @BRIEF end_time before start_time returns 400.
|
|
def test_start_end_time_before_start(self, client, mock_db, mock_task_manager):
|
|
response = client.post(
|
|
"/api/maintenance/start",
|
|
json={
|
|
"tables": ["raw.sales"],
|
|
"start_time": (datetime.now(timezone.utc) + timedelta(hours=3)).isoformat(),
|
|
"end_time": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
# #endregion test_start_end_time_before_start
|
|
|
|
# #region test_start_too_many_tables [C:2] [TYPE Function]
|
|
# @BRIEF More than 100 tables returns 422 (Pydantic validation).
|
|
def test_start_too_many_tables(self, client, mock_db, mock_task_manager):
|
|
tables = [f"raw.table{i}" for i in range(101)]
|
|
response = client.post(
|
|
"/api/maintenance/start",
|
|
json={
|
|
"tables": tables,
|
|
"start_time": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
# #endregion test_start_too_many_tables
|
|
|
|
# #region test_start_idempotency [C:2] [TYPE Function]
|
|
# @BRIEF Same (tables, start, end) returns already_active.
|
|
def test_start_idempotency(self, client, mock_db, mock_task_manager):
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
payload = {
|
|
"tables": ["raw.sales"],
|
|
"start_time": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
|
|
"end_time": (datetime.now(timezone.utc) + timedelta(hours=3)).isoformat(),
|
|
}
|
|
|
|
# First call creates event
|
|
response1 = client.post("/api/maintenance/start", json=payload)
|
|
assert response1.status_code == 202
|
|
|
|
# Manually set event to ACTIVE so idempotency check finds it
|
|
maintenance_id = response1.json()["maintenance_id"]
|
|
event = mock_db.query(MaintenanceEvent).filter(
|
|
MaintenanceEvent.id == maintenance_id
|
|
).first()
|
|
if event:
|
|
event.status = MaintenanceEventStatus.ACTIVE
|
|
mock_db.commit()
|
|
|
|
# Second call with same params — idempotent
|
|
response2 = client.post("/api/maintenance/start", json=payload)
|
|
assert response2.status_code == 202
|
|
data = response2.json()
|
|
# Should have maintenance_id and status
|
|
assert data.get("maintenance_id") or data.get("status")
|
|
# #endregion test_start_idempotency
|
|
|
|
|
|
# ── T025: POST /api/maintenance/{id}/end ──────────────────────
|
|
|
|
class TestEndEndpoint:
|
|
"""Contract tests for POST /api/maintenance/{id}/end."""
|
|
|
|
# #region test_end_returns_202 [C:2] [TYPE Function]
|
|
# @BRIEF Valid maintenance_id returns 202.
|
|
def test_end_returns_202(self, client, mock_db, mock_task_manager):
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(timezone.utc),
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
environment_id="test-env",
|
|
)
|
|
mock_db.add(event)
|
|
mock_db.commit()
|
|
|
|
response = client.post(f"/api/maintenance/{event.id}/end")
|
|
assert response.status_code == 202
|
|
data = response.json()
|
|
assert "task_id" in data
|
|
# #endregion test_end_returns_202
|
|
|
|
# #region test_end_not_found [C:2] [TYPE Function]
|
|
# @BRIEF Non-existent maintenance_id returns 404.
|
|
def test_end_not_found(self, client, mock_db, mock_task_manager):
|
|
response = client.post("/api/maintenance/nonexistent-id/end")
|
|
assert response.status_code == 404
|
|
# #endregion test_end_not_found
|
|
|
|
|
|
# ── POST /api/maintenance/end-all ─────────────────────────────
|
|
|
|
class TestEndAllEndpoint:
|
|
"""Contract tests for POST /api/maintenance/end-all."""
|
|
|
|
# #region test_end_all_returns_202 [C:2] [TYPE Function]
|
|
# @BRIEF end-all returns 202.
|
|
def test_end_all_returns_202(self, client, mock_db, mock_task_manager):
|
|
response = client.post("/api/maintenance/end-all")
|
|
assert response.status_code == 202
|
|
data = response.json()
|
|
assert "task_id" in data
|
|
assert data["status"] == "pending"
|
|
# #endregion test_end_all_returns_202
|
|
|
|
|
|
# ── GET /api/maintenance/events ───────────────────────────────
|
|
|
|
class TestEventsEndpoint:
|
|
"""Contract tests for GET /api/maintenance/events."""
|
|
|
|
# #region test_list_events [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(timezone.utc),
|
|
status=MaintenanceEventStatus.ACTIVE,
|
|
environment_id="test-env",
|
|
)
|
|
mock_db.add(active)
|
|
|
|
completed = MaintenanceEvent(
|
|
tables=["raw.done"],
|
|
start_time=datetime.now(timezone.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_list_events
|
|
|
|
# #region test_expired_event_auto_ends [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 src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
from datetime import timedelta
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(timezone.utc) - timedelta(hours=2),
|
|
end_time=datetime.now(timezone.utc) - timedelta(hours=1),
|
|
message="Expired 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_called_once_with(
|
|
plugin_id="maintenance_banner_apply",
|
|
params={
|
|
"operation": "end",
|
|
"event_id": event.id,
|
|
"environment_id": "test-env",
|
|
"user": "system",
|
|
},
|
|
)
|
|
# #endregion test_expired_event_auto_ends
|
|
|
|
# #region test_future_end_time_skipped [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 src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
from datetime import timedelta
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
end_time=datetime.now(timezone.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_future_end_time_skipped
|
|
|
|
# #region test_no_end_time_skipped [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 src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
from datetime import timedelta
|
|
|
|
event = MaintenanceEvent(
|
|
tables=["raw.sales"],
|
|
start_time=datetime.now(timezone.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_no_end_time_skipped
|
|
|
|
|
|
# ── GET /api/maintenance/dashboard-banners ────────────────────
|
|
|
|
class TestDashboardBannersEndpoint:
|
|
"""Contract tests for GET /api/maintenance/dashboard-banners."""
|
|
|
|
# #region test_list_banners [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_list_banners
|
|
|
|
|
|
# ── GET /api/maintenance/settings ─────────────────────────────
|
|
|
|
class TestSettingsEndpoint:
|
|
"""Contract tests for GET/PUT /api/maintenance/settings."""
|
|
|
|
# #region test_get_settings [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"
|
|
# #endregion test_get_settings
|
|
|
|
# #region test_put_settings [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"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["display_timezone"] == "Europe/Moscow"
|
|
# #endregion test_put_settings
|
|
# #endregion test_maintenance_api
|