"""Shared fixtures. Most of the suite is pure unit tests with no database — they run in seconds and gate every commit. The integration tests in `test_pipeline_integration.py` need real Postgres because what they verify (`FOR UPDATE SKIP LOCKED`, unique-constraint races) does not exist in a fake. They skip cleanly when no database is reachable so `make test` still works offline. """ import os import struct import uuid import zlib from collections.abc import AsyncIterator from datetime import UTC, datetime, time, timedelta from typing import Any import pytest import pytest_asyncio from sqlalchemy import text from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine, ) from app.models import Event, Guest, Invitation, ReminderSchedule, Wedding from app.models.enums import ( EventType, InvitationStatus, Locale, ReminderAudience, ) DATABASE_URL = os.getenv("DATABASE_URL", "") requires_db = pytest.mark.skipif( not DATABASE_URL.startswith("postgresql"), reason="requires a Postgres connection (DATABASE_URL)", ) # ------------------------------------------------------------------ image fixtures #: Link-preview images are measured by reading their header (design D6), so the fixtures are #: hand-built headers rather than real photographs. Shared from here because two suites need #: them to agree on what a well-formed one looks like: the parser tests and the upload tests. def png_bytes(width: int, height: int, *, pad: int = 0) -> bytes: """A structurally valid PNG: signature, IHDR, IEND, plus optional trailing bulk.""" def chunk(kind: bytes, payload: bytes) -> bytes: return ( struct.pack(">I", len(payload)) + kind + payload + struct.pack(">I", zlib.crc32(kind + payload) & 0xFFFFFFFF) ) ihdr = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0) signature = b"\x89PNG\r\n\x1a\n" return signature + chunk(b"IHDR", ihdr) + chunk(b"IEND", b"") + (b"\x00" * pad) def jpeg_bytes(width: int, height: int) -> bytes: """SOI, a comment segment to be stepped over, then an SOF0 frame header. The leading segment is the point: a JPEG in the wild opens with EXIF or a colour profile, so reaching the frame header means walking segments by length rather than searching for a marker byte. """ comment = b"\xff\xfe" + struct.pack(">H", 2 + 8) + b"x" * 8 sof = b"\xff\xc0" + struct.pack(">H", 17) + bytes([8]) + struct.pack(">HH", height, width) sof += b"\x03" + b"\x00" * 9 return b"\xff\xd8" + comment + sof @pytest_asyncio.fixture async def app_engine_per_loop() -> AsyncIterator[None]: """Give a test that drives the app through ASGI an engine bound to its own event loop. `app.db.get_engine` is lru_cached while pytest-asyncio gives every test a fresh loop, so an engine built by an earlier test holds connections belonging to a loop that is now closed; reusing it fails with "Event loop is closed" — a fact about the fixture, not about the code under test. Cleared going in as well as coming out, because in a full-suite run the cache is already poisoned before the first ASGI test starts. The stale engine is dropped rather than disposed: disposing it would touch the dead loop and raise the very error being avoided. """ from app import db db.get_engine.cache_clear() db.get_session_factory.cache_clear() yield await db.get_engine().dispose() db.get_engine.cache_clear() db.get_session_factory.cache_clear() @pytest_asyncio.fixture async def engine() -> AsyncIterator[AsyncEngine]: eng = create_async_engine(DATABASE_URL, pool_pre_ping=True) yield eng await eng.dispose() @pytest_asyncio.fixture async def sessions(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]: """Factory, not a session — the concurrency tests need two independent connections.""" return async_sessionmaker(engine, expire_on_commit=False) @pytest_asyncio.fixture async def session(sessions: async_sessionmaker[AsyncSession]) -> AsyncIterator[AsyncSession]: async with sessions() as s: yield s @pytest_asyncio.fixture async def super_admin(session: AsyncSession) -> AsyncIterator[uuid.UUID]: """Authenticate ASGI tests by overriding the guard rather than minting a cookie. Authorization itself is covered by `test_admin_route_guards`; what these tests exercise is what an endpoint does once past the guard, so the guard is stubbed. The admin row is real, not invented. Every write records an audit entry carrying `admin_user_id`, and that column is a foreign key — a fabricated id fails the insert and the test then reports a broken fixture as a broken endpoint. """ from app.main import app from app.models.enums import AdminRole from app.services.auth import CurrentAdmin, get_current_admin admin_id = uuid.uuid4() email = f"test-admin-{admin_id.hex[:8]}@example.com" await session.execute( text( "INSERT INTO admin_user (id, email, name, role, status, auth_method, " " must_change_password, failed_login_count) " "VALUES (:id, :email, 'Test Admin', 'super_admin', 'active', 'google', false, 0)" ), {"id": admin_id, "email": email}, ) await session.commit() app.dependency_overrides[get_current_admin] = lambda: CurrentAdmin( id=admin_id, email=email, name="Test Admin", role=AdminRole.SUPER_ADMIN ) yield admin_id app.dependency_overrides.pop(get_current_admin, None) # Audit rows reference the admin, so they go first or the delete is blocked. await session.execute(text("DELETE FROM audit_log WHERE admin_user_id = :id"), {"id": admin_id}) await session.execute(text("DELETE FROM admin_user WHERE id = :id"), {"id": admin_id}) await session.commit() async def make_owner_admin( session: AsyncSession, *, role: str = "super_admin", tag: str | None = None ) -> uuid.UUID: """Insert an admin row to own a test event (add-admin-access-control D1). `event.owner_admin_id` is NOT NULL with an ON DELETE RESTRICT foreign key, so a test that builds an Event needs a real admin first — a fabricated id fails the insert and the test then reports a broken fixture as a broken endpoint. Shared from here because four suites need one and they must agree on the columns; the row is written as raw SQL so a later model change surfaces as one failure here rather than as four mystifying ones spread across the suite. """ admin_id = uuid.uuid4() suffix = tag or admin_id.hex[:8] await session.execute( text( "INSERT INTO admin_user (id, email, name, role, status, auth_method, " " must_change_password, failed_login_count, session_epoch) " "VALUES (:id, :email, 'Test Owner', :role, 'active', 'google', false, 0, 0)" ), {"id": admin_id, "email": f"owner-{suffix}@example.test", "role": role}, ) return admin_id async def forget_admin(session: AsyncSession, admin_id: uuid.UUID) -> None: """Remove a test admin and the rows that reference it, in an order the FKs allow.""" await session.execute(text("DELETE FROM audit_log WHERE admin_user_id = :id"), {"id": admin_id}) await session.execute( text("UPDATE message_job SET sent_by_admin_id = NULL WHERE sent_by_admin_id = :id"), {"id": admin_id}, ) await session.execute(text("DELETE FROM admin_user WHERE id = :id"), {"id": admin_id}) @pytest_asyncio.fixture async def fixture_event(session: AsyncSession) -> AsyncIterator[dict[str, Any]]: """An isolated wedding + event + accepted guests, torn down afterwards. Everything is namespaced with a uuid so a run against the dev database with seed data already in it cannot collide, and so a failed run leaves nothing behind that breaks the next one. Deleting the wedding cascades to events, guests, invitations and jobs. """ tag = uuid.uuid4().hex[:8] # Every event needs an owner (add-admin-access-control D1), and the fixture makes its # own rather than borrowing the `super_admin` fixture's — the two are used together, and # a shared owner would make "a Host cannot reach another owner's event" untestable. owner_id = await make_owner_admin(session, tag=f"fixture-{tag}") wedding = Wedding( bride_name="Test", groom_name="Case", slug=f"test-{tag}", timezone="Asia/Dhaka" ) session.add(wedding) await session.flush() # Plain values, not ORM attributes: teardown rolls back first, and rollback expires # every instance regardless of expire_on_commit, so `wedding.id` there would issue a # lazy SELECT outside the async context and blow up with MissingGreenlet. wedding_id = wedding.id event = Event( wedding_id=wedding.id, owner_admin_id=owner_id, type=EventType.WALIMA, slug=f"walima-{tag}", title_bn="ওয়ালিমা", title_en="Walima", # Far enough out that the T-15 wave is still in the future. starts_at=datetime.now(UTC) + timedelta(days=40), venue_name="Test Hall", venue_address="Dhaka", host_name_1="Abdul Karim", host_phone="+8801711223344", is_published=True, ) session.add(event) await session.flush() schedule = ReminderSchedule( event_id=event.id, offset_days=15, send_at_local_time=time(10, 0), audience=ReminderAudience.ALL, is_enabled=True, ) session.add(schedule) # Templates are unique on (channel, purpose, locale) and the seed owns them globally, # so the fixture deliberately creates none. `_template_for` returning None is a valid # state the pipeline must handle anyway — a job with no template falls back to the # built-in body rather than failing. invitation_ids = [] for i in range(5): guest = Guest( wedding_id=wedding.id, # A guest belongs to one event (design D11), so the fixture's guests belong to # the fixture's event rather than floating at the wedding. event_id=event.id, full_name=f"Guest {i}", email=f"guest{i}-{tag}@example.test", preferred_locale=Locale.EN, ) session.add(guest) await session.flush() invitation = Invitation( guest_id=guest.id, event_id=event.id, token=f"tok-{tag}-{i}", short_code=f"c{tag[:4]}{i}", status=InvitationStatus.ACCEPTED, max_guests=2, ) session.add(invitation) await session.flush() invitation_ids.append(invitation.id) await session.commit() event_id, schedule_id = event.id, schedule.id yield { "tag": tag, "wedding_id": wedding_id, "event_id": event_id, "schedule_id": schedule_id, "invitation_ids": invitation_ids, "guest_count": 5, "owner_admin_id": owner_id, } await session.rollback() await session.execute(text("DELETE FROM wedding WHERE id = :id"), {"id": wedding_id}) # The owner outlives the cascade, and has to go second: `event.owner_admin_id` is # ON DELETE RESTRICT, so removing the admin first is refused by the database. await forget_admin(session, owner_id) await session.commit() @pytest.fixture(scope="session", autouse=True) def _accounts_the_suite_must_not_break() -> Any: """Fail the run if the suite leaves a pre-existing admin account unusable. `make test` runs against the dev database, which holds the operator's real account. A test that withdraws "every other super admin" to make itself the last one — which is the only way to exercise that invariant — will sweep up that account, and if its restore is partial the operator is locked out of their own system by running the test suite. That is not hypothetical: it happened, the operator could not sign in, and nothing in the suite noticed because every test still passed. Hence a check on the *database*, not on any one test: it records which accounts were active before the run and asserts they still are afterwards, so the next incomplete restore fails loudly instead of silently. """ import asyncio if not DATABASE_URL.startswith("postgresql"): yield return async def active_admins() -> set[str]: eng = create_async_engine(DATABASE_URL) try: async with eng.connect() as conn: rows = await conn.execute( text("SELECT email FROM admin_user WHERE status = 'active'") ) return {r[0] for r in rows} finally: await eng.dispose() before = asyncio.run(active_admins()) yield after = asyncio.run(active_admins()) # Accounts the suite created and removed are gone from both sets, so only a pre-existing # account that is still present but no longer active trips this. broken = {e for e in before - after if not e.endswith("@example.test")} assert not broken, ( "The suite left these pre-existing admin accounts inactive: " f"{sorted(broken)}. A test mutated rows it did not create and did not put them back — " "run against a dev database this locks a real operator out." )