2026-06-03 21:53:06 +02:00
|
|
|
"""Notifications integration fixtures.
|
|
|
|
|
|
2026-06-20 03:11:00 +05:30
|
|
|
The app's DB session and auth-context dependencies are overridden to ride the
|
2026-06-03 21:53:06 +02:00
|
|
|
test's transactional `db_session`, so API calls and seeded rows share one
|
2026-06-20 03:11:00 +05:30
|
|
|
transaction that rolls back per test. Overriding `get_auth_context` also bypasses
|
|
|
|
|
real JWT auth, so these tests don't depend on AUTH_TYPE.
|
2026-06-03 21:53:06 +02:00
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
|
|
|
|
|
|
import httpx
|
|
|
|
|
import pytest
|
|
|
|
|
import pytest_asyncio
|
|
|
|
|
from httpx import ASGITransport
|
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
|
|
|
|
from app.app import app, limiter
|
2026-06-20 03:11:00 +05:30
|
|
|
from app.auth.context import AuthContext
|
2026-06-03 21:53:06 +02:00
|
|
|
from app.db import User, get_async_session
|
2026-06-20 03:11:00 +05:30
|
|
|
from app.users import get_auth_context
|
2026-06-03 21:53:06 +02:00
|
|
|
|
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
|
|
|
|
limiter.enabled = False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
|
|
|
async def client(
|
|
|
|
|
db_session: AsyncSession,
|
|
|
|
|
db_user: User,
|
|
|
|
|
) -> AsyncGenerator[httpx.AsyncClient, None]:
|
|
|
|
|
async def override_session() -> AsyncGenerator[AsyncSession, None]:
|
|
|
|
|
yield db_session
|
|
|
|
|
|
2026-06-20 03:11:00 +05:30
|
|
|
async def override_auth() -> AuthContext:
|
|
|
|
|
return AuthContext.session(db_user)
|
2026-06-03 21:53:06 +02:00
|
|
|
|
|
|
|
|
previous_overrides = app.dependency_overrides.copy()
|
|
|
|
|
app.dependency_overrides[get_async_session] = override_session
|
2026-06-20 03:11:00 +05:30
|
|
|
app.dependency_overrides[get_auth_context] = override_auth
|
2026-06-03 21:53:06 +02:00
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
async with httpx.AsyncClient(
|
|
|
|
|
transport=ASGITransport(app=app),
|
|
|
|
|
base_url="http://test",
|
|
|
|
|
timeout=30.0,
|
|
|
|
|
follow_redirects=False,
|
|
|
|
|
) as test_client:
|
|
|
|
|
yield test_client
|
|
|
|
|
finally:
|
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
app.dependency_overrides.update(previous_overrides)
|