justin
84a743f5b9
can_view_tree() now distinguishes anonymous vs authenticated non-members so the
four-level model is enforceable:
- public / unlisted → anyone, including anonymous (unlisted gated only by the
link, so the API must never *list* it)
- site_members → any authenticated account (denies anonymous)
- private → members only
Members (any role) always view; soft-deleted trees stay hidden from everyone.
person_visibility (living-person redaction) is unchanged.
Tests: a full can_view_tree matrix across {anonymous, logged-in non-member,
member} × {public, unlisted, site_members, private}, plus deleted-tree-hidden
and the site_members anon-vs-logged-in case. Adds `engine`/`db_session` fixtures
(refactored out of `client`) so the engine can be unit-tested directly,
including the anonymous path that has no HTTP endpoint yet.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Justin Paul <justin@jpaul.me>
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
"""Test fixtures.
|
|
|
|
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. 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 import text
|
|
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, get_objectstore
|
|
from app.core.db import get_session
|
|
from app.integrations.mailer.base import Mailer
|
|
from app.integrations.objectstore.base import ObjectStore
|
|
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))
|
|
|
|
|
|
class FakeObjectStore(ObjectStore):
|
|
def __init__(self) -> None:
|
|
self.objects: dict[str, tuple[bytes, str]] = {}
|
|
|
|
async def ensure_bucket(self) -> None:
|
|
pass
|
|
|
|
async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
|
|
self.objects[key] = (data, content_type)
|
|
|
|
async def get_object(self, *, key: str) -> bytes:
|
|
return self.objects[key][0]
|
|
|
|
async def presigned_get_url(self, *, key: str) -> str:
|
|
return f"https://objects.test/{key}"
|
|
|
|
async def delete_object(self, *, key: str) -> None:
|
|
self.objects.pop(key, None)
|
|
|
|
|
|
_mailer = CapturingMailer()
|
|
_store = FakeObjectStore()
|
|
|
|
|
|
@pytest.fixture
|
|
def mailer() -> CapturingMailer:
|
|
return _mailer
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def engine():
|
|
if not TEST_DATABASE_URL:
|
|
pytest.skip("TEST_DATABASE_URL not set")
|
|
|
|
eng = create_async_engine(TEST_DATABASE_URL)
|
|
async with eng.begin() as conn:
|
|
await conn.execute(text("CREATE EXTENSION IF NOT EXISTS pg_trgm"))
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield eng
|
|
await eng.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client(engine):
|
|
sessionmaker = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
|
|
|
|
async def _override_session():
|
|
async with sessionmaker() as session:
|
|
yield session
|
|
|
|
_mailer.verifications.clear()
|
|
_mailer.resets.clear()
|
|
_store.objects.clear()
|
|
app.dependency_overrides[get_session] = _override_session
|
|
app.dependency_overrides[get_mailer] = lambda: _mailer
|
|
app.dependency_overrides[get_objectstore] = lambda: _store
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as http_client:
|
|
yield http_client
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(engine):
|
|
"""A raw AsyncSession on the test DB, for unit-testing services directly."""
|
|
sessionmaker = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
|
|
async with sessionmaker() as session:
|
|
yield session
|
|
|
|
|
|
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}"}
|