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
149 lines
4.3 KiB
Python
149 lines
4.3 KiB
Python
"""Unit tests for InstanceEventBus."""
|
|
|
|
import asyncio
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
|
|
|
|
|
|
@pytest.fixture
|
|
def event_bus() -> InstanceEventBus:
|
|
"""Provide a fresh EventBus instance with reset singleton state."""
|
|
bus = InstanceEventBus()
|
|
bus._reset_for_testing()
|
|
return bus
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_payload() -> InstanceEventPayload:
|
|
"""Provide a sample event payload."""
|
|
return {
|
|
"event": "instance.started",
|
|
"instance_id": str(uuid.uuid4()),
|
|
"status": "starting",
|
|
"message": "Container starting...",
|
|
"metadata": {},
|
|
"timestamp": "2026-05-28T12:00:00Z",
|
|
"correlation_id": str(uuid.uuid4()),
|
|
}
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_publish_delivers_to_all_subscribers(
|
|
event_bus: InstanceEventBus,
|
|
sample_payload: InstanceEventPayload,
|
|
) -> None:
|
|
"""All subscribed callbacks should receive the published payload."""
|
|
received: list[Any] = []
|
|
|
|
def callback_1(payload: InstanceEventPayload) -> None:
|
|
received.append(("callback_1", payload))
|
|
|
|
def callback_2(payload: InstanceEventPayload) -> None:
|
|
received.append(("callback_2", payload))
|
|
|
|
def callback_3(payload: InstanceEventPayload) -> None:
|
|
received.append(("callback_3", payload))
|
|
|
|
event_bus.subscribe("instance.started", callback_1)
|
|
event_bus.subscribe("instance.started", callback_2)
|
|
event_bus.subscribe("instance.started", callback_3)
|
|
|
|
await event_bus.publish("instance.started", sample_payload)
|
|
|
|
assert len(received) == 3
|
|
assert received[0][0] == "callback_1"
|
|
assert received[1][0] == "callback_2"
|
|
assert received[2][0] == "callback_3"
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_subscriber_exception_isolation(
|
|
event_bus: InstanceEventBus,
|
|
sample_payload: InstanceEventPayload,
|
|
) -> None:
|
|
"""If one subscriber raises, others should still receive the event."""
|
|
received: list[str] = []
|
|
|
|
def bad_callback(_payload: InstanceEventPayload) -> None:
|
|
raise RuntimeError("boom")
|
|
|
|
def good_callback(_payload: InstanceEventPayload) -> None:
|
|
received.append("good_callback")
|
|
|
|
event_bus.subscribe("instance.started", bad_callback)
|
|
event_bus.subscribe("instance.started", good_callback)
|
|
|
|
# Should not raise
|
|
await event_bus.publish("instance.started", sample_payload)
|
|
|
|
assert received == ["good_callback"]
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_unsubscribe_removes_callback(
|
|
event_bus: InstanceEventBus,
|
|
sample_payload: InstanceEventPayload,
|
|
) -> None:
|
|
"""After unsubscribing, the callback should not be called."""
|
|
received: list[str] = []
|
|
|
|
def callback(_payload: InstanceEventPayload) -> None:
|
|
received.append("callback")
|
|
|
|
unsubscribe = event_bus.subscribe("instance.started", callback)
|
|
unsubscribe()
|
|
|
|
await event_bus.publish("instance.started", sample_payload)
|
|
|
|
assert received == []
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_publish_to_empty_subscriber_list(
|
|
event_bus: InstanceEventBus,
|
|
sample_payload: InstanceEventPayload,
|
|
) -> None:
|
|
"""Publishing to an event type with no subscribers should not raise."""
|
|
await event_bus.publish("instance.started", sample_payload)
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_async_subscriber_supported(
|
|
event_bus: InstanceEventBus,
|
|
sample_payload: InstanceEventPayload,
|
|
) -> None:
|
|
"""Async callbacks should be awaited correctly."""
|
|
received: list[str] = []
|
|
|
|
async def async_callback(_payload: InstanceEventPayload) -> None:
|
|
await asyncio.sleep(0)
|
|
received.append("async_callback")
|
|
|
|
event_bus.subscribe("instance.started", async_callback)
|
|
await event_bus.publish("instance.started", sample_payload)
|
|
|
|
assert received == ["async_callback"]
|
|
|
|
|
|
@pytest.mark.unit
|
|
async def test_unsubscribe_all_clears_subscribers(
|
|
event_bus: InstanceEventBus,
|
|
sample_payload: InstanceEventPayload,
|
|
) -> None:
|
|
"""unsubscribe_all should remove all callbacks for an event type."""
|
|
received: list[str] = []
|
|
|
|
def callback(_payload: InstanceEventPayload) -> None:
|
|
received.append("callback")
|
|
|
|
event_bus.subscribe("instance.started", callback)
|
|
event_bus.unsubscribe_all("instance.started")
|
|
|
|
await event_bus.publish("instance.started", sample_payload)
|
|
|
|
assert received == []
|