"""The empty-static-group cleanup workbench. Deleting a mail group is a Google HARD delete with no undelete, so every test here is about a guard whose absence would destroy a live address: * the **pool predicate** must not admit a NULL — neither an undetermined static/dynamic kind nor a never-synced member count; * an **indeterminate signal** must never render as clean, because "we could not check" and "we checked, it is clean" are different answers and only one of them is safe to act on; * the executor must **re-verify live membership per group**, because the candidate list the operator is looking at can be a sync interval old; * one refusal must never abort the batch, and every deletion must leave an audit row that still means something after the group is gone. """ from __future__ import annotations import uuid from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient from sqlalchemy import select from app.audit.models import AuditLog from app.domains.email_group.models import EmailGroup from app.integrations.drivers.google_workspace import WorkspaceOpResult BASE = "/api/v1/email-groups/manage" DRIVER_PATH = "app.domains.email_group.service.EmailGroupService._get_driver" def _auth(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} def _driver( *, members: dict[str, list[dict]] | None = None, parents: dict[str, list[dict]] | None = None, missing: set[str] | None = None, delete_ok: bool = True, ) -> AsyncMock: d = AsyncMock() members = members or {} parents = parents or {} missing = missing or set() async def _list_members(group_email: str): return members.get(group_email, []) async def _groups_for(user_email: str, *, raise_on_missing: bool = False): # The real driver swallows Google's 404 into ``[]`` by default; # the cleanup probe passes ``raise_on_missing=True`` precisely so # a stale address raises instead of reading as "not nested". # Accepting the kwarg is what keeps this fake honest about the # contract the production call actually uses. if user_email in missing: if raise_on_missing: raise RuntimeError("404 not in directory") return [] return parents.get(user_email, []) d.list_group_members.side_effect = _list_members d.list_groups_for_user.side_effect = _groups_for d.delete_group.return_value = ( WorkspaceOpResult(ok=True) if delete_ok else WorkspaceOpResult(ok=False, error_message="Google refused") ) return d async def _mk_group( db, *, email: str, name: str | None = None, is_dynamic: bool | None = False, members: int | None = 0, status: str = "active", aliases: list[str] | None = None, never_alias_synced: bool = False, description: str | None = None, ) -> EmailGroup: """Build one candidate row. ``aliases`` defaults to ``[]`` — "synced, and Google has none" — because that is the state every test that is not ABOUT aliases means. The column is tri-state (NULL = the alias list has never been read), and a NULL now raises an ``alias_unknown`` INDETERMINATE signal that pushes the row out of ``risk: low``, so a NULL default would make every unrelated assertion in this file depend on a fact it never states. Pass ``never_alias_synced=True`` to get the NULL on purpose. """ row = EmailGroup( email=email, name=name or email.split("@")[0], description=description, status=status, is_dynamic=is_dynamic, direct_members_count=members, aliases=None if never_alias_synced else list(aliases or []), ) db.add(row) await db.commit() return row def _signal_keys(item: dict) -> set[str]: return {s["key"] for s in item["signals"]} def _levels(item: dict) -> set[str]: return {s["level"] for s in item["signals"]} # ── RBAC ──────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_employee_cannot_list_candidates( client: AsyncClient, db_session, employee_user, ): _u, token = employee_user r = await client.get(f"{BASE}/cleanup-candidates", headers=_auth(token)) assert r.status_code == 403, r.text @pytest.mark.asyncio async def test_employee_cannot_cleanup( client: AsyncClient, db_session, employee_user, ): _u, token = employee_user group = await _mk_group(db_session, email="test@pingcap.com") r = await client.post( f"{BASE}/cleanup", json={"group_ids": [str(group.id)], "reason": "spring cleaning 2026"}, headers=_auth(token), ) assert r.status_code == 403, r.text await db_session.refresh(group) assert group.status == "active" # ── Pool predicate ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_pool_admits_only_empty_static_active_groups( client: AsyncClient, db_session, hr_user, ): _u, token = hr_user await _mk_group(db_session, email="empty@pingcap.com", members=0) await _mk_group(db_session, email="populated@pingcap.com", members=5) await _mk_group( db_session, email="dynamic@pingcap.com", is_dynamic=True, members=0, ) # ``is_dynamic IS NULL`` — kind never determined. Might be dynamic, # so it must not enter a STATIC cleanup pool. await _mk_group( db_session, email="mystery@pingcap.com", is_dynamic=None, members=0, ) # ``direct_members_count IS NULL`` — never synced. A group nobody # has ever synced is not a group we know is empty. await _mk_group(db_session, email="unsynced@pingcap.com", members=None) await _mk_group( db_session, email="gone@pingcap.com", members=0, status="deleted", ) with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] assert [i["email"] for i in d["items"]] == ["empty@pingcap.com"] assert d["total"] == 1 @pytest.mark.asyncio async def test_response_always_states_its_blind_spots( client: AsyncClient, db_session, hr_user, ): """The surface must never present itself as an exhaustive check. And the gaps it declares must be the gaps it actually has. It used to declare ``cloud_iam_principal`` — "Helix cannot see whether a group is a cloud IAM principal" — which was wrong in the direction that matters: Helix's OWN data-access domain grants BigQuery dataset access to ``email_groups`` rows and compiles it into real GCP IAM, all of it reachable through an indexed FK. What is genuinely invisible is IAM added by hand in the GCP console (``cloud_iam_outside_helix``), and mail actually flowing through the address. Saying so on every response is what stops a caller — human or model — from treating the list as a safety clearance; declaring the WRONG gap teaches them to discount the field. """ _u, token = hr_user with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert set(r.json()["data"]["known_blind_spots"]) == { "cloud_iam_outside_helix", "mail_traffic", } # ── Signals ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_nested_group_is_blocking( client: AsyncClient, db_session, hr_user, ): """A memberless group can still be a delivery hop inside another. Deleting it silently breaks that path, which is why this is the strongest signal the workbench can compute. """ _u, token = hr_user await _mk_group(db_session, email="hop@pingcap.com") driver = _driver(parents={ "hop@pingcap.com": [{"email": "all-eng@pingcap.com"}], }) with patch(DRIVER_PATH, return_value=driver): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) item = r.json()["data"]["items"][0] assert "nested_member_of" in _signal_keys(item) assert item["risk"] == "blocking" nested = next( s for s in item["signals"] if s["key"] == "nested_member_of" ) assert nested["detail"]["parents"] == ["all-eng@pingcap.com"] @pytest.mark.asyncio async def test_self_reference_is_not_nesting( client: AsyncClient, db_session, hr_user, ): """Google can echo the group itself; that is not a parent.""" _u, token = hr_user await _mk_group(db_session, email="solo@pingcap.com") driver = _driver(parents={ "solo@pingcap.com": [{"email": "solo@pingcap.com"}], }) with patch(DRIVER_PATH, return_value=driver): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) item = r.json()["data"]["items"][0] assert "nested_member_of" not in _signal_keys(item) @pytest.mark.asyncio async def test_failed_nesting_probe_is_indeterminate_not_clean( client: AsyncClient, db_session, hr_user, ): """The single most important distinction in this whole feature. A probe we could not run must leave the row in ``review``. If it collapsed into ``low`` — the same bucket as "checked, nothing found" — an operator sweeping the low-risk rows would delete groups whose nesting was never actually verified. """ _u, token = hr_user await _mk_group(db_session, email="unknown@pingcap.com") driver = _driver() driver.list_groups_for_user.side_effect = RuntimeError("403 no scope") with patch(DRIVER_PATH, return_value=driver): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) d = r.json()["data"] item = d["items"][0] assert "nesting_unknown" in _signal_keys(item) assert "indeterminate" in _levels(item) assert item["risk"] == "review" assert d["nesting_probe_error"] @pytest.mark.asyncio async def test_probe_disabled_is_also_indeterminate( client: AsyncClient, db_session, hr_user, ): """Skipping the probe for speed must not look like a clean result.""" _u, token = hr_user await _mk_group(db_session, email="fast@pingcap.com") with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", params={"probe_nesting": False}, headers=_auth(token), ) d = r.json()["data"] assert d["nesting_probed"] is False assert d["items"][0]["risk"] == "review" assert "nesting_unknown" in _signal_keys(d["items"][0]) @pytest.mark.asyncio async def test_real_alias_is_blocking_and_google_mirror_is_not( client: AsyncClient, db_session, hr_user, ): """Only a REAL second address counts. Google mirrors ~every group into ``@.test-google-a.com``. Counting those would flag essentially the whole pool (131 of 133 on prod), which is the same as flagging nothing. """ _u, token = hr_user await _mk_group( db_session, email="noise@pingcap.com", aliases=["noise@pingcap.com.test-google-a.com"], ) await _mk_group( db_session, email="real@pingcap.com", aliases=[ "real@pingcap.com.test-google-a.com", "actual-alias@pingcap.com", ], ) with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) by_email = {i["email"]: i for i in r.json()["data"]["items"]} assert "has_real_alias" not in _signal_keys(by_email["noise@pingcap.com"]) assert "has_real_alias" in _signal_keys(by_email["real@pingcap.com"]) assert by_email["real@pingcap.com"]["risk"] == "blocking" @pytest.mark.asyncio async def test_helix_config_reference_is_blocking( client: AsyncClient, db_session, hr_user, ): """Helix's own configuration pointing at the address. This is the one reference class Helix can prove, and it is the one where deletion breaks a Helix flow rather than somebody else's. """ from app.integrations.models import PlatformSetting _u, token = hr_user await _mk_group(db_session, email="alerts@pingcap.com") await _mk_group(db_session, email="unused@pingcap.com") db_session.add(PlatformSetting( key="some.notify.target", value={"to": ["alerts@pingcap.com"]}, )) await db_session.commit() with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) by_email = {i["email"]: i for i in r.json()["data"]["items"]} refs = next( s for s in by_email["alerts@pingcap.com"]["signals"] if s["key"] == "referenced_in_helix_config" ) assert "platform_setting:some.notify.target" in refs["detail"]["references"] assert by_email["alerts@pingcap.com"]["risk"] == "blocking" assert "referenced_in_helix_config" not in _signal_keys( by_email["unused@pingcap.com"] ) @pytest.mark.asyncio async def test_infra_name_is_a_warning_not_a_verdict( client: AsyncClient, db_session, hr_user, ): """The cloud-IAM blind spot, surfaced honestly. ``gcp-billing-admins@`` is almost certainly a GCP IAM principal, and Helix has no way to know. Flagging the NAME as a heuristic is truthful; emitting "no cloud reference" would be a confident lie. """ _u, token = hr_user await _mk_group(db_session, email="gcp-billing-admins@pingcap.com") await _mk_group(db_session, email="bookclub@pingcap.com") with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) by_email = {i["email"]: i for i in r.json()["data"]["items"]} infra = by_email["gcp-billing-admins@pingcap.com"] sig = next( s for s in infra["signals"] if s["key"] == "possible_cloud_principal" ) assert sig["level"] == "warning" assert infra["risk"] == "review" assert "possible_cloud_principal" not in _signal_keys( by_email["bookclub@pingcap.com"] ) @pytest.mark.asyncio async def test_deprecated_and_test_markers_are_info_only( client: AsyncClient, db_session, hr_user, ): """These are the safest rows, so their signals must NOT raise risk.""" _u, token = hr_user await _mk_group( db_session, email="oldteam@pingcap.com", name="【弃用】Old Team", ) await _mk_group(db_session, email="test44@pingcap.com", name="test44") with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) by_email = {i["email"]: i for i in r.json()["data"]["items"]} assert "marked_deprecated" in _signal_keys(by_email["oldteam@pingcap.com"]) assert by_email["oldteam@pingcap.com"]["risk"] == "low" assert "looks_like_test" in _signal_keys(by_email["test44@pingcap.com"]) assert by_email["test44@pingcap.com"]["risk"] == "low" # ── Execute ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_cleanup_deletes_and_audits( client: AsyncClient, db_session, hr_user, ): _u, token = hr_user group = await _mk_group( db_session, email="test44@pingcap.com", name="test44", aliases=["test44@pingcap.com.test-google-a.com", "test66@pingcap.com"], ) driver = _driver() with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(group.id)], "reason": "确认无用的测试组,批量清理", }, headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] assert d["deleted"] == 1 assert d["results"][0]["outcome"] == "deleted" driver.delete_group.assert_awaited_once_with(email="test44@pingcap.com") await db_session.refresh(group) assert group.status == "deleted" rows = list((await db_session.execute( select(AuditLog).where(AuditLog.action == "email_group.cleanup_deleted") )).scalars()) assert len(rows) == 1 detail = rows[0].detail # The snapshot is what keeps the row meaningful after the group is # gone from Google. assert detail["group_email"] == "test44@pingcap.com" assert detail["name"] == "test44" assert detail["aliases"] == ["test66@pingcap.com"] assert detail["reason"] == "确认无用的测试组,批量清理" assert detail["batch_id"] == d["batch_id"] assert detail["sequence"] == 0 batch = list((await db_session.execute( select(AuditLog).where(AuditLog.action == "email_group.cleanup_batch") )).scalars()) assert len(batch) == 1 assert batch[0].resource_id == d["batch_id"] assert batch[0].detail["deleted"] == 1 @pytest.mark.asyncio async def test_cleanup_refuses_a_group_that_gained_members( client: AsyncClient, db_session, hr_user, ): """The guard between "stale list" and "deleted a live group". The candidate list an operator is looking at can be a whole sync interval old. Trusting the mirror here would delete a group that somebody joined five minutes ago. """ _u, token = hr_user group = await _mk_group(db_session, email="busy@pingcap.com") driver = _driver(members={ "busy@pingcap.com": [{"email": "someone@pingcap.com", "role": "MEMBER"}], }) with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={"group_ids": [str(group.id)], "reason": "batch cleanup run"}, headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] assert d["deleted"] == 0 assert d["refused"] == 1 assert d["results"][0]["live_member_count"] == 1 driver.delete_group.assert_not_awaited() await db_session.refresh(group) assert group.status == "active" @pytest.mark.asyncio async def test_cleanup_refuses_when_google_membership_is_unreadable( client: AsyncClient, db_session, hr_user, ): """Unable to confirm ⇒ refuse. Never delete on an unverified read.""" _u, token = hr_user group = await _mk_group(db_session, email="opaque@pingcap.com") driver = _driver() driver.list_group_members.side_effect = RuntimeError("Google 503") with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={"group_ids": [str(group.id)], "reason": "batch cleanup run"}, headers=_auth(token), ) d = r.json()["data"] assert d["refused"] == 1 driver.delete_group.assert_not_awaited() await db_session.refresh(group) assert group.status == "active" @pytest.mark.asyncio async def test_cleanup_refuses_a_dynamic_or_undetermined_group( client: AsyncClient, db_session, hr_user, ): """Re-checked at execute time, not trusted from the request. A client could post any id — including one that has since been made dynamic, or whose kind was never determined. """ _u, token = hr_user dyn = await _mk_group( db_session, email="dyn@pingcap.com", is_dynamic=True, members=0, ) unknown = await _mk_group( db_session, email="unk@pingcap.com", is_dynamic=None, members=0, ) populated = await _mk_group( db_session, email="pop@pingcap.com", members=3, ) driver = _driver() with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(dyn.id), str(unknown.id), str(populated.id)], "reason": "batch cleanup run", }, headers=_auth(token), ) d = r.json()["data"] assert d["deleted"] == 0 assert d["refused"] == 3 driver.delete_group.assert_not_awaited() @pytest.mark.asyncio async def test_one_refusal_never_aborts_the_batch( client: AsyncClient, db_session, hr_user, ): """A batch that died mid-way would leave the operator unable to tell which groups went through.""" _u, token = hr_user ok1 = await _mk_group(db_session, email="a@pingcap.com") busy = await _mk_group(db_session, email="b@pingcap.com") ok2 = await _mk_group(db_session, email="c@pingcap.com") driver = _driver(members={ "b@pingcap.com": [{"email": "x@pingcap.com", "role": "MEMBER"}], }) with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(ok1.id), str(busy.id), str(ok2.id)], "reason": "batch cleanup run", }, headers=_auth(token), ) d = r.json()["data"] assert d["deleted"] == 2 assert d["refused"] == 1 for row, expected in ((ok1, "deleted"), (busy, "active"), (ok2, "deleted")): await db_session.refresh(row) assert row.status == expected, row.email @pytest.mark.asyncio async def test_google_delete_failure_leaves_the_row_active( client: AsyncClient, db_session, hr_user, ): """Never soft-delete locally when Google refused. A local tombstone for a group that still exists on Google is the harder reconciliation: every Helix surface then treats a live address as gone. """ _u, token = hr_user group = await _mk_group(db_session, email="stuck@pingcap.com") with patch(DRIVER_PATH, return_value=_driver(delete_ok=False)): r = await client.post( f"{BASE}/cleanup", json={"group_ids": [str(group.id)], "reason": "batch cleanup run"}, headers=_auth(token), ) d = r.json()["data"] assert d["failed"] == 1 assert "Google refused" in d["results"][0]["reason"] await db_session.refresh(group) assert group.status == "active" assert list((await db_session.execute( select(AuditLog).where(AuditLog.action == "email_group.cleanup_deleted") )).scalars()) == [] @pytest.mark.asyncio async def test_reason_must_be_a_sentence( client: AsyncClient, db_session, hr_user, ): _u, token = hr_user group = await _mk_group(db_session, email="a@pingcap.com") with patch(DRIVER_PATH, return_value=_driver()): r = await client.post( f"{BASE}/cleanup", json={"group_ids": [str(group.id)], "reason": "x"}, headers=_auth(token), ) assert r.status_code == 422, r.text @pytest.mark.asyncio async def test_batch_cap_is_enforced(client: AsyncClient, db_session, hr_user): _u, token = hr_user with patch(DRIVER_PATH, return_value=_driver()): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(uuid.uuid4()) for _ in range(25)], "reason": "batch cleanup run", }, headers=_auth(token), ) assert r.status_code == 422, r.text @pytest.mark.asyncio async def test_unknown_id_is_reported_not_fatal( client: AsyncClient, db_session, hr_user, ): _u, token = hr_user ghost = uuid.uuid4() real = await _mk_group(db_session, email="a@pingcap.com") with patch(DRIVER_PATH, return_value=_driver()): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(ghost), str(real.id)], "reason": "batch cleanup run", }, headers=_auth(token), ) d = r.json()["data"] assert d["not_found"] == 1 assert d["deleted"] == 1 @pytest.mark.asyncio async def test_duplicate_ids_are_collapsed( client: AsyncClient, db_session, hr_user, ): """A double-submitted id must be one operation, not two. The second pass would hit Google's 404 and report ``already_absent`` — harmless but a lie about what happened. """ _u, token = hr_user group = await _mk_group(db_session, email="a@pingcap.com") driver = _driver() with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(group.id), str(group.id)], "reason": "batch cleanup run", }, headers=_auth(token), ) d = r.json()["data"] assert d["requested"] == 1 assert d["deleted"] == 1 assert driver.delete_group.await_count == 1 @pytest.mark.asyncio async def test_cleanup_row_stays_readable_in_the_operation_log( client: AsyncClient, db_session, hr_user, ): """The whole point of soft-deleting the local row. After a sweep, "what happened to that address?" must still be answerable from the 邮件组管理 operation log — searched by the address the group used to have. """ _u, token = hr_user group = await _mk_group(db_session, email="careers@pingcap.com") with patch(DRIVER_PATH, return_value=_driver()): await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(group.id)], "reason": "confirmed unused with HR", }, headers=_auth(token), ) r = await client.get( f"{BASE}/audit-log", params={"search": "careers", "category": "cleanup"}, headers=_auth(token), ) assert r.status_code == 200, r.text rows = r.json()["data"] assert len(rows) == 1 assert rows[0]["action"] == "email_group.cleanup_deleted" assert rows[0]["group_email"] == "careers@pingcap.com" assert rows[0]["group_status"] == "deleted" assert rows[0]["detail"]["reason"] == "confirmed unused with HR" @pytest.mark.asyncio async def test_one_failed_probe_marks_only_that_row_unknown( client: AsyncClient, db_session, hr_user, ): """The PER-ADDRESS indeterminate branch, not the wholesale one. A page of 50 rows where one probe 403s or times out is the realistic production shape. If that row emitted no nesting signal at all, ``_risk_of`` would return ``low`` — indistinguishable from "checked, nothing found" — and an operator sweeping the low-risk rows would hard-delete a group whose nesting was never verified. """ _u, token = hr_user await _mk_group(db_session, email="ok@pingcap.com") await _mk_group(db_session, email="broken@pingcap.com") driver = _driver() async def _groups_for(user_email: str, *, raise_on_missing: bool = False): if user_email == "broken@pingcap.com": raise RuntimeError("403 no scope") return [] driver.list_groups_for_user.side_effect = _groups_for with patch(DRIVER_PATH, return_value=driver): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) d = r.json()["data"] # The probe as a whole DID run, so this is not the wholesale branch. assert d["nesting_probed"] is True by_email = {i["email"]: i for i in d["items"]} assert "nesting_unknown" in _signal_keys(by_email["broken@pingcap.com"]) assert by_email["broken@pingcap.com"]["risk"] == "review" # …while its neighbour is genuinely clean. assert "nesting_unknown" not in _signal_keys(by_email["ok@pingcap.com"]) assert by_email["ok@pingcap.com"]["risk"] == "low" @pytest.mark.asyncio async def test_a_stale_address_is_unknown_not_unnested( client: AsyncClient, db_session, hr_user, ): """Google's 404 must not read as "nested nowhere". ``list_groups_for_user`` swallows 404 into ``[]`` for its original caller (a contractor with no Workspace account really is in zero groups). For a GROUP address a 404 means the address is stale — renamed in the console since the last sync — and ``[]`` would clear the row to ``low`` and make it freely selectable for an irreversible hard delete. """ _u, token = hr_user await _mk_group(db_session, email="renamed@pingcap.com") driver = _driver(missing={"renamed@pingcap.com"}) with patch(DRIVER_PATH, return_value=driver): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) item = r.json()["data"]["items"][0] assert "nesting_unknown" in _signal_keys(item) assert item["risk"] == "review" @pytest.mark.asyncio async def test_config_reference_match_is_boundary_anchored( client: AsyncClient, db_session, hr_user, ): """A short address must not inherit a longer one's references. An unanchored substring made ``bot@pingcap.com`` match ``tidbcloud-bot@pingcap.com`` and produce a BLOCKING signal whose evidence was a fabrication — on a surface whose entire premise is that each signal is trustworthy. """ from app.integrations.models import PlatformSetting _u, token = hr_user await _mk_group(db_session, email="bot@pingcap.com") db_session.add(PlatformSetting( key="some.alerting.target", value={"to": ["tidbcloud-bot@pingcap.com"]}, )) await db_session.commit() with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) item = r.json()["data"]["items"][0] assert item["email"] == "bot@pingcap.com" assert "referenced_in_helix_config" not in _signal_keys(item) # It is still ``review`` — ``bot@`` trips the infrastructure-name # HEURISTIC, which is correct and is labelled as a heuristic. What # must not happen is the BLOCKING verdict, which disables the # checkbox and shows fabricated evidence. assert item["risk"] == "review" assert "possible_cloud_principal" in _signal_keys(item) # Control: an address that really is referenced still blocks. await _mk_group(db_session, email="alerts@pingcap.com") db_session.add(PlatformSetting( key="another.target", value={"to": ["alerts@pingcap.com"]}, )) await db_session.commit() with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) by_email = {i["email"]: i for i in r.json()["data"]["items"]} assert "referenced_in_helix_config" in _signal_keys( by_email["alerts@pingcap.com"] ) @pytest.mark.asyncio async def test_audit_sequence_follows_the_operator_ordering( client: AsyncClient, db_session, hr_user, ): """``sequence`` must be the position the operator saw, not a counter. ``audit_logs.created_at`` is a whole-second DATETIME, so a 20-group batch lands inside one or two seconds and cannot be ordered by time. A success counter would renumber the rows after any refusal and the batch would no longer replay in screen order. """ _u, token = hr_user a = await _mk_group(db_session, email="a@pingcap.com") busy = await _mk_group(db_session, email="b@pingcap.com") c = await _mk_group(db_session, email="c@pingcap.com") driver = _driver(members={ "b@pingcap.com": [{"email": "x@pingcap.com", "role": "MEMBER"}], }) with patch(DRIVER_PATH, return_value=driver): await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(a.id), str(busy.id), str(c.id)], "reason": "batch cleanup run", }, headers=_auth(token), ) rows = list((await db_session.execute( select(AuditLog).where(AuditLog.action == "email_group.cleanup_deleted") )).scalars()) by_email = {r.detail["group_email"]: r.detail["sequence"] for r in rows} # b@ was refused at index 1, so c@ keeps index 2 — not 1. assert by_email == {"a@pingcap.com": 0, "c@pingcap.com": 2} @pytest.mark.asyncio async def test_local_bookkeeping_failure_still_reports_the_manifest( client: AsyncClient, db_session, hr_user, ): """Google delete succeeded, local write failed — say exactly that. An unguarded DB error on group 2 of 3 would propagate, skip group 3, never write the batch summary, and leave the operator with a 500 and no idea which addresses were hard-deleted — for an operation with no undelete. """ from app.audit.service import AuditService _u, token = hr_user a = await _mk_group(db_session, email="a@pingcap.com") b = await _mk_group(db_session, email="b@pingcap.com") calls = {"n": 0} real_log = AuditService.log async def _flaky(self, *args, **kwargs): if kwargs.get("action") == "email_group.cleanup_deleted": calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("audit table on fire") return await real_log(self, *args, **kwargs) with patch(DRIVER_PATH, return_value=_driver()), \ patch.object(AuditService, "log", _flaky): r = await client.post( f"{BASE}/cleanup", json={ "group_ids": [str(a.id), str(b.id)], "reason": "batch cleanup run", }, headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] # The batch ran to completion and the manifest survived. assert d["requested"] == 2 assert d["deleted"] == 1 assert d["deleted_record_failed"] == 1 outcomes = {row["email"]: row["outcome"] for row in d["results"]} assert outcomes["a@pingcap.com"] == "deleted_record_failed" assert outcomes["b@pingcap.com"] == "deleted" # ── Round-2 review fixes ──────────────────────────────────────────── @pytest.mark.asyncio async def test_cleanup_refuses_a_member_google_reports_without_an_address( client: AsyncClient, db_session, hr_user, ): """The last gate before an irreversible delete must count ENTRIES. Google returns whole-domain membership as ``type: "CUSTOMER"`` with an ``id`` and no ``email`` (that is how "everyone at pingcap.com is in this group" is expressed). The live re-verify used to reduce the roster to lowercased addresses and test THAT for emptiness, so a group whose only member was such an entry read as empty — and got hard-deleted, with an audit row asserting it had no members. Nothing else in the pipeline can catch this: the mirrored ``direct_members_count`` came from the same email-keyed reduction, so both the pool predicate and its live confirmation agreed on the same wrong answer. """ _u, token = hr_user group = await _mk_group(db_session, email="all-hands@pingcap.com") driver = _driver(members={ "all-hands@pingcap.com": [ {"id": "C0299abcd", "role": "MEMBER", "type": "CUSTOMER"}, ], }) with patch(DRIVER_PATH, return_value=driver): r = await client.post( f"{BASE}/cleanup", json={"group_ids": [str(group.id)], "reason": "batch cleanup run"}, headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] assert d["deleted"] == 0 assert d["refused"] == 1 assert d["results"][0]["live_member_count"] == 1 # No address to report — the count is what refuses, not the list. assert d["results"][0]["live_members"] == [] driver.delete_group.assert_not_awaited() await db_session.refresh(group) assert group.status == "active" @pytest.mark.asyncio async def test_a_never_alias_synced_group_is_not_low_risk( client: AsyncClient, db_session, hr_user, ): """NULL aliases and ``[]`` aliases are different facts. ``email_groups.aliases`` is tri-state: NULL means the alias list has never been read, ``[]`` means it was read and Google has none. The ``has_real_alias`` BLOCKING signal was decided on ``real_aliases(row.aliases)``, which maps both to ``[]`` — so a group nobody has ever alias-synced was indistinguishable from a confirmed-clean one and could reach ``risk: low``, the bucket whose whole meaning is "we checked". A group carrying an unread second address is exactly the one whose deletion bounces mail nobody knew was flowing, so the honest answer is INDETERMINATE — the same treatment the nesting probe gets when it cannot run. """ _u, token = hr_user await _mk_group(db_session, email="unknown@pingcap.com", never_alias_synced=True) await _mk_group(db_session, email="known@pingcap.com", aliases=[]) with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert r.status_code == 200, r.text by_email = {i["email"]: i for i in r.json()["data"]["items"]} assert "alias_unknown" in _signal_keys(by_email["unknown@pingcap.com"]) assert by_email["unknown@pingcap.com"]["risk"] == "review" # The confirmed-clean neighbour keeps its low-risk bucket, so this # is not just "everything is review now". assert "alias_unknown" not in _signal_keys(by_email["known@pingcap.com"]) assert by_email["known@pingcap.com"]["risk"] == "low" @pytest.mark.asyncio async def test_a_bigquery_access_principal_is_blocking( client: AsyncClient, db_session, hr_user, ): """The blind spot the module used to DECLARE was partly false. The docstring said no Helix query could tell whether a group is a cloud IAM principal. But Helix's own data-access domain grants BigQuery dataset access to ``email_groups`` rows — indexed FK, compiled into real GCP IAM as ``group:`` — so the worst case (delete the group, revoke live analyst access, orphan a binding whose reconcile can never converge) was both knowable and unreported, while the response told the operator the check was impossible. """ from app.domains.data_access.models import ( DataAccessBinding, DataAccessEntitlement, DataAccessResource, ) from app.integrations.models import Integration _u, token = hr_user group = await _mk_group(db_session, email="analytics@pingcap.com") integration = Integration( id=str(uuid.uuid4()), provider="gcp", kind="cloud_account", name="BQ", slug=f"bq-{uuid.uuid4().hex[:8]}", ) db_session.add(integration) await db_session.flush() resource = DataAccessResource( integration_id=integration.id, project_id="prod-analytics", dataset_id="finance_gl", display_name="Finance GL", owner_source="manual", source_version="v1", source_etag="etag-1", ) ent = DataAccessEntitlement(entitlement="reader") db_session.add_all([resource, ent]) await db_session.flush() db_session.add(DataAccessBinding( resource_id=resource.id, principal_email_group_id=group.id, principal_type_snapshot="email_group", principal_email_snapshot=group.email, entitlement_id=ent.id, entitlement_snapshot="reader", entitlement_version_snapshot="v1", status="active", )) await db_session.commit() with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert r.status_code == 200, r.text item = {i["email"]: i for i in r.json()["data"]["items"]}[ "analytics@pingcap.com" ] sig = {s["key"]: s for s in item["signals"]} assert "data_access_principal" in sig assert sig["data_access_principal"]["level"] == "blocking" assert sig["data_access_principal"]["detail"]["live_bindings"] == 1 assert item["risk"] == "blocking" @pytest.mark.asyncio async def test_a_test_domain_mirror_address_is_not_a_helix_reference( client: AsyncClient, db_session, hr_user, ): """The reference match needs a boundary on the RIGHT too. This tenant's Google test-domain mirrors are literally ``.test-google-a.com``. A left-only anchor made ``ops@pingcap.com`` match inside ``ops@pingcap.com.test-google-a.com``, so a config value naming the MIRROR produced a BLOCKING ``referenced_in_helix_config`` signal against the real group — fabricated evidence on a surface whose premise is that each signal is trustworthy. """ from app.integrations.models import PlatformSetting _u, token = hr_user await _mk_group(db_session, email="ops@pingcap.com", aliases=[]) db_session.add(PlatformSetting( key="some.mirror.target", value="ops@pingcap.com.test-google-a.com", )) await db_session.commit() with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert r.status_code == 200, r.text item = {i["email"]: i for i in r.json()["data"]["items"]}["ops@pingcap.com"] assert "referenced_in_helix_config" not in _signal_keys(item) @pytest.mark.asyncio async def test_rows_past_the_probe_budget_say_not_probed_not_failed( client: AsyncClient, db_session, hr_user, monkeypatch, ): """"Not attempted" and "attempted and failed" are different facts. The nesting probe is capped per request (one Google call per row, on a shared thread pool and a shared Directory quota). Rows past the cap come back by ABSENCE from the probe result — which landed on the "probe failed for this address" branch, telling the operator their probe failed for a limit Helix imposed on itself. They would open a Google-permissions investigation for nothing. Both are INDETERMINATE, so neither row can reach ``risk: low``; only the stated reason differs. """ from app.domains.email_group import cleanup_service as cs _u, token = hr_user monkeypatch.setattr(cs, "_MAX_PROBED_ROWS", 1) await _mk_group(db_session, email="a-first@pingcap.com", aliases=[]) await _mk_group(db_session, email="z-last@pingcap.com", aliases=[]) with patch(DRIVER_PATH, return_value=_driver()): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] by_email = {i["email"]: i for i in d["items"]} # Ordered by email, so the first row is inside the budget… assert "nesting_not_probed" not in _signal_keys(by_email["a-first@pingcap.com"]) assert by_email["a-first@pingcap.com"]["risk"] == "low" # …and the second is past it: unknown, but not "failed". second = by_email["z-last@pingcap.com"] assert "nesting_not_probed" in _signal_keys(second) assert "nesting_unknown" not in _signal_keys(second) assert second["risk"] == "review" # The budget travels in its OWN fields, never in # ``nesting_probe_error`` — the only banner that renders that string # is headed "some nesting probes failed", so folding the budget into # it announced failures when zero probes had failed, reasserting at # page level the lie this per-row signal removes. assert d["nesting_probe_error"] is None assert d["nesting_rows_not_probed"] == 1 assert d["nesting_probe_budget"] == 1 # …and the row's own signal carries a readable reason, because the # frontend's evidence chain reads ``reason`` and a detail key # nothing reads renders a tag with no tooltip at all. sig = {x["key"]: x for x in second["signals"]} assert "不是探测失败" in sig["nesting_not_probed"]["detail"]["reason"] @pytest.mark.asyncio async def test_a_failed_probe_on_a_mixed_case_row_is_not_called_unprobed( client: AsyncClient, db_session, hr_user, ): """A real failure must not be filed as a Helix-imposed cap. ``_probe_nesting`` keys its result by the address as STORED, and nothing lowercases ``email_groups.email``. The failure path keyed by the lowercased address instead — harmless while failures were dropped from the dict, but once they were kept (so "failed" and "never attempted" could be told apart) a mixed-case row missed the lookup and fell straight into the "never attempted" branch. The operator would then see 「未探测(超出本次预算)」 on a 2-row pool where no budget was hit, while the page banner said a probe failed: a genuine permissions or stale-address problem reported as a cap Helix put on itself. """ _u, token = hr_user await _mk_group(db_session, email="Ops@pingcap.com", aliases=[]) # A second, healthy row so this is a PARTIAL failure — the case the # page banner describes. With one row it would take the wholesale # "every probe failed" branch and prove less. await _mk_group(db_session, email="fine@pingcap.com", aliases=[]) driver = _driver(missing={"ops@pingcap.com"}) with patch(DRIVER_PATH, return_value=driver): r = await client.get( f"{BASE}/cleanup-candidates", headers=_auth(token), ) assert r.status_code == 200, r.text d = r.json()["data"] item = {i["email"]: i for i in d["items"]}["Ops@pingcap.com"] keys = _signal_keys(item) assert "nesting_unknown" in keys assert "nesting_not_probed" not in keys assert item["risk"] == "review" assert "probes failed" in (d["nesting_probe_error"] or "") assert d["nesting_rows_not_probed"] == 0