84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
# #region Test.McpMaintenance [C:4] [TYPE Module] [SEMANTICS test,mcp,maintenance,validation]
|
|
# @BRIEF Contract tests for the enqueue-only MCP maintenance adapter.
|
|
# @RELATION VERIFIES -> [Services.McpMaintenance.EnqueueStart]
|
|
# @TEST_EDGE empty_tables -> maintenance_tables_invalid
|
|
# @TEST_EDGE invalid_time -> maintenance_time_invalid
|
|
# @TEST_EDGE missing_environment -> maintenance_environment_required
|
|
# @TEST_EDGE oversized_message -> maintenance_message_invalid
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from src.services.mcp_maintenance import enqueue_end_maintenance, enqueue_start_maintenance
|
|
from src.services.mcp_maintenance import reconcile_orphaned_mcp_events
|
|
from src.core.database import Base
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from src.models.maintenance import MaintenanceEvent, MaintenanceEventStatus
|
|
|
|
|
|
# #region Test.McpMaintenance.InvalidPayload [C:3] [TYPE Function]
|
|
# @ingroup Test.McpMaintenance
|
|
# @BRIEF Reject malformed payloads before any database or scheduler dependency is resolved.
|
|
@pytest.mark.parametrize(
|
|
("payload", "error"),
|
|
[
|
|
({"tables": [], "start_time": "2026-08-27T10:00:00", "environment_id": "dev"}, "maintenance_tables_invalid"),
|
|
({"tables": ["orders"], "start_time": "not-a-time", "environment_id": "dev"}, "maintenance_time_invalid"),
|
|
({"tables": ["orders"], "start_time": "2026-08-27T10:00:00"}, "maintenance_environment_required"),
|
|
({"tables": ["orders"], "start_time": "2026-08-27T10:00:00", "environment_id": "dev", "message": "x" * 501}, "maintenance_message_invalid"),
|
|
],
|
|
)
|
|
def test_invalid_payload_fails_before_runtime_dependencies(payload: dict, error: str) -> None:
|
|
record = SimpleNamespace(id="inv-1", subject="user-1", continuation_payload=payload)
|
|
|
|
with pytest.raises(ValueError, match=error):
|
|
enqueue_start_maintenance(record)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("payload", "error"),
|
|
[
|
|
({}, "maintenance_event_or_end_all_required"),
|
|
({"event_id": 123}, "maintenance_event_id_invalid"),
|
|
({"end_all": "yes"}, "maintenance_end_all_invalid"),
|
|
],
|
|
)
|
|
def test_end_payload_requires_explicit_target(payload: dict, error: str) -> None:
|
|
record = SimpleNamespace(id="inv-end-1", subject="user-1", continuation_payload=payload)
|
|
|
|
with pytest.raises(ValueError, match=error):
|
|
enqueue_end_maintenance(record)
|
|
|
|
|
|
def test_orphan_reconciliation_is_bounded_and_idempotent() -> None:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
db = sessionmaker(bind=engine)()
|
|
event = MaintenanceEvent(
|
|
id="orphan-1",
|
|
environment_id="dev",
|
|
tables=["orders"],
|
|
start_time=datetime.now(UTC),
|
|
status=MaintenanceEventStatus.PENDING,
|
|
mcp_invocation_id="inv-1",
|
|
created_at=datetime.now(UTC) - timedelta(seconds=600),
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
|
|
assert reconcile_orphaned_mcp_events(db, timeout_seconds=300) == 1
|
|
assert event.status == MaintenanceEventStatus.FAILED
|
|
assert event.enqueue_error_code == "MCP_ENQUEUE_ORPHAN_TIMEOUT"
|
|
assert reconcile_orphaned_mcp_events(db, timeout_seconds=300) == 0
|
|
db.close()
|
|
|
|
|
|
# #endregion Test.McpMaintenance.InvalidPayload
|
|
|
|
# #endregion Test.McpMaintenance
|