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
327 lines
9.1 KiB
Python
327 lines
9.1 KiB
Python
"""Integration tests for notifications API."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.models.user.user import User
|
|
from src.models.user.user_config import UserConfig
|
|
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
|
|
|
|
|
|
def _mint_cookie_for_user(test_client: TestClient, user_id: uuid.UUID) -> None:
|
|
from src.auth.session import create_session_cookie
|
|
from src.config import Settings
|
|
|
|
settings = Settings()
|
|
cookie = create_session_cookie(
|
|
settings=settings,
|
|
user_id=str(user_id),
|
|
)
|
|
test_client.cookies.set("session", cookie)
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_list_requires_auth(test_client: TestClient) -> None:
|
|
response = test_client.get("/notifications")
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_list_returns_only_own_notifications(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
async def create_notifications() -> None:
|
|
await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="A"
|
|
)
|
|
await notification_service.create_notification(
|
|
db_session, user_b.id, category="instance", severity="info", title="B"
|
|
)
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(create_notifications())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.get("/notifications")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 1
|
|
assert data["items"][0]["title"] == "A"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_list_pagination(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
async def create_many() -> None:
|
|
for i in range(25):
|
|
n = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title=f"Notification {i}",
|
|
)
|
|
n.created_at = datetime.now(timezone.utc) - timedelta(seconds=i)
|
|
await db_session.commit()
|
|
await db_session.refresh(n)
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(create_many())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.get("/notifications?limit=10&offset=10")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 10
|
|
assert data["total"] == 25
|
|
assert data["limit"] == 10
|
|
assert data["offset"] == 10
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_unread_count_endpoint(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
async def create_unread() -> None:
|
|
for _ in range(3):
|
|
await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Unread",
|
|
)
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(create_unread())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.get("/notifications/unread")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["count"] == 3
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_mark_read_endpoint(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
async def create_and_get() -> uuid.UUID:
|
|
n = await notification_service.create_notification(
|
|
db_session, user_a.id, category="instance", severity="info", title="To read"
|
|
)
|
|
return n.id
|
|
|
|
import asyncio
|
|
|
|
nid = asyncio.run(create_and_get())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.patch(f"/notifications/{nid}/read")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["read_at"] is not None
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_mark_read_404_for_other_user(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
async def create_and_get() -> uuid.UUID:
|
|
n = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Owned by A",
|
|
)
|
|
return n.id
|
|
|
|
import asyncio
|
|
|
|
nid = asyncio.run(create_and_get())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_b.id)
|
|
response = authenticated_client.patch(f"/notifications/{nid}/read")
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_mark_all_read_endpoint(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
async def create_unread() -> None:
|
|
for _ in range(4):
|
|
await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Unread",
|
|
)
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(create_unread())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.post("/notifications/mark-all-read")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["marked_count"] == 4
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_dismiss_endpoint(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
async def create_and_get() -> uuid.UUID:
|
|
n = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="To dismiss",
|
|
)
|
|
return n.id
|
|
|
|
import asyncio
|
|
|
|
nid = asyncio.run(create_and_get())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.delete(f"/notifications/{nid}")
|
|
assert response.status_code == 204
|
|
|
|
response = authenticated_client.get("/notifications")
|
|
data = response.json()
|
|
assert len(data["items"]) == 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_dismiss_404_for_other_user(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
user_b: User,
|
|
) -> None:
|
|
async def create_and_get() -> uuid.UUID:
|
|
n = await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Owned by A",
|
|
)
|
|
return n.id
|
|
|
|
import asyncio
|
|
|
|
nid = asyncio.run(create_and_get())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_b.id)
|
|
response = authenticated_client.delete(f"/notifications/{nid}")
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_mute_categories_filter_in_list(
|
|
authenticated_client: TestClient,
|
|
db_session: AsyncSession,
|
|
notification_service: NotificationService,
|
|
user_a: User,
|
|
) -> None:
|
|
async def setup() -> None:
|
|
config = UserConfig(
|
|
user_id=user_a.id, config={"notification_mute_categories": ["instance"]}
|
|
)
|
|
db_session.add(config)
|
|
await db_session.commit()
|
|
|
|
await notification_service.create_notification(
|
|
db_session,
|
|
user_a.id,
|
|
category="instance",
|
|
severity="info",
|
|
title="Instance",
|
|
)
|
|
await notification_service.create_notification(
|
|
db_session, user_a.id, category="system", severity="info", title="System"
|
|
)
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(setup())
|
|
|
|
_mint_cookie_for_user(authenticated_client, user_a.id)
|
|
response = authenticated_client.get("/notifications")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 1
|
|
assert data["items"][0]["title"] == "System"
|