"""Regression: offboarding mail-group owner-transfer notification coalescing. Prod 2026-05-13: leaver P0092 huohao had 10 mailing groups all routing to the CEO as fallback owner. Each ``transfer_group_owner`` call fired its own Feishu push → 10 pings in one offboarding round. The fix in ``email_group/service.py::transfer_group_owner`` coalesces same-recipient same-leaver transfers within a 5-min window: the first one dispatches normally; follow-ups update the first outbox row's title/body/context for the inbox digest view and skip a fresh Feishu/email send. These tests pin the behavior so a future contributor can't accidentally restore the per-group fan-out. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from unittest.mock import AsyncMock, patch import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.email_group.service import EmailGroupService from app.identity.person_models import IdentityPerson from app.integrations.drivers.google_workspace import WorkspaceOpResult from app.notifications.models import NotificationOutbox async def _person( db: AsyncSession, *, email: str, name: str, ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=email, display_name=name, employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.flush() return p async def _group( db: AsyncSession, *, email: str, owner: IdentityPerson, ) -> EmailGroup: g = EmailGroup( id=uuid.uuid4(), email=email, name=email.split("@", 1)[0], status="active", ) db.add(g) await db.flush() db.add(EmailGroupOwner( id=uuid.uuid4(), email_group_id=g.id, person_id=owner.id, )) await db.flush() return g class _StubDriver: """Always-OK Google driver — exercises only the local + notification path.""" async def set_group_owner(self, **_kw) -> WorkspaceOpResult: return WorkspaceOpResult(ok=True, data={"already_absent": False}) async def remove_group_owner(self, **_kw) -> WorkspaceOpResult: return WorkspaceOpResult(ok=True, data={"already_absent": False}) async def _patch_send(svc: EmailGroupService): """Stub the channel sends so we don't try to talk to Feishu / SMTP.""" svc.notify._send_feishu = AsyncMock(return_value=(True, None, None)) # type: ignore[method-assign] svc.notify._send_email = AsyncMock(return_value=(True, None)) # type: ignore[method-assign] async def _outbox_for( db: AsyncSession, recipient_id: uuid.UUID, ) -> list[NotificationOutbox]: rows = ( await db.execute( select(NotificationOutbox) .where(NotificationOutbox.recipient_person_id == recipient_id) .order_by(NotificationOutbox.created_at.asc()) ) ).scalars().all() return list(rows) class TestOffboardingBatchCoalesce: @pytest.mark.asyncio async def test_three_groups_one_recipient_collapse_to_one_outbox( self, db_session: AsyncSession, ) -> None: leaver = await _person( db_session, email="leaver@pingcap.com", name="Leaver Lou", ) successor = await _person( db_session, email="succ@pingcap.com", name="Succ Successor", ) groups = [ await _group(db_session, email=f"team-{i}@pingcap.com", owner=leaver) for i in range(3) ] await db_session.commit() svc = EmailGroupService(db_session) await _patch_send(svc) with patch.object(svc, "_get_driver", return_value=_StubDriver()): for g in groups: await svc.transfer_group_owner( group=g, from_person=leaver, to_person=successor, actor_user_id=successor.id, trace_id=str(uuid.uuid4()), initiator="offboarding", ) await db_session.commit() rows = await _outbox_for(db_session, successor.id) assert len(rows) == 1, ( f"expected 1 coalesced outbox row, got {len(rows)}" ) ctx = rows[0].context or {} assert ctx.get("group_count") == 3 assert {g["email"] for g in ctx["groups"]} == { "team-0@pingcap.com", "team-1@pingcap.com", "team-2@pingcap.com", } assert "3 个邮件组" in rows[0].title for g in groups: assert g.email in rows[0].body_text @pytest.mark.asyncio async def test_on_demand_transfer_does_not_coalesce( self, db_session: AsyncSession, ) -> None: """Admin / owner-initiated UI transfers stay 1:1 — recipient didn't click anything, the "you've been given this group" push is genuinely informational and must reach them per-group. """ leaver = await _person( db_session, email="leaver2@pingcap.com", name="Leaver Two", ) successor = await _person( db_session, email="succ2@pingcap.com", name="Succ Two", ) groups = [ await _group(db_session, email=f"od-{i}@pingcap.com", owner=leaver) for i in range(2) ] await db_session.commit() svc = EmailGroupService(db_session) await _patch_send(svc) with patch.object(svc, "_get_driver", return_value=_StubDriver()): for g in groups: await svc.transfer_group_owner( group=g, from_person=leaver, to_person=successor, actor_user_id=successor.id, trace_id=str(uuid.uuid4()), initiator="on_demand", ) await db_session.commit() rows = await _outbox_for(db_session, successor.id) assert len(rows) == 2, ( "on_demand transfers must not coalesce — each is an " "independent admin action that deserves its own push" ) @pytest.mark.asyncio async def test_different_leavers_do_not_coalesce( self, db_session: AsyncSession, ) -> None: """Two unrelated offboardings that happen to land on the same successor within 5 min must not silently merge — the digest would lie about who lost which group. """ leaver_a = await _person( db_session, email="la@pingcap.com", name="Leaver A", ) leaver_b = await _person( db_session, email="lb@pingcap.com", name="Leaver B", ) successor = await _person( db_session, email="succ3@pingcap.com", name="Succ Three", ) g_a = await _group(db_session, email="a@pingcap.com", owner=leaver_a) g_b = await _group(db_session, email="b@pingcap.com", owner=leaver_b) await db_session.commit() svc = EmailGroupService(db_session) await _patch_send(svc) with patch.object(svc, "_get_driver", return_value=_StubDriver()): await svc.transfer_group_owner( group=g_a, from_person=leaver_a, to_person=successor, actor_user_id=successor.id, trace_id=str(uuid.uuid4()), initiator="offboarding", ) await svc.transfer_group_owner( group=g_b, from_person=leaver_b, to_person=successor, actor_user_id=successor.id, trace_id=str(uuid.uuid4()), initiator="offboarding", ) await db_session.commit() rows = await _outbox_for(db_session, successor.id) assert len(rows) == 2, ( "transfers from different leavers must stay distinct" ) @pytest.mark.asyncio async def test_stale_burst_outside_window_does_not_coalesce( self, db_session: AsyncSession, ) -> None: """An outbox row older than the 5-min window must NOT pull a new transfer into its digest — a leaver's stale digest from last hour shouldn't silently swallow a fresh offboarding's first-of-burst notification. """ leaver = await _person( db_session, email="lstale@pingcap.com", name="Leaver Stale", ) successor = await _person( db_session, email="ssstale@pingcap.com", name="Succ Stale", ) g1 = await _group(db_session, email="old@pingcap.com", owner=leaver) g2 = await _group(db_session, email="new@pingcap.com", owner=leaver) await db_session.commit() svc = EmailGroupService(db_session) await _patch_send(svc) with patch.object(svc, "_get_driver", return_value=_StubDriver()): await svc.transfer_group_owner( group=g1, from_person=leaver, to_person=successor, actor_user_id=successor.id, trace_id=str(uuid.uuid4()), initiator="offboarding", ) # Push the first outbox row's created_at outside the 5-min window. rows = await _outbox_for(db_session, successor.id) assert len(rows) == 1 rows[0].created_at = datetime.now(UTC) - timedelta(minutes=10) await db_session.flush() await svc.transfer_group_owner( group=g2, from_person=leaver, to_person=successor, actor_user_id=successor.id, trace_id=str(uuid.uuid4()), initiator="offboarding", ) await db_session.commit() rows = await _outbox_for(db_session, successor.id) assert len(rows) == 2, ( "transfers spaced > 5 min apart must each fire their own push" )