alex
4a7f24348c
- Add instance_events and health_checks tables with Alembic migration - InstanceEventBus: typed pub/sub singleton with wildcard support - HealthMonitor: async background loop polling containers every 15s - SSE endpoint GET /events/stream with auth and connection limits - Lifecycle hooks in tool_instances.py (create/start/stop/restart/delete) - Structured JSON logging with correlation IDs - 15 new unit tests (EventBus, HealthMonitor, MonitoringModels) Quality gates: pytest 15 new passed, ruff clean
293 lines
8.9 KiB
Python
293 lines
8.9 KiB
Python
"""Unit tests for HealthMonitor state-transition logic."""
|
|
|
|
import asyncio
|
|
import uuid
|
|
from contextlib import suppress
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from src.models.health_check import HealthCheck
|
|
from src.models.tool_instance import ToolInstance
|
|
from src.models.user import User
|
|
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
|
|
from src.services.health_monitor import HealthMonitor, HealthSnapshot
|
|
|
|
|
|
@pytest.fixture
|
|
def event_bus() -> InstanceEventBus:
|
|
"""Provide a fresh EventBus instance."""
|
|
bus = InstanceEventBus()
|
|
bus._reset_for_testing()
|
|
return bus
|
|
|
|
|
|
@pytest.fixture
|
|
def health_monitor(event_bus: InstanceEventBus) -> HealthMonitor:
|
|
"""Provide a HealthMonitor with a short poll interval for testing."""
|
|
monitor = HealthMonitor(event_bus)
|
|
monitor.POLL_INTERVAL_SECONDS = 0.1
|
|
return monitor
|
|
|
|
|
|
async def _create_running_instance(db_session) -> ToolInstance:
|
|
"""Helper to create a user and a running tool instance."""
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email="hm@example.com",
|
|
name="HM Test",
|
|
authentik_id="auth-hm",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
instance = ToolInstance(
|
|
id=uuid.uuid4(),
|
|
name="hm-test-instance",
|
|
display_name="HM Test Instance",
|
|
tool_type_id=uuid.uuid4(),
|
|
repository_id=uuid.uuid4(),
|
|
project_id=uuid.uuid4(),
|
|
owner_id=user.id,
|
|
status="running",
|
|
container_id="container123",
|
|
public_url="https://example.trycloudflare.com",
|
|
)
|
|
db_session.add(instance)
|
|
await db_session.commit()
|
|
return instance
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_detects_container_crash(
|
|
db_session,
|
|
event_bus: InstanceEventBus,
|
|
health_monitor: HealthMonitor,
|
|
) -> None:
|
|
"""Monitor should detect exited container and publish error event."""
|
|
instance = await _create_running_instance(db_session)
|
|
|
|
events_captured: list[InstanceEventPayload] = []
|
|
|
|
def capture_event(payload: InstanceEventPayload) -> None:
|
|
events_captured.append(payload)
|
|
|
|
event_bus.subscribe("instance.error", capture_event)
|
|
|
|
with (
|
|
patch(
|
|
"src.services.health_monitor.get_container_status",
|
|
return_value={"status": "exited", "exit_code": 137, "health": None},
|
|
),
|
|
patch(
|
|
"src.services.health_monitor.check_tunnel_health",
|
|
return_value={"healthy": False, "tunnel_status": "not_applicable"},
|
|
),
|
|
):
|
|
await health_monitor._check_instance(db_session, instance)
|
|
|
|
# Refresh instance from DB
|
|
await db_session.refresh(instance)
|
|
assert instance.status == "error"
|
|
|
|
# Event published
|
|
assert len(events_captured) == 1
|
|
assert events_captured[0]["event"] == "instance.error"
|
|
assert events_captured[0]["status"] == "error"
|
|
assert events_captured[0]["metadata"]["exit_code"] == 137
|
|
|
|
# Health check row inserted
|
|
result = await db_session.execute(
|
|
select(HealthCheck).where(HealthCheck.instance_id == instance.id)
|
|
)
|
|
check = result.scalar_one()
|
|
assert check.container_status == "exited"
|
|
assert check.exit_code == 137
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_detects_tunnel_failure(
|
|
db_session,
|
|
event_bus: InstanceEventBus,
|
|
health_monitor: HealthMonitor,
|
|
) -> None:
|
|
"""Monitor should detect tunnel failure and mark unhealthy."""
|
|
instance = await _create_running_instance(db_session)
|
|
|
|
events_captured: list[InstanceEventPayload] = []
|
|
|
|
def capture_event(payload: InstanceEventPayload) -> None:
|
|
events_captured.append(payload)
|
|
|
|
event_bus.subscribe("instance.health_changed", capture_event)
|
|
|
|
with (
|
|
patch(
|
|
"src.services.health_monitor.get_container_status",
|
|
return_value={"status": "running", "exit_code": None, "health": "healthy"},
|
|
),
|
|
patch(
|
|
"src.services.health_monitor.check_tunnel_health",
|
|
return_value={
|
|
"healthy": False,
|
|
"tunnel_status": "error_response",
|
|
"status_code": 502,
|
|
},
|
|
),
|
|
):
|
|
await health_monitor._check_instance(db_session, instance)
|
|
|
|
await db_session.refresh(instance)
|
|
assert instance.status == "unhealthy"
|
|
|
|
assert len(events_captured) == 1
|
|
assert events_captured[0]["event"] == "instance.health_changed"
|
|
assert events_captured[0]["status"] == "unhealthy"
|
|
assert events_captured[0]["metadata"]["previous_status"] == "running"
|
|
|
|
result = await db_session.execute(
|
|
select(HealthCheck).where(HealthCheck.instance_id == instance.id)
|
|
)
|
|
check = result.scalar_one()
|
|
assert check.tunnel_healthy is False
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_detects_recovery(
|
|
db_session,
|
|
event_bus: InstanceEventBus,
|
|
health_monitor: HealthMonitor,
|
|
) -> None:
|
|
"""Monitor should detect recovery from unhealthy to running."""
|
|
instance = await _create_running_instance(db_session)
|
|
instance.status = "unhealthy"
|
|
await db_session.commit()
|
|
|
|
# Seed last known state as unhealthy
|
|
health_monitor._last_known_state[instance.id] = HealthSnapshot(
|
|
container_status="running",
|
|
container_healthy=None,
|
|
tunnel_healthy=False,
|
|
exit_code=None,
|
|
)
|
|
|
|
events_captured: list[InstanceEventPayload] = []
|
|
|
|
def capture_event(payload: InstanceEventPayload) -> None:
|
|
events_captured.append(payload)
|
|
|
|
event_bus.subscribe("instance.health_changed", capture_event)
|
|
|
|
with (
|
|
patch(
|
|
"src.services.health_monitor.get_container_status",
|
|
return_value={"status": "running", "exit_code": None, "health": None},
|
|
),
|
|
patch(
|
|
"src.services.health_monitor.check_tunnel_health",
|
|
return_value={
|
|
"healthy": True,
|
|
"tunnel_status": "healthy",
|
|
"status_code": 200,
|
|
},
|
|
),
|
|
):
|
|
await health_monitor._check_instance(db_session, instance)
|
|
|
|
await db_session.refresh(instance)
|
|
assert instance.status == "running"
|
|
|
|
assert len(events_captured) == 1
|
|
assert events_captured[0]["status"] == "running"
|
|
assert events_captured[0]["metadata"]["previous_status"] == "unhealthy"
|
|
|
|
result = await db_session.execute(
|
|
select(HealthCheck).where(HealthCheck.instance_id == instance.id)
|
|
)
|
|
check = result.scalar_one()
|
|
assert check.tunnel_healthy is True
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_skips_writes_when_no_state_change(
|
|
db_session,
|
|
event_bus: InstanceEventBus,
|
|
health_monitor: HealthMonitor,
|
|
) -> None:
|
|
"""Two identical polls should result in only one health_checks row."""
|
|
instance = await _create_running_instance(db_session)
|
|
|
|
with (
|
|
patch(
|
|
"src.services.health_monitor.get_container_status",
|
|
return_value={"status": "running", "exit_code": None, "health": None},
|
|
),
|
|
patch(
|
|
"src.services.health_monitor.check_tunnel_health",
|
|
return_value={
|
|
"healthy": True,
|
|
"tunnel_status": "healthy",
|
|
"status_code": 200,
|
|
},
|
|
),
|
|
):
|
|
await health_monitor._check_instance(db_session, instance)
|
|
await health_monitor._check_instance(db_session, instance)
|
|
|
|
result = await db_session.execute(
|
|
select(HealthCheck).where(HealthCheck.instance_id == instance.id)
|
|
)
|
|
assert len(result.scalars().all()) == 1
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_docker_exception_resilience(
|
|
db_session,
|
|
event_bus: InstanceEventBus,
|
|
health_monitor: HealthMonitor,
|
|
) -> None:
|
|
"""Docker exception should be caught and not propagate."""
|
|
instance = await _create_running_instance(db_session)
|
|
|
|
events_captured: list[InstanceEventPayload] = []
|
|
|
|
def capture_event(payload: InstanceEventPayload) -> None:
|
|
events_captured.append(payload)
|
|
|
|
event_bus.subscribe("instance.error", capture_event)
|
|
event_bus.subscribe("instance.health_changed", capture_event)
|
|
|
|
with patch(
|
|
"src.services.health_monitor.get_container_status",
|
|
side_effect=RuntimeError("docker exploded"),
|
|
):
|
|
# Should not raise
|
|
await health_monitor._check_instance(db_session, instance)
|
|
|
|
# No DB writes
|
|
result = await db_session.execute(
|
|
select(HealthCheck).where(HealthCheck.instance_id == instance.id)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
# No events published
|
|
assert events_captured == []
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_monitor_start_stop(health_monitor: HealthMonitor) -> None:
|
|
"""Start and stop should manage the background task."""
|
|
health_monitor.start()
|
|
task = health_monitor._task
|
|
assert task is not None
|
|
assert not task.done()
|
|
|
|
health_monitor.stop()
|
|
if task is not None and not task.done():
|
|
with suppress(asyncio.CancelledError):
|
|
await task
|
|
assert task is not None
|
|
assert task.cancelled() or task.done()
|
|
assert health_monitor._last_known_state == {}
|