Test auth flows and switch core tests to session auth
New auth suite covers registration, login (incl. wrong-password), email verification, password reset (old sessions + old password rejected), logout revocation, and no-enumeration on reset. Core tenancy tests now authenticate via real sessions. A capturing mailer makes email flows assertable. 13 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Justin Paul <justin@jpaul.me>
This commit is contained in:
@@ -3,28 +3,49 @@
|
||||
DB-backed tests require ``TEST_DATABASE_URL`` (an async URL to a *disposable*
|
||||
Postgres); without it they skip, so the no-DB unit suite still runs anywhere.
|
||||
The schema is built from the models via ``create_all`` and dropped per test for
|
||||
isolation.
|
||||
isolation. A capturing mailer replaces the real one so email flows are testable.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
import app.models # noqa: F401 — register all models on Base.metadata
|
||||
from app.api.deps import get_mailer
|
||||
from app.core.db import get_session
|
||||
from app.integrations.mailer.base import Mailer
|
||||
from app.main import app
|
||||
from app.models import Base
|
||||
|
||||
TEST_DATABASE_URL = os.getenv("TEST_DATABASE_URL")
|
||||
|
||||
|
||||
class CapturingMailer(Mailer):
|
||||
def __init__(self) -> None:
|
||||
self.verifications: list[tuple[str, str]] = []
|
||||
self.resets: list[tuple[str, str]] = []
|
||||
|
||||
async def send_email_verification(self, *, to: str, link: str) -> None:
|
||||
self.verifications.append((to, link))
|
||||
|
||||
async def send_password_reset(self, *, to: str, link: str) -> None:
|
||||
self.resets.append((to, link))
|
||||
|
||||
|
||||
_mailer = CapturingMailer()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mailer() -> CapturingMailer:
|
||||
return _mailer
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client():
|
||||
if not TEST_DATABASE_URL:
|
||||
import pytest
|
||||
|
||||
pytest.skip("TEST_DATABASE_URL not set")
|
||||
|
||||
engine = create_async_engine(TEST_DATABASE_URL)
|
||||
@@ -38,10 +59,31 @@ async def client():
|
||||
async with sessionmaker() as session:
|
||||
yield session
|
||||
|
||||
_mailer.verifications.clear()
|
||||
_mailer.resets.clear()
|
||||
app.dependency_overrides[get_session] = _override_session
|
||||
app.dependency_overrides[get_mailer] = lambda: _mailer
|
||||
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as http_client:
|
||||
yield http_client
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
def token_from_link(link: str) -> str:
|
||||
return link.split("token=", 1)[1]
|
||||
|
||||
|
||||
async def register(client, email: str, password: str = "password123") -> str:
|
||||
"""Register a user and return their bearer session token."""
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/register", json={"email": email, "password": password}
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["token"]
|
||||
|
||||
|
||||
def auth(token: str) -> dict:
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
Reference in New Issue
Block a user