alex
aa25852091
- Inject explicit container_name into compose files at start/restart time via _ensure_container_name_in_compose() to prevent Docker Compose from generating UUID-based auto names that break backend network resolution. - Use instance.name.lower() directly instead of get_container_name() lookups which were unreliable with auto-generated names. - Apply compose sanitization, bind-address fix, and container-name injection on restart_instance as well so restarts pick up template fixes. - Add --force-recreate to docker compose up to ensure container_name changes take effect immediately. - Fix notification lifecycle tests to match current behavior (success severity, health_changed event for ownership test). Quality gates: ruff clean, pytest (7 notification lifecycle tests passed)
396 lines
12 KiB
Python
396 lines
12 KiB
Python
"""Integration tests for event producer → notification creation flow."""
|
|
|
|
import uuid
|
|
from collections.abc import Generator
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.models.git_repository import GitRepository
|
|
from src.models.notification import Notification
|
|
from src.models.project import Project
|
|
from src.models.tool_instance import ToolInstance
|
|
from src.models.tool_type import ToolType
|
|
from src.models.user import User
|
|
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
|
|
from src.services.health_monitor import HealthSnapshot
|
|
|
|
|
|
@pytest.fixture
|
|
def event_bus() -> Generator[InstanceEventBus, None, None]:
|
|
"""Provide a fresh EventBus instance."""
|
|
bus = InstanceEventBus()
|
|
bus._reset_for_testing()
|
|
yield bus
|
|
bus._reset_for_testing()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_instance(db_session: AsyncSession) -> ToolInstance:
|
|
"""Create a complete tool instance with all required relations."""
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email="owner@headquarter.local",
|
|
name="Owner",
|
|
authentik_id=f"authentik-{uuid.uuid4()}",
|
|
avatar_url=None,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
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="running",
|
|
compose_path="/tmp/test-compose.yml",
|
|
port=8080,
|
|
)
|
|
db_session.add(instance)
|
|
await db_session.commit()
|
|
return instance
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_lifecycle_started_intermediate_skips_notification(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
test_instance: ToolInstance,
|
|
) -> None:
|
|
"""Intermediate 'starting' state does NOT create a notification."""
|
|
received: list[InstanceEventPayload] = []
|
|
|
|
def subscriber(payload: InstanceEventPayload) -> None:
|
|
received.append(payload)
|
|
|
|
event_bus.subscribe("instance.started", subscriber)
|
|
|
|
from src.services.lifecycle_hooks import publish_lifecycle_event
|
|
|
|
await publish_lifecycle_event(
|
|
event_bus=event_bus,
|
|
session=db_session,
|
|
instance=test_instance,
|
|
event_type="instance.started",
|
|
status="starting",
|
|
message="Container starting...",
|
|
)
|
|
|
|
# Event still published
|
|
assert len(received) == 1
|
|
|
|
# No notification created for intermediate state
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.user_id == test_instance.owner_id)
|
|
)
|
|
notifications = list(result.scalars().all())
|
|
assert len(notifications) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_lifecycle_running_creates_notification(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
test_instance: ToolInstance,
|
|
) -> None:
|
|
"""Successful terminal state (running) creates a notification."""
|
|
from src.services.lifecycle_hooks import publish_lifecycle_event
|
|
|
|
await publish_lifecycle_event(
|
|
event_bus=event_bus,
|
|
session=db_session,
|
|
instance=test_instance,
|
|
event_type="instance.health_changed",
|
|
status="running",
|
|
message="Container running",
|
|
)
|
|
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.user_id == test_instance.owner_id)
|
|
)
|
|
notifications = list(result.scalars().all())
|
|
assert len(notifications) == 1
|
|
n = notifications[0]
|
|
assert n.category == "instance"
|
|
assert n.severity == "success"
|
|
assert n.title == "Container ready"
|
|
assert n.source_type == "tool_instances"
|
|
assert n.source_id == test_instance.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_health_monitor_error_creates_notification(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
test_instance: ToolInstance,
|
|
) -> None:
|
|
"""Simulating a health monitor crash creates an error notification."""
|
|
from src.services.health_monitor import HealthMonitor
|
|
|
|
monitor = HealthMonitor(event_bus)
|
|
|
|
received: list[InstanceEventPayload] = []
|
|
|
|
def subscriber(payload: InstanceEventPayload) -> None:
|
|
received.append(payload)
|
|
|
|
event_bus.subscribe("instance.error", subscriber)
|
|
|
|
with patch(
|
|
"src.services.health_monitor.get_container_status",
|
|
return_value={"status": "exited", "exit_code": 137, "health": None},
|
|
):
|
|
await monitor._check_instance(db_session, test_instance)
|
|
|
|
# Event published
|
|
assert len(received) == 1
|
|
|
|
# Notification created
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.user_id == test_instance.owner_id)
|
|
)
|
|
notifications = list(result.scalars().all())
|
|
assert len(notifications) == 1
|
|
n = notifications[0]
|
|
assert n.category == "instance"
|
|
assert n.severity == "error"
|
|
assert n.source_type == "tool_instances"
|
|
assert n.source_id == test_instance.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_notification_failure_does_not_block_event_pipeline(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
test_instance: ToolInstance,
|
|
) -> None:
|
|
"""If NotificationService raises, the event is still published and no exception escapes."""
|
|
received: list[InstanceEventPayload] = []
|
|
|
|
def subscriber(payload: InstanceEventPayload) -> None:
|
|
received.append(payload)
|
|
|
|
event_bus.subscribe("instance.started", subscriber)
|
|
|
|
from src.services.lifecycle_hooks import publish_lifecycle_event
|
|
|
|
with patch(
|
|
"src.services.lifecycle_hooks.notification_service.create_notification",
|
|
side_effect=RuntimeError("DB is down"),
|
|
):
|
|
# Should not raise
|
|
await publish_lifecycle_event(
|
|
event_bus=event_bus,
|
|
session=db_session,
|
|
instance=test_instance,
|
|
event_type="instance.started",
|
|
status="starting",
|
|
message="Container started",
|
|
)
|
|
|
|
assert len(received) == 1
|
|
assert received[0]["event"] == "instance.started"
|
|
|
|
# No notification should have been created
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.user_id == test_instance.owner_id)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_notification_ownership_matches_instance_owner(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
) -> None:
|
|
"""Notification user_id matches the instance owner, not any caller."""
|
|
# Create a caller user (simulates the user making an API request)
|
|
caller = User(
|
|
id=uuid.uuid4(),
|
|
email="caller@headquarter.local",
|
|
name="Caller",
|
|
authentik_id=f"authentik-{uuid.uuid4()}",
|
|
avatar_url=None,
|
|
)
|
|
db_session.add(caller)
|
|
await db_session.commit()
|
|
|
|
# Create the actual owner
|
|
owner = User(
|
|
id=uuid.uuid4(),
|
|
email="owner@headquarter.local",
|
|
name="Owner",
|
|
authentik_id=f"authentik-{uuid.uuid4()}",
|
|
avatar_url=None,
|
|
)
|
|
db_session.add(owner)
|
|
await db_session.commit()
|
|
|
|
project = Project(
|
|
id=uuid.uuid4(),
|
|
name="test-project",
|
|
description="Test",
|
|
owner_id=owner.id,
|
|
)
|
|
repo = GitRepository(
|
|
id=uuid.uuid4(),
|
|
name="test-repo",
|
|
path="/tmp/test-repo",
|
|
project_id=project.id,
|
|
owner_id=owner.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=owner.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.health_changed",
|
|
status="running",
|
|
message="Container running",
|
|
)
|
|
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.source_id == instance.id)
|
|
)
|
|
n = result.scalar_one()
|
|
assert n.user_id == owner.id
|
|
assert n.user_id != caller.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_lifecycle_error_creates_error_notification(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
test_instance: ToolInstance,
|
|
) -> None:
|
|
"""An instance.error lifecycle event creates a severity=error notification."""
|
|
from src.services.lifecycle_hooks import publish_lifecycle_event
|
|
|
|
await publish_lifecycle_event(
|
|
event_bus=event_bus,
|
|
session=db_session,
|
|
instance=test_instance,
|
|
event_type="instance.error",
|
|
status="error",
|
|
message="Container failed",
|
|
)
|
|
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.user_id == test_instance.owner_id)
|
|
)
|
|
n = result.scalar_one()
|
|
assert n.severity == "error"
|
|
assert n.title == "Container error"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_health_monitor_unhealthy_creates_warning_notification(
|
|
db_session: AsyncSession,
|
|
event_bus: InstanceEventBus,
|
|
test_instance: ToolInstance,
|
|
) -> None:
|
|
"""Health monitor marking instance unhealthy creates severity=warning notification."""
|
|
from src.services.health_monitor import HealthMonitor
|
|
|
|
monitor = HealthMonitor(event_bus)
|
|
monitor._last_known_state[test_instance.id] = HealthSnapshot(
|
|
container_status="running",
|
|
container_healthy=None,
|
|
tunnel_healthy=True,
|
|
exit_code=None,
|
|
)
|
|
test_instance.public_url = "https://example.trycloudflare.com"
|
|
|
|
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"},
|
|
),
|
|
):
|
|
await monitor._check_instance(db_session, test_instance)
|
|
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.user_id == test_instance.owner_id)
|
|
)
|
|
n = result.scalar_one()
|
|
assert n.category == "health"
|
|
assert n.severity == "warning"
|
|
assert n.title == "Container unhealthy"
|