"""Self-service mail-group membership view + leave flow. The Google Workspace driver is patched: tests assert RBAC + DB side-effects (EmployeeRequest record, owner-row cleanup) without hitting Google. Behaviours covered: 1. ``GET /me/email-groups`` lists the caller's memberships (live from Google) with ``is_owner`` / ``sole_owner`` derived from the synced ``email_group_owners`` table. 2. The sole owner of a group CANNOT leave — 409 ``SOLE_OWNER_CANNOT_LEAVE``, and Google is never called. 3. A plain member CAN leave — 200, a completed ``google_group_leave`` EmployeeRequest is recorded, Google is called once. 4. A co-owner CAN leave — their local owner row is removed while other owners survive. """ from __future__ import annotations import uuid from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.asset.models import EmployeeRequest from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.identity.person_models import IdentityPerson async def _make_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 _make_group( db: AsyncSession, *, email: str, owners: list[IdentityPerson], is_dynamic: bool | None = None, ) -> EmailGroup: g = EmailGroup( id=uuid.uuid4(), email=email, name=email.split("@", 1)[0], status="active", is_dynamic=is_dynamic, ) db.add(g) await db.flush() for p in owners: db.add(EmailGroupOwner( id=uuid.uuid4(), email_group_id=g.id, person_id=p.id, )) await db.flush() return g def _ok_result(**data): from app.integrations.drivers.google_workspace import WorkspaceOpResult return WorkspaceOpResult(ok=True, data=data) @pytest.fixture def patch_driver(): """Patch ``_get_driver`` with an AsyncMock exposing the membership + leave hooks. ``list_groups_for_user`` returns three groups so the owner-flag derivation can be asserted across all three roles. """ fake = AsyncMock() fake.list_groups_for_user.return_value = [ {"email": "sole@pingcap.com", "name": "Sole"}, {"email": "coown@pingcap.com", "name": "Coowned"}, {"email": "member@pingcap.com", "name": "MemberOnly"}, {"email": "dynamic@pingcap.com", "name": "DynamicAllHands"}, ] fake.remove_group_member.return_value = _ok_result(already_absent=False) # EVERY self-leave now consults the live owner roster, not just the # unsynced-group path (the ``email_group_owners`` mirror only holds # owners resolvable to an ``identity_persons`` row, so it is wrong in # both directions — including the direction that strands a group). # An unconfigured ``AsyncMock`` answers with a ``MagicMock`` whose # ``__iter__`` is empty, i.e. "this group has no owners at all", # which silently permits every leave. Default to "somebody ELSE # owns it" and let the tests that care say otherwise. fake.list_group_owners.return_value = [{"email": "other@pingcap.com"}] with patch( "app.domains.email_group.service.EmailGroupService._get_driver", return_value=fake, ): yield fake @pytest.mark.asyncio async def test_list_memberships_with_owner_flags( client: AsyncClient, db_session, employee_user, patch_driver, ): user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" other = await _make_person(db_session, email="other@pingcap.com", name="Other") await _make_group(db_session, email="sole@pingcap.com", owners=[me]) await _make_group(db_session, email="coown@pingcap.com", owners=[me, other]) await _make_group(db_session, email="member@pingcap.com", owners=[other]) await _make_group( db_session, email="dynamic@pingcap.com", owners=[other], is_dynamic=True, ) await db_session.commit() r = await client.get( "/api/v1/me/email-groups", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200, r.text groups = {g["email"]: g for g in r.json()["data"]["groups"]} assert groups["sole@pingcap.com"]["is_owner"] is True assert groups["sole@pingcap.com"]["sole_owner"] is True assert groups["sole@pingcap.com"]["is_dynamic"] is False assert groups["coown@pingcap.com"]["is_owner"] is True assert groups["coown@pingcap.com"]["sole_owner"] is False assert groups["member@pingcap.com"]["is_owner"] is False assert groups["member@pingcap.com"]["sole_owner"] is False assert groups["dynamic@pingcap.com"]["is_dynamic"] is True @pytest.mark.asyncio async def test_sole_owner_cannot_leave( client: AsyncClient, db_session, employee_user, patch_driver, ): user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" await _make_group(db_session, email="sole@pingcap.com", owners=[me]) await db_session.commit() # Google agrees with the mirror: I am the only owner. Stated # explicitly because the refusal is decided from THIS, not from the # ``email_group_owners`` rows ``_make_group`` wrote. patch_driver.list_group_owners.return_value = [ {"email": "shuai.feng@pingcap.com"}, ] r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "sole@pingcap.com"}, ) assert r.status_code == 409, r.text assert r.json()["errors"][0]["code"] == "SOLE_OWNER_CANNOT_LEAVE" # Google must NOT be touched when we refuse up-front. assert patch_driver.remove_group_member.await_count == 0 @pytest.mark.asyncio async def test_sole_owner_unsynced_group_blocked_via_live_check( client: AsyncClient, db_session, employee_user, patch_driver, ): """Regression (2026-07-08): a sole owner of a group NOT synced into Helix (empty local mirror) must STILL be blocked from self-leaving. The local owner check is blind for unsynced groups, so ownership is verified live via the driver's list_group_owners.""" user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" await db_session.commit() # NO _make_group → group is unsynced patch_driver.list_group_owners.return_value = [ {"email": "shuai.feng@pingcap.com"}, ] r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "unsynced@pingcap.com"}, ) assert r.status_code == 409, r.text assert r.json()["errors"][0]["code"] == "SOLE_OWNER_CANNOT_LEAVE" assert patch_driver.remove_group_member.await_count == 0 @pytest.mark.asyncio async def test_co_owner_unsynced_group_can_leave( client: AsyncClient, db_session, employee_user, patch_driver, ): """A co-owner (2+ live owners) of an unsynced group CAN self-leave — leaving still leaves an owner behind.""" user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" await db_session.commit() patch_driver.list_group_owners.return_value = [ {"email": "shuai.feng@pingcap.com"}, {"email": "other@pingcap.com"}, ] r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "unsynced@pingcap.com"}, ) assert r.status_code == 200, r.text assert patch_driver.remove_group_member.await_count == 1 @pytest.mark.asyncio async def test_unsynced_group_ownership_unverifiable_refused( client: AsyncClient, db_session, employee_user, patch_driver, ): """If live ownership can't be read (driver returns None), refuse the leave rather than risk stranding the group with zero owners.""" user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" await db_session.commit() patch_driver.list_group_owners.return_value = None r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "unsynced@pingcap.com"}, ) assert r.status_code == 409, r.text assert r.json()["errors"][0]["code"] == "GROUP_OWNERSHIP_UNVERIFIABLE" assert patch_driver.remove_group_member.await_count == 0 @pytest.mark.asyncio async def test_member_leaves_group( client: AsyncClient, db_session, employee_user, patch_driver, ): user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" other = await _make_person(db_session, email="other@pingcap.com", name="Other") await _make_group(db_session, email="member@pingcap.com", owners=[other]) await db_session.commit() r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "member@pingcap.com", "reason": "no longer relevant"}, ) assert r.status_code == 200, r.text body = r.json()["data"] assert body["group_email"] == "member@pingcap.com" assert patch_driver.remove_group_member.await_count == 1 er = ( await db_session.execute( select(EmployeeRequest).where( EmployeeRequest.request_type == "google_group_leave", ) ) ).scalars().first() assert er is not None assert er.status == "completed" assert er.detail["group_email"] == "member@pingcap.com" assert er.detail["was_owner"] is False @pytest.mark.asyncio async def test_co_owner_leaves_and_owner_row_removed( client: AsyncClient, db_session, employee_user, patch_driver, ): user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" other = await _make_person(db_session, email="other@pingcap.com", name="Other") g = await _make_group(db_session, email="coown@pingcap.com", owners=[me, other]) await db_session.commit() r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "coown@pingcap.com"}, ) assert r.status_code == 200, r.text assert patch_driver.remove_group_member.await_count == 1 remaining = ( await db_session.execute( select(EmailGroupOwner).where( EmailGroupOwner.email_group_id == g.id, ) ) ).scalars().all() owner_ids = {o.person_id for o in remaining} assert me.id not in owner_ids assert other.id in owner_ids @pytest.mark.asyncio async def test_dynamic_group_cannot_leave( client: AsyncClient, db_session, employee_user, patch_driver, ): user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" other = await _make_person(db_session, email="other@pingcap.com", name="Other") await _make_group( db_session, email="dynamic@pingcap.com", owners=[other], is_dynamic=True, ) await db_session.commit() r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "dynamic@pingcap.com"}, ) assert r.status_code == 409, r.text assert r.json()["errors"][0]["code"] == "DYNAMIC_GROUP_CANNOT_LEAVE" # Dynamic groups are rule-managed — Google must NOT be touched. assert patch_driver.remove_group_member.await_count == 0 @pytest.mark.asyncio async def test_stale_owner_mirror_cannot_strand_a_synced_group( client: AsyncClient, db_session, employee_user, patch_driver, ): """The sole-owner guard must read Google, even for a synced group. ``email_group_owners`` only holds owners that resolve to an ``identity_persons`` row by ``work_email``, and it is refreshed on a 6-hour tick — so it can show a co-owner Google no longer has. Here the mirror says two owners (me + other) while Google says I am the only one left. Deciding from the mirror let me walk out and leave the group with ZERO owners: the exact 无人管理 state the guard exists to prevent, and one that surfaces later on the 邮件组管理 desk as an unowned group with no explanation. """ user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" other = await _make_person(db_session, email="other@pingcap.com", name="O") await _make_group(db_session, email="coown@pingcap.com", owners=[me, other]) await db_session.commit() # Google's answer, which is the one that counts: the co-owner was # removed in the console and the mirror has not caught up. patch_driver.list_group_owners.return_value = [ {"email": "shuai.feng@pingcap.com"}, ] r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "coown@pingcap.com"}, ) assert r.status_code == 409, r.text assert r.json()["errors"][0]["code"] == "SOLE_OWNER_CANNOT_LEAVE" assert patch_driver.remove_group_member.await_count == 0 @pytest.mark.asyncio async def test_an_unresolvable_co_owner_does_not_block_a_safe_leave( client: AsyncClient, db_session, employee_user, patch_driver, ): """The same mirror is wrong in the other direction too. An external co-owner (contractor, vendor address) has no ``identity_persons`` row, so ``email_group_owners`` cannot hold them and the mirror shows me as the only owner. Deciding from it turned a leave that is perfectly safe — Google keeps an owner behind — into a hard 409 with advice ("transfer ownership first") the user cannot act on. """ user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" await _make_group(db_session, email="sole@pingcap.com", owners=[me]) await db_session.commit() patch_driver.list_group_owners.return_value = [ {"email": "shuai.feng@pingcap.com"}, {"email": "vendor@partner.example"}, ] r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "sole@pingcap.com"}, ) assert r.status_code == 200, r.text assert patch_driver.remove_group_member.await_count == 1 @pytest.mark.asyncio async def test_a_google_read_outage_does_not_break_every_leave( client: AsyncClient, db_session, employee_user, patch_driver, ): """An INDETERMINATE ownership read is not "you might be the owner". ``list_group_owners`` answers ``None`` for every non-404 failure — 403, 5xx, transport, paginator overrun — and it needs a scope the leave itself does not. Once the live check ran on EVERY leave, refusing on ``None`` meant one Google blip returned 「请联系 IT 管理员」 to ordinary members of groups where no ownership question exists at all, killing self-service leave platform-wide for the duration. So an indeterminate read falls back to the mirror, and only in the safe direction: the mirror says this person is not an owner, which is exactly the state this whole check was NOT written for. """ user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" other = await _make_person(db_session, email="other@pingcap.com", name="O") await _make_group(db_session, email="member@pingcap.com", owners=[other]) await db_session.commit() patch_driver.list_group_owners.return_value = None r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "member@pingcap.com"}, ) assert r.status_code == 200, r.text assert patch_driver.remove_group_member.await_count == 1 @pytest.mark.asyncio async def test_an_outage_still_refuses_when_the_mirror_says_sole_owner( client: AsyncClient, db_session, employee_user, patch_driver, ): """The other half of the fallback: it must not open the hole. When Google cannot be read AND the mirror shows this person holding ownership alone, refusing is the pre-existing, correct behaviour — stranding a group with zero owners is worse than a blocked leave. """ user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" await _make_group(db_session, email="sole@pingcap.com", owners=[me]) await db_session.commit() patch_driver.list_group_owners.return_value = None r = await client.post( "/api/v1/me/email-groups/leave", headers={"Authorization": f"Bearer {token}"}, json={"group_email": "sole@pingcap.com"}, ) assert r.status_code == 409, r.text assert r.json()["errors"][0]["code"] == "GROUP_OWNERSHIP_UNVERIFIABLE" assert patch_driver.remove_group_member.await_count == 0 @pytest.mark.asyncio async def test_member_list_keeps_an_entry_google_gives_no_address( client: AsyncClient, db_session, employee_user, patch_driver, ): """The list that SHOWS members must agree with the count. Google represents whole-domain membership as ``type: "CUSTOMER"`` with an ``id`` and no ``email``. Every counting path was fixed to count raw entries, but this endpoint still filtered by address — so the estate row read "1 member" while the drawer opened from that row rendered 「该组没有直接成员」, i.e. the desk contradicted itself on exactly the case the raw-count fix was written for. """ user, token = employee_user me = await db_session.get(IdentityPerson, user.id) me.work_email = "shuai.feng@pingcap.com" group = await _make_group(db_session, email="all@pingcap.com", owners=[me]) await db_session.commit() patch_driver.list_group_members.return_value = [ {"email": "shuai.feng@pingcap.com", "role": "OWNER", "type": "USER"}, {"id": "C0299abcd", "role": "MEMBER", "type": "CUSTOMER"}, ] r = await client.get( f"/api/v1/email-groups/{group.id}/members", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200, r.text members = r.json()["data"]["members"] assert len(members) == 2 addressless = [m for m in members if m["email"] is None] assert len(addressless) == 1 assert addressless[0]["member_id"] == "C0299abcd" assert addressless[0]["type"] == "CUSTOMER"