"""Tests for the Google Workspace driver methods added 2026-05-02: ``create_user``, ``list_all_users``, ``list_group_members``. Reuses the ``patch_service`` fluent-stub pattern from ``test_google_workspace_ops.py``. Where a method paginates we drive two pages with a ``nextPageToken`` to confirm the loop chains correctly. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any import httplib2 import pytest from googleapiclient.errors import HttpError from app.integrations.driver_base import ProvisioningContext from app.integrations.drivers import google_workspace as gw from app.integrations.drivers.google_workspace import ( GOOGLE_TAG_SCOPE_MISSING, GOOGLE_TAG_SUCCESSOR_NOT_IN_DIRECTORY, GOOGLE_TAG_TRANSFER_PERMISSION_DENIED, GoogleApiError, GoogleWorkspaceDriver, ) _CONFIG = { "service_account_json": "{\"client_email\":\"sa@example.iam.gserviceaccount.com\"}", "impersonate_subject_email": "admin@example.com", } def _driver() -> GoogleWorkspaceDriver: return GoogleWorkspaceDriver(integration_id="gw-test", config=_CONFIG) @dataclass class _CallLog: chain: list[tuple[str, tuple, dict]] = field(default_factory=list) class _Leaf: def __init__(self, response: Any | Exception): self._response = response def execute(self) -> Any: if isinstance(self._response, Exception): raise self._response return self._response class _FluentNode: def __init__(self, log: _CallLog, path: tuple[str, ...], routes: dict[str, Any]): self._log = log self._path = path self._routes = routes def __getattr__(self, name: str): def _call(*args: Any, **kwargs: Any): self._log.chain.append((name, args, kwargs)) new_path = self._path + (name,) key = ".".join(new_path) if key in self._routes: response = self._routes[key] # Routes may be an iterator of pages → return next page # each time .execute() is called. if hasattr(response, "__next__"): try: return _Leaf(next(response)) except StopIteration: return _Leaf({}) return _Leaf(response) return _FluentNode(self._log, new_path, self._routes) return _call def _make_http_error(status: int, reason: str = "err") -> HttpError: resp = httplib2.Response({"status": status, "reason": reason}) resp.reason = reason return HttpError(resp, b"{\"error\": {\"message\": \"boom\"}}") @pytest.fixture def patch_service(monkeypatch): routes: dict[str, Any] = {} log = _CallLog() def _set(new_routes: dict[str, Any]) -> _CallLog: routes.clear() routes.update(new_routes) log.chain.clear() return log def _fake_build(sa_json: str, subject: str, scopes: list[str]): return _FluentNode(log, (), routes) monkeypatch.setattr(gw, "_build_directory_service", _fake_build) return _set # ─── create_user ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_create_user_happy_path(patch_service): log = patch_service({ "users.insert": { "primaryEmail": "tom.smith@pingcap.com", "id": "1234567890", "orgUnitPath": "/Employees", }, }) result = await _driver().create_user( primary_email="tom.smith@pingcap.com", given_name="Tom", family_name="Smith", password="Init-Pwd!2345678", org_unit_path="/Employees", ) assert result.ok is True assert result.already_exists is False assert result.data["primary_email"] == "tom.smith@pingcap.com" assert result.data["id"] == "1234567890" # The body sent to Google must include change-password-at-next-login insert_call = next(c for c in log.chain if c[0] == "insert") body = insert_call[2]["body"] assert body["primaryEmail"] == "tom.smith@pingcap.com" assert body["changePasswordAtNextLogin"] is True assert body["orgUnitPath"] == "/Employees" @pytest.mark.asyncio async def test_create_user_409_marks_already_exists(patch_service): patch_service({ "users.insert": _make_http_error(409, "duplicate"), }) result = await _driver().create_user( primary_email="zhang.wei@pingcap.com", given_name="Zhang", family_name="Wei", password="abc", ) assert result.ok is True assert result.already_exists is True assert result.status == 409 @pytest.mark.asyncio async def test_create_user_500_surfaces_failure(patch_service): patch_service({ "users.insert": _make_http_error(500, "oops"), }) result = await _driver().create_user( primary_email="x@pingcap.com", given_name="X", family_name="Y", password="pwd", ) assert result.ok is False assert "500" in (result.error_message or "") @pytest.mark.asyncio async def test_create_user_retries_without_recovery_on_400(patch_service): # Google 400s the whole insert on a bad recoveryPhone/recoveryEmail. # Account creation is mandatory; recovery fields are best-effort — so a # first-try 400 must trigger ONE retry with those fields stripped, not # fail onboarding (real case: a Brazil mobile missing its 9-prefix). log = patch_service({ "users.insert": iter([ _make_http_error(400, "Invalid Input: recoveryPhone"), {"primaryEmail": "victor.schmidt@pingcap.com", "id": "42"}, ]), }) result = await _driver().create_user( primary_email="victor.schmidt@pingcap.com", given_name="Victor", family_name="Schmidt", password="Init-Pwd!2345678", recovery_email="victor_schmidt@outlook.com", recovery_phone="+551194976627", home_email="victor_schmidt@outlook.com", home_phone="1194976627", ) assert result.ok is True assert result.data["primary_email"] == "victor.schmidt@pingcap.com" inserts = [c for c in log.chain if c[0] == "insert"] assert len(inserts) == 2, "expected one retry after the 400" first_body = inserts[0][2]["body"] retry_body = inserts[1][2]["body"] # First attempt carried the recovery fields; the retry dropped them... assert "recoveryPhone" in first_body and "recoveryEmail" in first_body assert "recoveryPhone" not in retry_body assert "recoveryEmail" not in retry_body # ...but kept the mandatory + free-form display fields. assert retry_body["primaryEmail"] == "victor.schmidt@pingcap.com" assert retry_body["phones"] == [{"value": "1194976627", "type": "home"}] @pytest.mark.asyncio async def test_create_user_400_without_recovery_fields_does_not_retry(patch_service): # A 400 with NO recovery fields present is a genuine failure (bad OU, # password policy, …) — surface it, don't pointlessly retry. log = patch_service({ "users.insert": _make_http_error(400, "Invalid Input: orgUnitPath"), }) result = await _driver().create_user( primary_email="x@pingcap.com", given_name="X", family_name="Y", password="pwd", org_unit_path="/Nonexistent", ) assert result.ok is False assert "400" in (result.error_message or "") inserts = [c for c in log.chain if c[0] == "insert"] assert len(inserts) == 1, "must not retry when there are no recovery fields to drop" # ─── list_all_users ────────────────────────────────────────────── @pytest.mark.asyncio async def test_list_all_users_paginates_until_no_token(patch_service): pages = iter([ {"users": [{"primaryEmail": "a@x"}], "nextPageToken": "tok2"}, {"users": [{"primaryEmail": "b@x"}, {"primaryEmail": "c@x"}]}, ]) patch_service({"users.list": pages}) out = await _driver().list_all_users() assert [u["primaryEmail"] for u in out] == ["a@x", "b@x", "c@x"] @pytest.mark.asyncio async def test_list_all_users_first_page_403_raises_with_hint(patch_service): patch_service({"users.list": _make_http_error(403, "forbidden")}) with pytest.raises(GoogleApiError) as excinfo: await _driver().list_all_users() assert excinfo.value.status == 403 assert "admin.directory.user.readonly" in str(excinfo.value) # ─── list_group_members ───────────────────────────────────────── @pytest.mark.asyncio async def test_list_group_members_paginates(patch_service): pages = iter([ {"members": [{"email": "a@x"}], "nextPageToken": "n"}, {"members": [{"email": "b@x"}]}, ]) patch_service({"members.list": pages}) out = await _driver().list_group_members("team@pingcap.com") assert {m["email"] for m in out} == {"a@x", "b@x"} @pytest.mark.asyncio async def test_list_group_members_404_raises_with_hint(patch_service): patch_service({"members.list": _make_http_error(404, "not found")}) with pytest.raises(GoogleApiError) as excinfo: await _driver().list_group_members("ghost@pingcap.com") assert excinfo.value.status == 404 assert "Group not found" in str(excinfo.value) or "Group" in str(excinfo.value) # ─── _set_suspended idempotency on revoke direction ───────────── @pytest.mark.asyncio async def test_set_suspended_true_404_treated_idempotent(patch_service): """If the Google account was hard-deleted out-of-band between the Helix data-transfer poll and the suspend call, the desired end-state ('account dark') is already met. Suspend on a deleted user must close the offboarding blocker as already_absent — mirrors `_group_remove`'s 404 contract.""" patch_service({"users.update": _make_http_error(404, "not found")}) result = await _driver()._set_suspended( ProvisioningContext(user_email="ghost@pingcap.com"), suspended=True, ) assert result.ok is True assert result.raw_response == {"already_absent": True} @pytest.mark.asyncio async def test_set_suspended_false_404_does_not_swallow(patch_service): """On the unsuspend / grant direction we MUST NOT treat 404 as success — that would silently flip a grant to ok=True when the target user does not exist, masking a real provisioning bug. Asymmetric on purpose: 404 is only idempotent on the 'desired state = absent' direction.""" patch_service({"users.update": _make_http_error(404, "not found")}) result = await _driver()._set_suspended( ProvisioningContext(user_email="ghost@pingcap.com"), suspended=False, ) assert result.ok is False assert "404" in (result.error_message or "") # ── Round-1 offboarding hardening: error tagging + 403 paths ───── @pytest.mark.asyncio async def test_set_suspended_403_tagged_scope_missing(patch_service): """HIGH Round-1 #4: 403 on users.update means the Service Account lost ``admin.directory.user`` scope (org policy change, domain- wide delegation revoked). Tag so admin gets the specific 'fix the Service Account delegation' remediation hint instead of a bare 'HTTP 403: forbidden'.""" patch_service({"users.update": _make_http_error(403, "forbidden")}) result = await _driver()._set_suspended( ProvisioningContext(user_email="x@pingcap.com"), suspended=True, ) assert result.ok is False assert result.error_message.startswith(f"[{GOOGLE_TAG_SCOPE_MISSING}]") assert result.raw_response["error_tag"] == GOOGLE_TAG_SCOPE_MISSING assert result.raw_response["scope"] == "admin.directory.user" @pytest.mark.asyncio async def test_group_add_403_tagged_scope_missing(patch_service): """MEDIUM Round-1 #6: 403 on members.insert means missing admin.directory.group.member scope.""" patch_service({"members.insert": _make_http_error(403, "forbidden")}) result = await _driver()._group_add( ProvisioningContext( user_email="x@pingcap.com", extra={"group_email": "eng@pingcap.com"}, ), ) assert result.ok is False assert result.error_message.startswith(f"[{GOOGLE_TAG_SCOPE_MISSING}]") assert result.raw_response["scope"] == "admin.directory.group.member" @pytest.mark.asyncio async def test_group_remove_403_tagged_scope_missing(patch_service): """MEDIUM Round-1 #6 (symmetric): same tagging on the revoke side so admin gets the same hint regardless of which lifecycle phase tripped the scope issue.""" patch_service({"members.delete": _make_http_error(403, "forbidden")}) result = await _driver()._group_remove( ProvisioningContext( user_email="x@pingcap.com", extra={"group_email": "eng@pingcap.com"}, ), ) assert result.ok is False assert result.error_message.startswith(f"[{GOOGLE_TAG_SCOPE_MISSING}]") @pytest.mark.asyncio async def test_initiate_transfer_recipient_404_tagged_successor_not_in_directory( patch_service, ): """CRITICAL Round-1 #2: when the successor doesn't exist in Google's directory, ``_initiate_data_transfer`` MUST emit a distinct tag so admin knows to provision the successor's Google account first instead of looping on a transfer that will never succeed.""" # First .users.get() call fetches the RECIPIENT (new_email). Patch # that to 404 — driver should short-circuit before any transfer # body construction. patch_service({"users.get": _make_http_error(404, "no such user")}) result = await _driver()._initiate_data_transfer( old_email="leaver@pingcap.com", new_email="ghost-successor@pingcap.com", ) assert result.ok is False assert result.error_message.startswith( f"[{GOOGLE_TAG_SUCCESSOR_NOT_IN_DIRECTORY}]" ) assert ( result.raw_response["error_tag"] == GOOGLE_TAG_SUCCESSOR_NOT_IN_DIRECTORY ) assert result.raw_response["missing_recipient"] == "ghost-successor@pingcap.com" @pytest.mark.asyncio async def test_initiate_transfer_recipient_403_tagged_scope_missing( patch_service, ): """403 on recipient lookup is distinct from 404 — service account can't read the user (scope / domain delegation). Tag accordingly.""" patch_service({"users.get": _make_http_error(403, "forbidden")}) result = await _driver()._initiate_data_transfer( old_email="leaver@pingcap.com", new_email="locked-domain@external.com", ) assert result.ok is False assert result.error_message.startswith(f"[{GOOGLE_TAG_SCOPE_MISSING}]") assert result.raw_response["scope"] == "admin.directory.user" @pytest.mark.asyncio async def test_initiate_transfer_insert_403_tagged_transfer_permission( monkeypatch, ): """HIGH Round-1 #5: recipient lookup succeeds, but the actual transfers.insert() call returns 403 (recipient domain not delegated for Data Transfer / Service Account missing admin.datatransfer scope). Tag distinct from the recipient- lookup 403 so the hint pinpoints the data-transfer path. Requires patching BOTH ``_build_directory_service`` (for the pre-lookup of the recipient) AND ``_build_datatransfer_service`` (for the transfers.insert call). The standard ``patch_service`` fixture only patches the former, so this test wires both manually. """ dir_log = _CallLog() dt_log = _CallLog() dir_routes: dict[str, Any] = { "users.get": iter([ # 1st users.get = recipient (from _lookup_recipient). {"id": "new-uid", "primaryEmail": "successor@external.com"}, # 2nd users.get = old user (from inner _call). {"id": "old-uid", "primaryEmail": "leaver@pingcap.com"}, ]), } dt_routes: dict[str, Any] = { "transfers.insert": _make_http_error(403, "forbidden"), } def _fake_dir(sa_json: str, subject: str, scopes: list[str]): return _FluentNode(dir_log, (), dir_routes) def _fake_dt(sa_json: str, subject: str): return _FluentNode(dt_log, (), dt_routes) monkeypatch.setattr(gw, "_build_directory_service", _fake_dir) monkeypatch.setattr(gw, "_build_datatransfer_service", _fake_dt) result = await _driver()._initiate_data_transfer( old_email="leaver@pingcap.com", new_email="successor@external.com", ) assert result.ok is False assert result.error_message.startswith( f"[{GOOGLE_TAG_TRANSFER_PERMISSION_DENIED}]" ) assert ( result.raw_response["error_tag"] == GOOGLE_TAG_TRANSFER_PERMISSION_DENIED ) # ── 2026-06-02 retry + idempotency hardening ──────────────────── @pytest.mark.asyncio async def test_initiate_transfer_502_retries_then_succeeds(monkeypatch): """Production 502 incident: Google's Admin Data Transfer API returns a single 502, the driver retries via the exponential backoff schedule, the next attempt succeeds. End result is a happy-path transfer with no admin-facing failure DM. Backoff sleeps are monkeypatched to 0 so the test runs in <1ms. """ # Skip the real sleep so we don't burn 26 seconds of CI per test. monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) dir_log = _CallLog() dt_log = _CallLog() dir_routes: dict[str, Any] = { "users.get": iter([ {"id": "new-uid", "primaryEmail": "successor@x.com"}, {"id": "old-uid", "primaryEmail": "leaver@x.com"}, ]), } dt_routes: dict[str, Any] = { "transfers.insert": iter([ _make_http_error(502, "Bad Gateway"), { "id": "xfer-9999", "overallTransferStatusCode": "inProgress", }, ]), } def _fake_dir(sa_json: str, subject: str, scopes: list[str]): return _FluentNode(dir_log, (), dir_routes) def _fake_dt(sa_json: str, subject: str): return _FluentNode(dt_log, (), dt_routes) monkeypatch.setattr(gw, "_build_directory_service", _fake_dir) monkeypatch.setattr(gw, "_build_datatransfer_service", _fake_dt) result = await _driver()._initiate_data_transfer( old_email="leaver@x.com", new_email="successor@x.com", ) assert result.ok is True assert result.raw_response["transfer_id"] == "xfer-9999" # Two transfers.insert calls were made (one 502, one success) insert_calls = [c for c in dt_log.chain if c[0] == "insert"] assert len(insert_calls) == 2 @pytest.mark.asyncio async def test_initiate_transfer_502_exhausts_retries_then_fails( monkeypatch, ): """If every retry returns 502, the schedule eventually surfaces a failure. The error_message should carry the final 502 detail.""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) dir_log = _CallLog() dt_log = _CallLog() dir_routes: dict[str, Any] = { "users.get": iter([ {"id": "new-uid", "primaryEmail": "s@x.com"}, {"id": "old-uid", "primaryEmail": "l@x.com"}, ]), } # 4 calls total = 1 initial + 3 retries. All 502. dt_routes: dict[str, Any] = { "transfers.insert": iter([ _make_http_error(502, "Bad Gateway"), _make_http_error(502, "Bad Gateway"), _make_http_error(502, "Bad Gateway"), _make_http_error(502, "Bad Gateway"), ]), } monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(dir_log, (), dir_routes), ) monkeypatch.setattr( gw, "_build_datatransfer_service", lambda *a, **kw: _FluentNode(dt_log, (), dt_routes), ) result = await _driver()._initiate_data_transfer( old_email="l@x.com", new_email="s@x.com", ) assert result.ok is False assert "502" in (result.error_message or "") # Confirm all 4 attempts ran insert_calls = [c for c in dt_log.chain if c[0] == "insert"] assert len(insert_calls) == 4 @pytest.mark.asyncio async def test_initiate_transfer_source_user_404_idempotent_success( patch_service, ): """Pre-existing prod data: HRIS-late-detected leaver whose Google account was already cleaned manually earlier. Recipient lookup succeeds, source lookup returns 404 → driver short-circuits to idempotent success WITHOUT calling transfers.insert. Caller (revoke_seat) recognises ``skipped="source_user_already_gone"`` in raw_response and skips both poll and suspend.""" pages = iter([ # 1st users.get = recipient — succeeds {"id": "new-uid", "primaryEmail": "manager@x.com"}, # 2nd users.get = source — 404 _make_http_error(404, "user not found"), ]) patch_service({"users.get": pages}) result = await _driver()._initiate_data_transfer( old_email="long-gone@x.com", new_email="manager@x.com", ) assert result.ok is True assert ( result.raw_response.get("skipped") == "source_user_already_gone" ) assert result.raw_response.get("source_email") == "long-gone@x.com" @pytest.mark.asyncio async def test_revoke_seat_short_circuits_when_source_already_gone( monkeypatch, ): """End-to-end: revoke_seat(user_suspend) given a leaver whose source account is already removed from Google should return ok without polling and without calling suspend. Mirrors the 2026-06-01 production case for seats 6b060c1d / f86f29b0 (the 404 userKey failures).""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) pages = iter([ # recipient lookup succeeds {"id": "new-uid", "primaryEmail": "manager@x.com"}, # source lookup 404 _make_http_error(404, "user not found"), ]) dir_log = _CallLog() monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(dir_log, (), {"users.get": pages}), ) # If anything tries to call data transfer, fail loudly. def _explode(*a, **kw): raise AssertionError( "transfer service should NOT be built when source is gone", ) monkeypatch.setattr(gw, "_build_datatransfer_service", _explode) ctx = ProvisioningContext( user_email="long-gone@x.com", successor_email="manager@x.com", extra={"operation": "user_suspend"}, ) result = await _driver().revoke_seat(ctx) assert result.ok is True assert ( result.raw_response.get("skipped") == "source_user_already_gone" ) @pytest.mark.asyncio async def test_initiate_transfer_409_adopts_in_flight_transfer(monkeypatch): """Retry idempotency: a first attempt kicked off a transfer, our poll budget expired, the seat task failed, and the retry re-enters ``_initiate_data_transfer``. Google returns 409 "already in progress" on the second insert. The driver must locate the in-flight transfer via transfers.list and hand its id back (ok=True, adopted_in_progress) so revoke_seat polls the SAME transfer to completion — instead of failing the retry forever. Prod: sunyishen@pingcap.com → cwen@pingcap.com, task 52d1eba8, stuck at 2 attempts with a bare 409.""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) dir_log = _CallLog() dt_log = _CallLog() dir_routes: dict[str, Any] = { "users.get": iter([ {"id": "new-uid", "primaryEmail": "cwen@pingcap.com"}, {"id": "old-uid", "primaryEmail": "sunyishen@pingcap.com"}, ]), } dt_routes: dict[str, Any] = { "transfers.insert": _make_http_error(409, "in progress"), "transfers.list": { "dataTransfers": [ { "id": "xfer-inflight", "oldOwnerUserId": "old-uid", "newOwnerUserId": "new-uid", "overallTransferStatusCode": "inProgress", }, ], }, } monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(dir_log, (), dir_routes), ) monkeypatch.setattr( gw, "_build_datatransfer_service", lambda *a, **kw: _FluentNode(dt_log, (), dt_routes), ) result = await _driver()._initiate_data_transfer( old_email="sunyishen@pingcap.com", new_email="cwen@pingcap.com", ) assert result.ok is True assert result.external_ref == "xfer-inflight" assert result.raw_response["adopted_in_progress"] is True assert result.raw_response["transfer_id"] == "xfer-inflight" @pytest.mark.asyncio async def test_initiate_transfer_409_no_adoptable_transfer_fails(monkeypatch): """409 but the only transfer for the source user is ``failed`` — the driver must NOT adopt it (that's a real failure the operator needs to see) and returns an actionable error pointing at the Admin Console.""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) dir_log = _CallLog() dt_log = _CallLog() dir_routes: dict[str, Any] = { "users.get": iter([ {"id": "new-uid", "primaryEmail": "s@x.com"}, {"id": "old-uid", "primaryEmail": "l@x.com"}, ]), } dt_routes: dict[str, Any] = { "transfers.insert": _make_http_error(409, "in progress"), "transfers.list": { "dataTransfers": [ { "id": "xfer-dead", "oldOwnerUserId": "old-uid", "newOwnerUserId": "new-uid", "overallTransferStatusCode": "failed", }, ], }, } monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(dir_log, (), dir_routes), ) monkeypatch.setattr( gw, "_build_datatransfer_service", lambda *a, **kw: _FluentNode(dt_log, (), dt_routes), ) result = await _driver()._initiate_data_transfer( old_email="l@x.com", new_email="s@x.com", ) assert result.ok is False assert "409" in (result.error_message or "") assert "Data Migration" in (result.error_message or "") @pytest.mark.asyncio async def test_set_suspended_already_suspended_pre_read_short_circuit( monkeypatch, ): """If the account is ALREADY suspended in Google (a previous successful revoke or manual admin action), ``_set_suspended`` must short-circuit on the pre-read without firing the destructive ``users.update`` call. The update would succeed too (PUT is idempotent) but the pre-read is cheaper and the raw_response signal ``already_suspended=True`` lets the audit log distinguish the case.""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) log = _CallLog() routes: dict[str, Any] = { # pre-read returns suspended=True "users.get": { "primaryEmail": "already-locked@x.com", "suspended": True, "id": "uid-1", }, } monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(log, (), routes), ) result = await _driver()._set_suspended( ProvisioningContext(user_email="already-locked@x.com"), suspended=True, ) assert result.ok is True assert result.raw_response.get("already_suspended") is True # Critical: users.update was NEVER called — only users.get. update_calls = [c for c in log.chain if c[0] == "update"] assert update_calls == [] @pytest.mark.asyncio async def test_set_suspended_503_retries_then_succeeds(monkeypatch): """Google rate-limit / transient 503 on the suspend update must retry via the same backoff schedule as data-transfer init. Without this, a transient 503 fails the revoke task with an admin-facing DM that's pure noise.""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) log = _CallLog() routes: dict[str, Any] = { # pre-read says NOT yet suspended (so update fires) "users.get": { "primaryEmail": "x@x.com", "suspended": False, "id": "uid-1", }, "users.update": iter([ _make_http_error(503, "Service Unavailable"), {"primaryEmail": "x@x.com", "suspended": True}, ]), } monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(log, (), routes), ) result = await _driver()._set_suspended( ProvisioningContext(user_email="x@x.com"), suspended=True, ) assert result.ok is True update_calls = [c for c in log.chain if c[0] == "update"] assert len(update_calls) == 2 def _noop_async_sleep(): """Helper: an async sleep that returns immediately. Used to make backoff-based tests sub-millisecond. Picked over ``monkeypatch.setattr(asyncio, 'sleep', ...)`` plus a lambda because the assignment target must be an async coroutine, not a bare lambda.""" async def _zero(_: float) -> None: return None return _zero @pytest.mark.asyncio async def test_initiate_transfer_409_does_not_adopt_wrong_recipient(monkeypatch): """Round-5 HIGH fix: on a 409, the only in-flight transfer is to a DIFFERENT recipient than the one we intend (ops re-triggered revoke of the same leaver picking successor B while the to-A transfer is still running). The driver must NOT adopt the to-A transfer (that would suspend the leaver, report transferred_to=B, and eventually hard-delete them with Drive/Calendar handed to A). It returns an actionable 409 error instead of silently adopting the wrong-owner transfer.""" monkeypatch.setattr(gw.asyncio, "sleep", _noop_async_sleep()) dir_log = _CallLog() dt_log = _CallLog() dir_routes: dict[str, Any] = { "users.get": iter([ {"id": "b-uid", "primaryEmail": "successor-b@x.com"}, {"id": "old-uid", "primaryEmail": "leaver@x.com"}, ]), } dt_routes: dict[str, Any] = { "transfers.insert": _make_http_error(409, "in progress"), "transfers.list": { "dataTransfers": [ { "id": "xfer-to-a", "oldOwnerUserId": "old-uid", # In-flight transfer to a DIFFERENT successor (A), not B. "newOwnerUserId": "a-uid", "overallTransferStatusCode": "inProgress", }, ], }, } monkeypatch.setattr( gw, "_build_directory_service", lambda *a, **kw: _FluentNode(dir_log, (), dir_routes), ) monkeypatch.setattr( gw, "_build_datatransfer_service", lambda *a, **kw: _FluentNode(dt_log, (), dt_routes), ) result = await _driver()._initiate_data_transfer( old_email="leaver@x.com", new_email="successor-b@x.com", ) assert result.ok is False assert "409" in (result.error_message or "")