"""Batched sends against real Postgres (add-bulk-invitation-send D5, D8, D10). These need a real database for the same reason the rest of the pipeline suite does: what they check is the unique constraint doing its job and the sender's live re-read doing its job, neither of which exists in a fake. The fixture here is deliberately not `fixture_event`. A batch is interesting precisely where that one is uniform — it needs both invitation types, both locales, suppressed guests, and invitations that are **pending**, because an invitation by definition goes to someone who has not answered yet. That last one is what D8 turns on. Run with `make test` (the api container has DATABASE_URL); they skip elsewhere. """ import uuid from collections.abc import AsyncIterator from datetime import UTC, datetime, timedelta from typing import Any import pytest import pytest_asyncio from sqlalchemy import func, select, text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.orm import selectinload from app.models import Event, Guest, Invitation, MessageJob, Wedding from app.models.enums import ( Channel, EventType, InvitationStatus, InvitationType, Locale, MessageJobStatus, ) from app.services import messaging from tests.conftest import forget_admin, make_owner_admin, requires_db pytestmark = requires_db @pytest_asyncio.fixture async def batch_event(session: AsyncSession) -> AsyncIterator[dict[str, Any]]: """A wedding whose guest list looks like a real one: mixed, and mostly unanswered. Six guests — three sendable in English (two single, one family), one sendable in Bangla, and two suppressed in different ways. Their invitations are `pending`, which is the state every real invitation is in when it is sent. """ tag = uuid.uuid4().hex[:8] # Every event needs an owner (add-admin-access-control D1). owner_id = await make_owner_admin(session, tag=f"bulk-{tag}") wedding = Wedding( bride_name="Ayesha", groom_name="Rahim", slug=f"bulk-{tag}", timezone="Asia/Dhaka" ) session.add(wedding) await session.flush() 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", 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, invitation_messages={"en": {"single": "Come alone", "family": "Bring everyone"}}, ) session.add(event) await session.flush() specs: list[dict[str, Any]] = [ {"key": "single_en", "type": InvitationType.SINGLE, "locale": Locale.EN}, {"key": "single_en_2", "type": InvitationType.SINGLE, "locale": Locale.EN}, {"key": "family_en", "type": InvitationType.FAMILY, "locale": Locale.EN}, {"key": "single_bn", "type": InvitationType.SINGLE, "locale": Locale.BN}, { "key": "opted_out", "type": InvitationType.SINGLE, "locale": Locale.EN, "do_not_contact": True, }, {"key": "no_email", "type": InvitationType.SINGLE, "locale": Locale.EN, "email": None}, ] invitation_ids: dict[str, uuid.UUID] = {} for i, spec in enumerate(specs): guest = Guest( wedding_id=wedding.id, event_id=event.id, full_name=f"Guest {spec['key']}", email=(None if spec.get("email", "") is None else f"{spec['key']}-{tag}@example.test"), # A guest row must carry some way to reach them, so the one without an address # has a phone — which is exactly the real case v1 cannot mail, email-only. phone_e164=f"+88017{i:08d}" if spec.get("email", "") is None else None, preferred_locale=spec["locale"], invitation_type=spec["type"], do_not_contact=bool(spec.get("do_not_contact")), ) session.add(guest) await session.flush() invitation = Invitation( guest_id=guest.id, event_id=event.id, token=f"btok-{tag}-{i}", short_code=f"b{tag[:4]}{i}", # Pending, not accepted: this is what an invitation is sent into. status=InvitationStatus.PENDING, max_guests=4, ) session.add(invitation) await session.flush() invitation_ids[str(spec["key"])] = invitation.id await session.commit() event_id = event.id yield { "tag": tag, "wedding_id": wedding_id, "event_id": event_id, "invitation_ids": invitation_ids, "sendable": 4, "suppressed": 2, } await session.rollback() await session.execute(text("DELETE FROM wedding WHERE id = :id"), {"id": wedding_id}) # After the cascade: the owner FK is ON DELETE RESTRICT, so the events must go first. await forget_admin(session, owner_id) await session.commit() async def _invitations(session: AsyncSession, event_id: uuid.UUID) -> list[Invitation]: rows = await session.scalars( select(Invitation) .where(Invitation.event_id == event_id) .options( selectinload(Invitation.guest), selectinload(Invitation.event).selectinload(Event.wedding), ) .order_by(Invitation.token) ) return list(rows) def _templates(event: Event, wedding: Wedding) -> dict[tuple[Locale, InvitationType], Any]: return { (locale, invitation_type): messaging.compose_bulk( event, wedding, locale=locale, invitation_type=invitation_type ) for locale in (Locale.EN, Locale.BN) for invitation_type in (InvitationType.SINGLE, InvitationType.FAMILY) } async def _enqueue( session: AsyncSession, event_id: uuid.UUID, batch_id: uuid.UUID ) -> messaging.BatchEnqueueResult: invitations = await _invitations(session, event_id) event = invitations[0].event return await messaging.enqueue_batch( session, invitations=invitations, templates=_templates(event, event.wedding), batch_id=batch_id, # Set so the tests do not depend on what time of day they run at; the quiet-hours # branch itself is covered in the manual-send suite and in the route tests. override_quiet_hours=True, sent_by_admin_id=None, ) async def _batch_jobs(session: AsyncSession, batch_id: uuid.UUID) -> list[MessageJob]: await session.rollback() rows = await session.scalars(select(MessageJob).where(MessageJob.batch_id == batch_id)) return list(rows) async def _batch_statuses(session: AsyncSession, batch_id: uuid.UUID) -> set[MessageJobStatus]: """Statuses as plain values. Columns rather than entities because the rollback that starts each read expires every instance the session is holding — a second read would leave the first read's objects lazy-loading outside the async context, which is a fact about the fixture rather than about the code under test. """ await session.rollback() rows = await session.scalars(select(MessageJob.status).where(MessageJob.batch_id == batch_id)) return set(rows) def _capture_sends() -> Any: """The dry-run provider, emptied, so what actually reached it can be read back.""" from app.services import providers provider = providers.get_provider(Channel.EMAIL) assert isinstance(provider, providers.DryRunProvider), ( "these tests read what was sent, which requires DRY_RUN" ) provider.sent.clear() return provider # ------------------------------------------------------------------ enqueue async def test_a_batch_creates_one_job_per_sendable_guest( session: AsyncSession, batch_event: dict[str, Any] ) -> None: batch_id = uuid.uuid4() result = await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() assert result.queued == batch_event["sendable"] assert result.duplicate == 0 assert len(await _batch_jobs(session, batch_id)) == batch_event["sendable"] async def test_suppressed_guests_get_no_job_and_are_counted_by_reason( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """D10: the admin needs the count before they press Send, and a job that exists only to be refused is noise in the log.""" batch_id = uuid.uuid4() result = await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() assert result.excluded == { messaging.BLOCK_CODE_OPTED_OUT: 1, messaging.BLOCK_CODE_NO_EMAIL: 1, } jobs = await _batch_jobs(session, batch_id) excluded_ids = { batch_event["invitation_ids"]["opted_out"], batch_event["invitation_ids"]["no_email"], } assert not ({j.invitation_id for j in jobs} & excluded_ids) async def test_each_job_carries_its_own_guests_name_and_link( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """D3: one authored text becomes as many personal messages as there are recipients.""" batch_id = uuid.uuid4() await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() jobs = await _batch_jobs(session, batch_id) bodies = [j.body_text or "" for j in jobs] assert len(bodies) == batch_event["sendable"] assert len(set(bodies)) == len(bodies), "two guests were sent identical bodies" for job, body in zip(jobs, bodies, strict=True): assert f"btok-{batch_event['tag']}" in body, "a guest's own invitation link is missing" assert "{invitation_link}" not in body assert "{guest_name}" not in body assert job.subject async def test_each_guest_gets_the_message_for_their_type_and_locale( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """D11: a Bangla reader is never sent the English pane.""" batch_id = uuid.uuid4() await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() jobs = await _batch_jobs(session, batch_id) by_invitation = {j.invitation_id: (j.body_text or "") for j in jobs} ids = batch_event["invitation_ids"] assert "Come alone" in by_invitation[ids["single_en"]] assert "Bring everyone" in by_invitation[ids["family_en"]] # The event customised English only, so the Bangla guest gets the Bangla built-in default # rather than the English override leaking across locales. assert "আপনাকে সাদর আমন্ত্রণ" in by_invitation[ids["single_bn"]] assert "Come alone" not in by_invitation[ids["single_bn"]] # ------------------------------------------------------------------ the duplicate boundary async def test_the_same_batch_id_twice_inserts_nothing_new( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """D5, against the constraint that enforces it: a double-clicked Send is a no-op.""" batch_id = uuid.uuid4() first = await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() second = await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() assert first.queued == batch_event["sendable"] assert second.queued == 0 assert second.duplicate == batch_event["sendable"] assert len(await _batch_jobs(session, batch_id)) == batch_event["sendable"] async def test_two_different_batches_both_send( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """D5, the other half: a host deliberately sending again is delivered.""" first_id, second_id = uuid.uuid4(), uuid.uuid4() first = await _enqueue(session, batch_event["event_id"], first_id) await session.commit() second = await _enqueue(session, batch_event["event_id"], second_id) await session.commit() assert first.queued == second.queued == batch_event["sendable"] assert len(await _batch_jobs(session, first_id)) == batch_event["sendable"] assert len(await _batch_jobs(session, second_id)) == batch_event["sendable"] async def test_a_scheduled_wave_planned_twice_still_yields_one_job( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """Repeatability belongs to batches alone; automated duplicate protection is untouched.""" invitation_id = batch_event["invitation_ids"]["single_en"] schedule_id = uuid.uuid4() first = await messaging.enqueue( session, invitation_id=invitation_id, template_id=None, schedule_id=None, scheduled_for=datetime.now(UTC), key=messaging.idempotency_key(invitation_id, schedule_id, Channel.EMAIL), ) second = await messaging.enqueue( session, invitation_id=invitation_id, template_id=None, schedule_id=None, scheduled_for=datetime.now(UTC), key=messaging.idempotency_key(invitation_id, schedule_id, Channel.EMAIL), ) await session.commit() assert first is not None assert second is None # ------------------------------------------------------------------ D8: who may be sent to async def test_a_batched_job_sends_to_a_pending_invitation_under_the_workers_own_rule( session: AsyncSession, sessions: async_sessionmaker[AsyncSession], batch_event: dict[str, Any], ) -> None: """D8 — the hole this change had to close. The worker claims batched jobs on retry and after a restart, and it passes `audience_accepted=True`, which skips anything not already accepted. An invitation goes to someone who has not answered, so under the old rule every batched invitation the worker touched would be dropped as "no longer accepted" and nobody would notice. """ provider = _capture_sends() batch_id = uuid.uuid4() await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() jobs = await _batch_jobs(session, batch_id) job_id = jobs[0].id async with sessions() as s: job = await s.get(MessageJob, job_id) assert job is not None # The worker's own call: no `audience_accepted=False` anywhere. status = await messaging.process_job(s, job) await s.commit() assert status == MessageJobStatus.SENT assert len(provider.sent) == 1 async def test_a_reminder_to_a_pending_invitation_still_skips( session: AsyncSession, sessions: async_sessionmaker[AsyncSession], batch_event: dict[str, Any], ) -> None: """The other side of D8: a job with no stored content keeps the old rule exactly. This is what makes a cancellation retroactive, and loosening it for reminders would mean mailing "see you in 2 days" to someone who cancelled last week. """ job_id = await messaging.enqueue( session, invitation_id=batch_event["invitation_ids"]["single_en"], template_id=None, schedule_id=None, scheduled_for=datetime.now(UTC), override_quiet_hours=True, ) await session.commit() assert job_id is not None async with sessions() as s: job = await s.get(MessageJob, job_id) assert job is not None status = await messaging.process_job(s, job) await s.commit() assert status == MessageJobStatus.SKIPPED assert job.skip_reason == messaging.SKIP_NOT_ACCEPTED async def test_a_guest_who_opts_out_after_recording_is_not_sent_to( session: AsyncSession, sessions: async_sessionmaker[AsyncSession], batch_event: dict[str, Any], ) -> None: """D10: filtering at enqueue is for the count; the send-time re-read is the guarantee.""" _capture_sends() batch_id = uuid.uuid4() await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() invitation_id = batch_event["invitation_ids"]["single_en"] invitation = await session.get(Invitation, invitation_id) assert invitation is not None guest = await session.get(Guest, invitation.guest_id) assert guest is not None guest.do_not_contact = True await session.commit() job = next(j for j in await _batch_jobs(session, batch_id) if j.invitation_id == invitation_id) async with sessions() as s: claimed = await s.get(MessageJob, job.id) assert claimed is not None status = await messaging.process_job(s, claimed) await s.commit() assert status == MessageJobStatus.SKIPPED assert claimed.skip_reason == messaging.SKIP_DO_NOT_CONTACT # ------------------------------------------------------------------ dispatch and progress async def test_dispatching_a_batch_sends_only_that_batch( session: AsyncSession, batch_event: dict[str, Any], app_engine_per_loop: None ) -> None: """D6: pressing Send must not turn the web process into a second sender for the whole queue — the claim is narrowed to this batch. `app_engine_per_loop` because `dispatch_batch` opens its own session from the process-wide cached factory, exactly as it does behind a real request — and under pytest that cache would otherwise hold an engine belonging to a closed loop. """ mine_id, other_id = uuid.uuid4(), uuid.uuid4() await _enqueue(session, batch_event["event_id"], mine_id) await _enqueue(session, batch_event["event_id"], other_id) await session.commit() await messaging.dispatch_batch(mine_id) assert await _batch_statuses(session, mine_id) == {MessageJobStatus.SENT} # Deliberately no assertion that `other_id` is still queued. The dev stack's own worker # polls this queue and may legitimately have sent it — what this change promises is that # *this* call claims only its own batch, and that is asserted directly below rather than # inferred from what another sender did or did not get to first. async def test_a_batch_scoped_claim_takes_only_that_batch( session: AsyncSession, sessions: async_sessionmaker[AsyncSession], batch_event: dict[str, Any], ) -> None: """The filter behind D6, on its own. Without it, pressing Send would turn the web process into a second sender for every queued reminder in the system — correct by accident, and a surprise the first time a wave is pending. """ mine_id, other_id = uuid.uuid4(), uuid.uuid4() await _enqueue(session, batch_event["event_id"], mine_id) await _enqueue(session, batch_event["event_id"], other_id) await session.commit() async with sessions() as claimer: claimed = await messaging.claim_due_jobs(claimer, 100, batch_id=mine_id) batches = {job.batch_id for job in claimed} await claimer.commit() assert batches <= {mine_id}, "a batch-scoped claim reached outside its batch" async def test_progress_counts_a_batch_that_has_gone_out( session: AsyncSession, batch_event: dict[str, Any], app_engine_per_loop: None ) -> None: batch_id = uuid.uuid4() await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() # Every job accounted for, in whatever state it has reached. Deliberately not # `waiting == sendable`: the dev stack's own worker polls this queue every 30 seconds and # is free to claim these rows between the commit above and this read, so a snapshot that # insists nothing has gone out yet is a race with the environment rather than a # statement about the code — `test_pipeline_integration` carries the same caveat. before = await messaging.batch_progress(session, batch_id) assert before.total == batch_event["sendable"] assert before.waiting + before.sent + before.failed + before.skipped == batch_event["sendable"] await messaging.dispatch_batch(batch_id) await session.rollback() # The end state is the real guarantee, and it holds however many senders were involved: # `SKIP LOCKED` means they take disjoint rows, so every job lands sent exactly once. after = await messaging.batch_progress(session, batch_id) assert after.sent == batch_event["sendable"] assert after.waiting == 0 assert after.finished assert after.problems == [] async def test_progress_names_the_guests_it_could_not_send_to( session: AsyncSession, batch_event: dict[str, Any], app_engine_per_loop: None ) -> None: """D7: a count of failures the admin cannot act on is barely better than no count.""" batch_id = uuid.uuid4() await _enqueue(session, batch_event["event_id"], batch_id) await session.commit() invitation_id = batch_event["invitation_ids"]["family_en"] invitation = await session.get(Invitation, invitation_id) assert invitation is not None guest = await session.get(Guest, invitation.guest_id) assert guest is not None guest.do_not_contact = True await session.commit() await messaging.dispatch_batch(batch_id) await session.rollback() progress = await messaging.batch_progress(session, batch_id) assert progress.skipped == 1 assert len(progress.problems) == 1 assert progress.problems[0].guest_name == "Guest family_en" assert progress.problems[0].reason == messaging.SKIP_DO_NOT_CONTACT async def test_progress_for_an_unknown_batch_is_empty_rather_than_an_error( session: AsyncSession, ) -> None: progress = await messaging.batch_progress(session, uuid.uuid4()) assert progress.total == 0 assert progress.finished async def test_a_batch_of_nobody_records_nothing( session: AsyncSession, batch_event: dict[str, Any] ) -> None: """Every selected guest suppressed is a real state, and it must not insert an empty statement or claim a send happened.""" invitations = [ inv for inv in await _invitations(session, batch_event["event_id"]) if inv.guest is not None and messaging.suppression(inv.guest) is not None ] assert invitations, "the fixture must contain suppressed guests for this to mean anything" batch_id = uuid.uuid4() event = invitations[0].event result = await messaging.enqueue_batch( session, invitations=invitations, templates=_templates(event, event.wedding), batch_id=batch_id, override_quiet_hours=True, sent_by_admin_id=None, ) await session.commit() assert result.queued == 0 assert sum(result.excluded.values()) == len(invitations) assert ( await session.scalar( select(func.count(MessageJob.id)).where(MessageJob.batch_id == batch_id) ) ) == 0 @pytest.mark.parametrize("override", [True, False]) async def test_the_quiet_hours_decision_is_stamped_on_every_job( session: AsyncSession, batch_event: dict[str, Any], override: bool ) -> None: """D9: the flag rides on the row, so a job the worker picks up at 23:50 still honours what the admin was told.""" batch_id = uuid.uuid4() invitations = await _invitations(session, batch_event["event_id"]) event = invitations[0].event await messaging.enqueue_batch( session, invitations=invitations, templates=_templates(event, event.wedding), batch_id=batch_id, override_quiet_hours=override, sent_by_admin_id=None, ) await session.commit() jobs = await _batch_jobs(session, batch_id) assert jobs assert all(job.override_quiet_hours is override for job in jobs)