"""The upgrade backfill: what it derives, and what it refuses to invent. This runs the migration's own UPDATE statements — imported from the revision file rather than retyped — against real rows. A test that reimplemented the SQL would agree with itself while the migration went out wrong, which is the only failure mode worth guarding here. What it does not cover: the add-column → backfill → alter-column sequencing, which needs a real `alembic upgrade` against a throwaway database. There is no harness for that in this repo and building one against the shared dev database is not worth the risk; `make migrate` on a database with existing events (task 1.5) is what proves the ordering. """ import importlib.util import uuid from pathlib import Path from typing import Any import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from .conftest import forget_admin, make_owner_admin, requires_db pytestmark = [requires_db] def _load_revision() -> Any: """Alembic versions are not an importable package, so load the module by path.""" path = ( Path(__file__).resolve().parents[1] / "alembic" / "versions" / "a3c9d05e2f71_event_host_contact.py" ) spec = importlib.util.spec_from_file_location("event_host_contact_revision", path) assert spec and spec.loader, f"could not load {path}" module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module revision = _load_revision() async def _seed( session: AsyncSession, *, wedding_phone: str | None ) -> tuple[uuid.UUID, str, uuid.UUID]: """One wedding and one event, with the host columns set to values the backfill must overwrite — so a statement that silently matched nothing would fail the assertions.""" tag = uuid.uuid4().hex[:8] wedding_id = uuid.uuid4() event_id = uuid.uuid4() # Every event needs an owner (add-admin-access-control D1). owner_id = await make_owner_admin(session, tag=f"backfill-{tag}") await session.execute( text( "INSERT INTO wedding (id, bride_name, groom_name, slug, default_locale, " "timezone, host_contact_phone) VALUES (:id, 'Nazifa', 'Abdullah', :slug, " "'en', 'Asia/Dhaka', :phone)" ), {"id": wedding_id, "slug": f"backfill-{tag}", "phone": wedding_phone}, ) await session.execute( text( "INSERT INTO event (id, wedding_id, owner_admin_id, type, slug, title_bn, " "title_en, starts_at, venue_name, venue_address, theme_key, is_published, " "invitation_messages, host_name_1, host_name_2, host_phone) " "VALUES (:id, :wedding_id, :owner_id, 'walima', :slug, 'ওয়ালিমা', 'Walima', now(), " "'Hall', 'Dhaka', 'classic', true, '{}', " "'OVERWRITE ME', 'OVERWRITE ME TOO', 'OVERWRITE ME')" ), { "id": event_id, "wedding_id": wedding_id, "owner_id": owner_id, "slug": f"backfill-{tag}", }, ) await session.commit() return wedding_id, str(event_id), owner_id async def _run_backfill(session: AsyncSession) -> None: params = { "placeholder_phone": revision.PLACEHOLDER_PHONE, "placeholder_name": revision.PLACEHOLDER_HOST_NAME, } await session.execute(text(revision.BACKFILL_FROM_WEDDING_SQL), params) await session.execute(text(revision.BACKFILL_ORPHANS_SQL), params) await session.commit() async def _read(session: AsyncSession, event_id: str) -> Any: row = await session.execute( text("SELECT host_name_1, host_name_2, host_phone FROM event WHERE id = :id"), {"id": event_id}, ) return row.one() async def _cleanup(session: AsyncSession, wedding_id: uuid.UUID, owner_id: uuid.UUID) -> None: """Wedding first: the event's `owner_admin_id` is ON DELETE RESTRICT, so the admin cannot go until the event that points at it is gone.""" await session.execute(text("DELETE FROM wedding WHERE id = :id"), {"id": wedding_id}) await forget_admin(session, owner_id) await session.commit() @pytest.mark.usefixtures("app_engine_per_loop") async def test_the_phone_comes_from_the_wedding(session: AsyncSession) -> None: """A genuine derivation: the wedding-wide number always was the family's contact.""" wedding_id, event_id, owner_id = await _seed(session, wedding_phone="+8801700000000") try: await _run_backfill(session) _, _, host_phone = await _read(session, event_id) assert host_phone == "+8801700000000" finally: await _cleanup(session, wedding_id, owner_id) @pytest.mark.usefixtures("app_engine_per_loop") async def test_no_wedding_phone_gets_the_sentinel(session: AsyncSession) -> None: """Not a blank and not a plausible number — one a dialler reaches nobody on.""" wedding_id, event_id, owner_id = await _seed(session, wedding_phone=None) try: await _run_backfill(session) _, _, host_phone = await _read(session, event_id) assert host_phone == revision.PLACEHOLDER_PHONE assert host_phone == "+880000000000" finally: await _cleanup(session, wedding_id, owner_id) @pytest.mark.usefixtures("app_engine_per_loop") async def test_a_blank_wedding_phone_is_treated_as_absent(session: AsyncSession) -> None: """`NULLIF(TRIM(...), '')` earns its place: an empty string stored years ago would otherwise become an event's host phone and render as an empty `tel:` link.""" wedding_id, event_id, owner_id = await _seed(session, wedding_phone=" ") try: await _run_backfill(session) _, _, host_phone = await _read(session, event_id) assert host_phone == revision.PLACEHOLDER_PHONE finally: await _cleanup(session, wedding_id, owner_id) @pytest.mark.usefixtures("app_engine_per_loop") async def test_the_host_name_is_a_placeholder_never_the_couple(session: AsyncSession) -> None: """The one that matters most. The host is a *parent* of the couple. Writing `bride_name`/`groom_name` here would put the couple's own names under "Invited By" on every existing invitation — not an approximation of the truth but a statement the domain says is false (design D5). """ wedding_id, event_id, owner_id = await _seed(session, wedding_phone="+8801700000000") try: await _run_backfill(session) host_name_1, host_name_2, _ = await _read(session, event_id) assert host_name_1 == revision.PLACEHOLDER_HOST_NAME assert host_name_1 not in {"Nazifa", "Abdullah"} # No second host either: inventing one would be the same mistake twice. assert host_name_2 is None finally: await _cleanup(session, wedding_id, owner_id)