alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote dev branch

Browse Source
This commit is contained in:
Alex Blank
2026-05-29 12:55:03 +02:00
parent 57ff236f2d 6085859874
commit cbd3436ff7
6 changed files with 690 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/api/user_config.py
+10 -2
View File
@@ -14,7 +14,9 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/users/me", tags=["user-config"])
async def _get_or_create_config(session: AsyncSession, user_id: uuid.UUID) -> UserConfig:
async def _get_or_create_config(
session: AsyncSession, user_id: uuid.UUID
) -> UserConfig:
"""Get or create user config record.
Args:
@@ -24,7 +26,9 @@ async def _get_or_create_config(session: AsyncSession, user_id: uuid.UUID) -> Us
Returns:
The user's config, creating a new one if it doesn't exist.
"""
result = await session.execute(select(UserConfig).where(UserConfig.user_id == user_id))
result = await session.execute(
select(UserConfig).where(UserConfig.user_id == user_id)
)
config = result.scalar_one_or_none()
if config is None:
config = UserConfig(user_id=user_id, config={})
@@ -42,6 +46,8 @@ class UserConfigResponse(BaseModel):
git_user_name: str | None = None
git_user_email: str | None = None
last_session_id: str | None = None
notification_mute_categories: list[str] | None = None
notification_toast_level: str | None = None
class UserConfigUpdate(BaseModel):
@@ -50,6 +56,8 @@ class UserConfigUpdate(BaseModel):
git_user_name: str | None = None
git_user_email: str | None = None
last_session_id: str | None = None
notification_mute_categories: list[str] | None = None
notification_toast_level: str | None = None
@router.get(
apps/api/src/services/health_monitor.py
+34
View File
@@ -15,6 +15,7 @@ from src.models.tool_instance import ToolInstance
from src.services.correlation import get_correlation_id
from src.services.docker import check_tunnel_health, get_container_status
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.notification_service import notification_service
logger = logging.getLogger(__name__)
@@ -217,3 +218,36 @@ class HealthMonitor:
}
await self._event_bus.publish(event_type, payload)
# Create notification for instance owner (fire-and-forget)
if new_status == "error":
category = "instance"
severity = "error"
title = "Container failed"
else:
category = "health"
if new_status == "unhealthy":
severity = "warning"
title = "Container unhealthy"
else:
severity = "info"
title = "Container recovered"
try:
await notification_service.create_notification(
session=session,
user_id=instance.owner_id,
category=category,
severity=severity,
title=title,
message=message,
source_type="tool_instances",
source_id=instance.id,
metadata=metadata,
)
except Exception:
logger.exception(
"Failed to create notification for health event %s",
event_type,
extra={"correlation_id": correlation_id},
)
apps/api/src/services/lifecycle_hooks.py
+43
View File
@@ -1,5 +1,6 @@
"""Lifecycle hook helpers for instrumenting tool instance transitions."""
import logging
import uuid
from datetime import datetime, timezone
@@ -9,6 +10,25 @@ from src.models.instance_event import InstanceEvent
from src.models.tool_instance import ToolInstance
from src.services.correlation import get_correlation_id
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.notification_service import notification_service
logger = logging.getLogger(__name__)
def _derive_title(event_type: str) -> str:
"""Map lifecycle event type to a human-readable notification title."""
mapping = {
"instance.created": "Container created",
"instance.started": "Container started",
"instance.stopped": "Container stopped",
"instance.restarted": "Container restarted",
"instance.deleted": "Container deleted",
"instance.error": "Container error",
}
return mapping.get(
event_type,
event_type.replace("instance.", "").replace("_", " ").title(),
)
def _build_payload(
@@ -96,3 +116,26 @@ async def publish_lifecycle_event(
# Publish to bus
await event_bus.publish(event_type, payload)
# Create notification for instance owner (fire-and-forget)
severity = "error" if event_type == "instance.error" else "info"
title = _derive_title(event_type)
try:
await notification_service.create_notification(
session=session,
user_id=instance.owner_id,
category="instance",
severity=severity,
title=title,
message=message,
source_type="tool_instances",
source_id=instance.id,
metadata=metadata,
)
except Exception:
logger.exception(
"Failed to create notification for lifecycle event %s",
event_type,
extra={"correlation_id": payload.get("correlation_id", "unknown")},
)
apps/api/tests/integration/test_notifications_lifecycle.py
+370
View File
@@ -0,0 +1,370 @@
"""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_event_creates_notification(
db_session: AsyncSession,
event_bus: InstanceEventBus,
test_instance: ToolInstance,
) -> None:
"""Triggering a lifecycle event creates a notification for the instance owner."""
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 started",
)
# Event still 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 == "info"
assert n.title == "Container started"
assert n.source_type == "tool_instances"
assert n.source_id == test_instance.id
assert n.message == "Container started"
@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.created",
status="pending",
message="Instance created",
)
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"