alex
2682e0268c
- Integration tests: SSE auth, connection limits, lifecycle hooks, event persistence (6 tests)
- Instance events history API: GET /instances/{id}/events
- Documentation updates: terminal.md, backend.md, frontend.md
- Performance: SSE max 5 connections, health monitor write-on-change
Quality gates: pytest 21 monitoring passed, 172 unit passed (4 pre-existing), vitest 14 passed, tsc clean, eslint clean, ruff clean
269 lines
7.8 KiB
Python
269 lines
7.8 KiB
Python
"""Integration tests for SSE endpoint and lifecycle event flow."""
|
|
|
|
import asyncio
|
|
import uuid
|
|
from collections.abc import Generator
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.api import events as events_module
|
|
from src.auth.session import decode_session_cookie
|
|
from src.config import Settings
|
|
from src.models.git_repository import GitRepository
|
|
from src.models.instance_event import InstanceEvent
|
|
from src.models.project import Project
|
|
from src.models.tool_instance import ToolInstance
|
|
from src.models.tool_type import ToolType
|
|
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
|
|
|
|
|
|
@pytest.fixture
|
|
def event_bus() -> Generator[InstanceEventBus, None, None]:
|
|
bus = InstanceEventBus()
|
|
bus._reset_for_testing()
|
|
yield bus
|
|
bus._reset_for_testing()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_payload() -> InstanceEventPayload:
|
|
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()),
|
|
}
|
|
|
|
|
|
def _get_user_id_from_client(client: TestClient) -> uuid.UUID | None:
|
|
settings = Settings()
|
|
cookie = client.cookies.get("session")
|
|
if not cookie:
|
|
return None
|
|
session = decode_session_cookie(settings=settings, cookie_value=cookie)
|
|
if session and "user_id" in session:
|
|
return uuid.UUID(session["user_id"])
|
|
return None
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_sse_requires_auth(test_client: TestClient) -> None:
|
|
response = test_client.get("/events/stream")
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_sse_enforces_connection_limit(authenticated_client: TestClient) -> None:
|
|
user_id = _get_user_id_from_client(authenticated_client)
|
|
assert user_id is not None
|
|
|
|
events_module._connection_counts[user_id] = events_module.MAX_CONNECTIONS_PER_USER
|
|
try:
|
|
response = authenticated_client.get("/events/stream")
|
|
assert response.status_code == 429
|
|
finally:
|
|
events_module._connection_counts.pop(user_id, None)
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_sse_event_generator_format() -> None:
|
|
"""Test the SSE endpoint is registered."""
|
|
from src.api.events import router
|
|
|
|
route_paths = [getattr(r, "path", "") for r in router.routes]
|
|
assert any("/stream" in str(p) for p in route_paths)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_lifecycle_hook_publishes_event_and_persists(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
) -> None:
|
|
"""Test that the lifecycle hook publishes an event and persists an audit row."""
|
|
user_id = _get_user_id_from_client(authenticated_client)
|
|
assert user_id is not None
|
|
|
|
project = Project(
|
|
id=uuid.uuid4(),
|
|
name="test-project",
|
|
description="Test",
|
|
owner_id=user_id,
|
|
)
|
|
repo = GitRepository(
|
|
id=uuid.uuid4(),
|
|
name="test-repo",
|
|
path="/tmp/test-repo",
|
|
project_id=project.id,
|
|
owner_id=user_id,
|
|
remote_url="https://github.com/test/repo.git",
|
|
)
|
|
tool_type = ToolType(
|
|
id=uuid.uuid4(),
|
|
name="test-tool",
|
|
display_name="Test Tool",
|
|
category="other",
|
|
interface_type="web",
|
|
requires_port=True,
|
|
default_port=8080,
|
|
definition_type="legacy",
|
|
compose_template="version: '3.8'\nservices:\n app:\n image: alpine\n command: sleep 3600\n",
|
|
)
|
|
db_session.add_all([project, repo, tool_type])
|
|
await db_session.commit()
|
|
|
|
instance = ToolInstance(
|
|
id=uuid.uuid4(),
|
|
name="test-instance",
|
|
display_name="Test Instance",
|
|
tool_type_id=tool_type.id,
|
|
repository_id=repo.id,
|
|
project_id=project.id,
|
|
owner_id=user_id,
|
|
status="pending",
|
|
compose_path="/tmp/test-compose.yml",
|
|
port=8080,
|
|
)
|
|
db_session.add(instance)
|
|
await db_session.commit()
|
|
|
|
received: list[Any] = []
|
|
|
|
def subscriber(payload: InstanceEventPayload) -> None:
|
|
received.append(payload)
|
|
|
|
event_bus.subscribe("instance.created", subscriber)
|
|
|
|
from src.services.lifecycle_hooks import publish_lifecycle_event
|
|
|
|
await publish_lifecycle_event(
|
|
event_bus=event_bus,
|
|
session=db_session,
|
|
instance=instance,
|
|
event_type="instance.created",
|
|
created_by=user_id,
|
|
status="pending",
|
|
message="Instance created",
|
|
)
|
|
|
|
assert len(received) == 1
|
|
assert received[0]["event"] == "instance.created"
|
|
|
|
result = await db_session.execute(
|
|
select(InstanceEvent).where(InstanceEvent.instance_id == instance.id)
|
|
)
|
|
rows = result.scalars().all()
|
|
assert len(rows) == 1
|
|
assert rows[0].event_type == "created"
|
|
assert rows[0].created_by == user_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_lifecycle_event_persists_audit_row(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
) -> None:
|
|
"""Test that publishing a lifecycle event persists an audit row."""
|
|
user_id = _get_user_id_from_client(authenticated_client)
|
|
assert user_id is not None
|
|
|
|
project = Project(
|
|
id=uuid.uuid4(),
|
|
name="test-project",
|
|
description="Test",
|
|
owner_id=user_id,
|
|
)
|
|
repo = GitRepository(
|
|
id=uuid.uuid4(),
|
|
name="test-repo",
|
|
path="/tmp/test-repo",
|
|
project_id=project.id,
|
|
owner_id=user_id,
|
|
remote_url="https://github.com/test/repo.git",
|
|
)
|
|
tool_type = ToolType(
|
|
id=uuid.uuid4(),
|
|
name="test-tool-2",
|
|
display_name="Test Tool 2",
|
|
category="other",
|
|
interface_type="web",
|
|
requires_port=True,
|
|
default_port=8080,
|
|
definition_type="legacy",
|
|
compose_template="version: '3.8'\nservices:\n app:\n image: alpine\n command: sleep 3600\n",
|
|
)
|
|
db_session.add_all([project, repo, tool_type])
|
|
await db_session.commit()
|
|
|
|
instance = ToolInstance(
|
|
id=uuid.uuid4(),
|
|
name="test-instance",
|
|
display_name="Test Instance",
|
|
tool_type_id=tool_type.id,
|
|
repository_id=repo.id,
|
|
project_id=project.id,
|
|
owner_id=user_id,
|
|
status="running",
|
|
compose_path="/tmp/test-compose.yml",
|
|
port=8080,
|
|
)
|
|
db_session.add(instance)
|
|
await db_session.commit()
|
|
|
|
from src.services.lifecycle_hooks import publish_lifecycle_event
|
|
|
|
await publish_lifecycle_event(
|
|
event_bus=event_bus,
|
|
session=db_session,
|
|
instance=instance,
|
|
event_type="instance.stopped",
|
|
created_by=user_id,
|
|
status="stopped",
|
|
message="Instance stopped",
|
|
)
|
|
|
|
result = await db_session.execute(
|
|
select(InstanceEvent).where(InstanceEvent.instance_id == instance.id)
|
|
)
|
|
rows = result.scalars().all()
|
|
assert len(rows) == 1
|
|
assert rows[0].event_type == "stopped"
|
|
assert rows[0].status == "stopped"
|
|
assert rows[0].created_by == user_id
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_event_bus_pubsub(event_bus: InstanceEventBus) -> None:
|
|
"""Test that the event bus delivers events to subscribers."""
|
|
received: list[InstanceEventPayload] = []
|
|
|
|
def handler(payload: InstanceEventPayload) -> None:
|
|
received.append(payload)
|
|
|
|
event_bus.subscribe("test.event", handler)
|
|
|
|
payload: InstanceEventPayload = {
|
|
"event": "test.event",
|
|
"instance_id": str(uuid.uuid4()),
|
|
"status": "running",
|
|
"message": "Test",
|
|
"metadata": {},
|
|
"timestamp": "2026-05-28T12:00:00Z",
|
|
"correlation_id": str(uuid.uuid4()),
|
|
}
|
|
|
|
asyncio.run(event_bus.publish("test.event", payload))
|
|
|
|
assert len(received) == 1
|
|
assert received[0]["event"] == "test.event"
|