Developer
81b9a66ef5
- Delete 4 obsolete unit tests tied to removed git mount/clone models - Update imports and assertions across unit/integration/service tests - Fix Settings defaults (postgres host, JWT props, cookie_samesite) - Add skip guards for PostgreSQL-dependent integration tests - Fix GitService env assertions and HealthMonitor state-change tests - Repair docker/container inspect assertions in test_docker_service - Fix ToolTypeCreate default_port validator ordering bug - Fix check_port_exposed substring false-positive for port 0 - Update test_tool_types_api_extended to use interface_type field Quality gates: pytest 311 passed, 34 skipped; npm typecheck/lint/test 87 passed
388 lines
11 KiB
Python
388 lines
11 KiB
Python
"""Unit tests for NotificationService."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.models.system.notification import Notification
|
|
from src.models.user.user import User
|
|
from src.services.shared.notification_service import NotificationService
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_service() -> NotificationService:
|
|
return NotificationService()
|
|
|
|
|
|
@pytest.fixture
|
|
async def user_a(db_session: AsyncSession) -> User:
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email="user-a@headquarter.local",
|
|
name="User A",
|
|
authentik_id=f"authentik-{uuid.uuid4()}",
|
|
avatar_url=None,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
async def user_b(db_session: AsyncSession) -> User:
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email="user-b@headquarter.local",
|
|
name="User B",
|
|
authentik_id=f"authentik-{uuid.uuid4()}",
|
|
avatar_url=None,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
return user
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_create_notification(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
notification = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Container started",
|
|
message="Instance is running",
|
|
source_type="tool_instances",
|
|
source_id=uuid.uuid4(),
|
|
metadata={"key": "value"},
|
|
)
|
|
|
|
assert notification.user_id == user_a.id
|
|
assert notification.category == "instance"
|
|
assert notification.severity == "info"
|
|
assert notification.title == "Container started"
|
|
assert notification.message == "Instance is running"
|
|
assert notification.source_type == "tool_instances"
|
|
assert notification.notification_metadata == {"key": "value"}
|
|
assert notification.read_at is None
|
|
assert notification.dismissed_at is None
|
|
assert notification.created_at is not None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_list_notifications_orders_by_created_at_desc(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
n1 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="First"
|
|
)
|
|
n1.created_at = datetime.now(timezone.utc) - timedelta(seconds=2)
|
|
await db_session.commit()
|
|
await db_session.refresh(n1)
|
|
|
|
n2 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Second"
|
|
)
|
|
n2.created_at = datetime.now(timezone.utc) - timedelta(seconds=1)
|
|
await db_session.commit()
|
|
await db_session.refresh(n2)
|
|
|
|
n3 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Third"
|
|
)
|
|
|
|
items, total = await notification_service.list_notifications(db_session, user_a.id)
|
|
|
|
assert total == 3
|
|
assert [item.id for item in items] == [n3.id, n2.id, n1.id]
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_list_notifications_excludes_dismissed(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
n1 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Visible"
|
|
)
|
|
n2 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Dismissed"
|
|
)
|
|
await notification_service.dismiss(db_session, n2.id, user_a.id)
|
|
|
|
items, total = await notification_service.list_notifications(db_session, user_a.id)
|
|
|
|
assert total == 1
|
|
assert items[0].id == n1.id
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_list_notifications_unread_only(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
n1 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Unread"
|
|
)
|
|
n2 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Read"
|
|
)
|
|
await notification_service.mark_read(db_session, n2.id, user_a.id)
|
|
|
|
items, total = await notification_service.list_notifications(
|
|
db_session, user_a.id, unread_only=True
|
|
)
|
|
|
|
assert total == 1
|
|
assert items[0].id == n1.id
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_get_unread_count(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
for i in range(5):
|
|
n = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title=f"Notification {i}",
|
|
)
|
|
if i >= 3:
|
|
await notification_service.mark_read(db_session, n.id, user_a.id)
|
|
|
|
count = await notification_service.get_unread_count(db_session, user_a.id)
|
|
|
|
assert count == 3
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_mark_read_sets_read_at(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
n = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Unread"
|
|
)
|
|
|
|
updated = await notification_service.mark_read(db_session, n.id, user_a.id)
|
|
|
|
assert updated.read_at is not None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_mark_all_read_affects_all_unread(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
for i in range(4):
|
|
await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title=f"Notification {i}",
|
|
)
|
|
|
|
marked = await notification_service.mark_all_read(db_session, user_a.id)
|
|
|
|
assert marked == 4
|
|
count = await notification_service.get_unread_count(db_session, user_a.id)
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_dismiss_sets_dismissed_at(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
n = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="To dismiss"
|
|
)
|
|
|
|
await notification_service.dismiss(db_session, n.id, user_a.id)
|
|
|
|
result = await db_session.execute(
|
|
select(Notification).where(Notification.id == n.id)
|
|
)
|
|
row = result.scalar_one()
|
|
assert row.dismissed_at is not None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_mark_read_wrong_owner_raises(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
n = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Owned by A"
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Notification not found"):
|
|
await notification_service.mark_read(db_session, n.id, user_b.id)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_dismiss_wrong_owner_raises(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
n = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Owned by A"
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Notification not found"):
|
|
await notification_service.dismiss(db_session, n.id, user_b.id)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_list_notifications_mute_categories(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="Instance"
|
|
)
|
|
n2 = await notification_service.create_notification(
|
|
db_session, user_a.id, category="system", severity="info", title="System"
|
|
)
|
|
|
|
items, total = await notification_service.list_notifications(
|
|
db_session, user_a.id, mute_categories=["instance"]
|
|
)
|
|
|
|
assert total == 1
|
|
assert items[0].id == n2.id
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_get_unread_count_excludes_dismissed(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
n = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Unread dismissed",
|
|
)
|
|
await notification_service.dismiss(db_session, n.id, user_a.id)
|
|
|
|
count = await notification_service.get_unread_count(db_session, user_a.id)
|
|
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_dismiss_all_affects_all_non_dismissed(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
for i in range(4):
|
|
await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title=f"Notification {i}",
|
|
)
|
|
|
|
cleared = await notification_service.dismiss_all(db_session, user_a.id)
|
|
|
|
assert cleared == 4
|
|
items, total = await notification_service.list_notifications(db_session, user_a.id)
|
|
assert total == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_dismiss_all_affects_only_caller(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
for i in range(3):
|
|
await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title=f"A-{i}"
|
|
)
|
|
for i in range(2):
|
|
await notification_service.create_notification(
|
|
db_session, user_b.id, category="instance", severity="info", title=f"B-{i}"
|
|
)
|
|
|
|
cleared = await notification_service.dismiss_all(db_session, user_a.id)
|
|
|
|
assert cleared == 3
|
|
items_a, total_a = await notification_service.list_notifications(
|
|
db_session, user_a.id
|
|
)
|
|
items_b, total_b = await notification_service.list_notifications(
|
|
db_session, user_b.id
|
|
)
|
|
assert total_a == 0
|
|
assert total_b == 2
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_mark_all_read_affects_only_caller(
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
for i in range(3):
|
|
await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title=f"A-{i}"
|
|
)
|
|
for i in range(2):
|
|
await notification_service.create_notification(
|
|
db_session, user_b.id, category="instance", severity="info", title=f"B-{i}"
|
|
)
|
|
|
|
marked = await notification_service.mark_all_read(db_session, user_a.id)
|
|
|
|
assert marked == 3
|
|
count_a = await notification_service.get_unread_count(db_session, user_a.id)
|
|
count_b = await notification_service.get_unread_count(db_session, user_b.id)
|
|
assert count_a == 0
|
|
assert count_b == 2
|