"""API + RBAC + drift tests for dynamic email-group management. The Google Workspace driver is fully patched: we assert RBAC, the structured rule builder round-trip, the expected-vs-actual reconcile, the large-drop guard, and the drift-detection handler's dead-value detection — all without hitting Google. """ from __future__ import annotations from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient from app.domains.email_group.models import EmailGroup from app.integrations.drivers.google_workspace import WorkspaceOpResult def _fake_driver(*, members=None, users=None): d = AsyncMock() d.scan_user_attributes.return_value = users if users is not None else [ {"email": "a@pingcap.com", "suspended": False, "fields": {"Department_Information.L1_Dept": "HR"}}, {"email": "b@pingcap.com", "suspended": False, "fields": {"Department_Information.L1_Dept": "HR"}}, {"email": "c@pingcap.com", "suspended": False, "fields": {"Department_Information.L1_Dept": "Finance"}}, ] d.list_group_members.return_value = members if members is not None else [ {"email": "a@pingcap.com", "role": "MEMBER"}, {"email": "b@pingcap.com", "role": "MEMBER"}, ] d.patch_dynamic_group_query.return_value = WorkspaceOpResult(ok=True) d.patch_group_display.return_value = WorkspaceOpResult(ok=True) d.create_dynamic_group.return_value = WorkspaceOpResult(ok=True, data={}) d.add_alias.return_value = WorkspaceOpResult(ok=True) d.remove_alias.return_value = WorkspaceOpResult(ok=True) d.delete_group.return_value = WorkspaceOpResult(ok=True) return d async def _mk_group(db, *, email="g-hr@pingcap.com", query=None, rule=None): row = EmailGroup( email=email, name="HR", status="active", is_dynamic=True, membership_query=query or "user.custom_schemas.Department_Information.L1_Dept=='HR'", dynamic_rule=rule or { "field": "Department_Information.L1_Dept", "match_values": ["HR"], "include_emails": [], "negate": False, "parseable": True, "raw_query": "user.custom_schemas.Department_Information.L1_Dept=='HR'", }, ) db.add(row) await db.commit() return row @pytest.mark.asyncio async def test_employee_cannot_list(client: AsyncClient, db_session, employee_user): _user, token = employee_user await _mk_group(db_session) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=_fake_driver(), ): r = await client.get( "/api/v1/email-groups/dynamic", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 403 @pytest.mark.asyncio async def test_hr_lists_with_health(client: AsyncClient, db_session, hr_user): _user, token = hr_user # A group whose rule references a dead value ("Ghost Dept" → 0 users). await _mk_group( db_session, email="g-ghost@pingcap.com", rule={ "field": "Department_Information.L1_Dept", "match_values": ["Ghost Dept"], "include_emails": [], "negate": False, "parseable": True, "raw_query": "user.custom_schemas.Department_Information.L1_Dept=='Ghost Dept'", }, ) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=_fake_driver(), ): r = await client.get( "/api/v1/email-groups/dynamic", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200 body = r.json()["data"] assert body["attributes_available"] is True ghost = next(g for g in body["groups"] if g["email"] == "g-ghost@pingcap.com") assert ghost["health"]["status"] == "dead_value" assert "Ghost Dept" in ghost["health"]["dead_values"] @pytest.mark.asyncio async def test_reconcile_expected_vs_actual(client: AsyncClient, db_session, hr_user): _user, token = hr_user row = await _mk_group(db_session) # rule matches HR = a,b # Actual on Google is only a@ + a stale extra → expect add b, remove stale. driver = _fake_driver(members=[ {"email": "a@pingcap.com", "role": "MEMBER"}, {"email": "stale@pingcap.com", "role": "MEMBER"}, ]) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=driver, ): r = await client.get( f"/api/v1/email-groups/dynamic/{row.id}/reconcile", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200 delta = r.json()["data"]["delta"] assert delta["evaluable"] is True assert "b@pingcap.com" in delta["added"] assert "stale@pingcap.com" in delta["removed"] @pytest.mark.asyncio async def test_export_members_enriches_and_audits( client: AsyncClient, db_session, hr_user, ): """HR exports the group's live Google members to xlsx. Members that match a Helix person are name-enriched; a non-person member (nested group / external) still appears with blank person fields.""" import io import uuid from openpyxl import load_workbook from app.identity.person_models import IdentityPerson _user, token = hr_user row = await _mk_group(db_session) db_session.add( IdentityPerson( id=uuid.uuid4(), work_email="a@pingcap.com", display_name="Alice", employee_no="E001", job_title="HR Manager", employment_status="active", source_system="successfactors", hris_person_id="sf-a-001", ) ) await db_session.commit() driver = _fake_driver(members=[ {"email": "a@pingcap.com", "role": "OWNER", "type": "USER", "status": "ACTIVE"}, {"email": "nested-group@pingcap.com", "role": "MEMBER", "type": "GROUP", "status": "ACTIVE"}, ]) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=driver, ): r = await client.get( f"/api/v1/email-groups/dynamic/{row.id}/members/export.xlsx", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200 assert "spreadsheetml" in r.headers["content-type"] assert r.headers["x-export-returned"] == "2" wb = load_workbook(io.BytesIO(r.content)) ws = wb.active # Row 1 = zh labels, row 2 = machine keys, data from row 3. keys = [c.value for c in ws[2]] rows = { r_[keys.index("email")].value: r_ for r_ in ws.iter_rows(min_row=3) } alice = rows["a@pingcap.com"] assert alice[keys.index("display_name")].value == "Alice" assert alice[keys.index("employee_no")].value == "E001" assert alice[keys.index("role")].value == "OWNER" nested = rows["nested-group@pingcap.com"] assert nested[keys.index("display_name")].value in (None, "") assert nested[keys.index("member_type")].value == "GROUP" @pytest.mark.asyncio async def test_export_members_forbidden_for_employee( client: AsyncClient, db_session, employee_user, ): _user, token = employee_user row = await _mk_group(db_session) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=_fake_driver(), ): r = await client.get( f"/api/v1/email-groups/dynamic/{row.id}/members/export.xlsx", headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 403 @pytest.mark.asyncio async def test_update_rule_large_drop_guard(client: AsyncClient, db_session, hr_user): """The g-all incident: a rule edit that empties the group must be refused without explicit confirmation, then succeed with it.""" _user, token = hr_user row = await _mk_group(db_session) # Current members a,b; new rule matches Finance only (c) → drops a,b (100%). driver = _fake_driver(members=[ {"email": "a@pingcap.com", "role": "MEMBER"}, {"email": "b@pingcap.com", "role": "MEMBER"}, ]) new_rule = { "field": "Department_Information.L1_Dept", "match_values": ["Finance"], "include_emails": [], "negate": False, } with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=driver, ): # Without confirm → 409 LARGE_MEMBERSHIP_DROP. r = await client.put( f"/api/v1/email-groups/dynamic/{row.id}/rule", json={"rule": new_rule, "confirm_large_drop": False}, headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 409 assert "LARGE_MEMBERSHIP_DROP" in r.text assert driver.patch_dynamic_group_query.await_count == 0 # With confirm → applied. r2 = await client.put( f"/api/v1/email-groups/dynamic/{row.id}/rule", json={"rule": new_rule, "confirm_large_drop": True}, headers={"Authorization": f"Bearer {token}"}, ) assert r2.status_code == 200 driver.patch_dynamic_group_query.assert_awaited_once() @pytest.mark.asyncio async def test_preview_compiles_and_flags_drop(client: AsyncClient, db_session, hr_user): _user, token = hr_user row = await _mk_group(db_session) driver = _fake_driver(members=[ {"email": "a@pingcap.com", "role": "MEMBER"}, {"email": "b@pingcap.com", "role": "MEMBER"}, ]) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=driver, ): r = await client.post( f"/api/v1/email-groups/dynamic/{row.id}/preview", json={"rule": { "field": "Department_Information.L1_Dept", "match_values": ["Finance"], "include_emails": [], "negate": False, }}, headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200 data = r.json()["data"] assert data["compiled_query"] == ( "user.custom_schemas.Department_Information.L1_Dept=='Finance'" ) assert data["large_drop"] is True @pytest.mark.asyncio async def test_create_dynamic_group(client: AsyncClient, db_session, hr_user): _user, token = hr_user driver = _fake_driver(members=[]) with patch( "app.domains.email_group.dynamic_service.EmailGroupService._get_driver", return_value=driver, ): r = await client.post( "/api/v1/email-groups/dynamic", json={ "email": "g-new@pingcap.com", "name": "New Group", "description": "", "rule": { "field": "Department_Information.L1_Dept", "match_values": ["HR"], "include_emails": [], "negate": False, }, }, headers={"Authorization": f"Bearer {token}"}, ) assert r.status_code == 200, r.text driver.create_dynamic_group.assert_awaited_once() row = ( await db_session.execute( EmailGroup.__table__.select().where( EmailGroup.email == "g-new@pingcap.com" ) ) ).first() assert row is not None @pytest.mark.asyncio async def test_drift_handler_flags_dead_value(db_session): """The drift cron stamps dead values and only counts a group as drifted when a value matches zero active users.""" from app.scheduled_jobs.google_handlers import run_dynamic_group_drift_check await _mk_group( db_session, email="g-dead@pingcap.com", rule={ "field": "Department_Information.L1_Dept", "match_values": ["Nonexistent Dept"], "include_emails": [], "negate": False, "parseable": True, "raw_query": "user.custom_schemas.Department_Information.L1_Dept=='Nonexistent Dept'", }, ) driver = _fake_driver() with patch( "app.scheduled_jobs.google_handlers._load_google_driver", return_value=driver, ), patch( "app.scheduled_jobs.failure_notifier.list_admin_recipients", new=AsyncMock(return_value=[]), ): summary = await run_dynamic_group_drift_check(db_session, {}) assert summary["groups_with_dead_values"] >= 1 row = ( await db_session.execute( EmailGroup.__table__.select().where( EmailGroup.email == "g-dead@pingcap.com" ) ) ).first() assert "Nonexistent Dept" in (row.dynamic_dead_values or [])