"""Every surface that lists cloud scopes must agree on what counts. Origin (2026-09-07): a user filtered 云子账号/订阅 by Owner as a global_admin and got **7** rows; the same Owner filter on the employee 我的云订阅 page returned **12**. Both filter ``owner_person_id``. The difference was five ``legacy_billing_unit`` rows plus a ``closed`` one: the admin list has excluded cost-allocation rows since migration 466, and the employee-facing directory hand-rolled its own WHERE clause that excluded only ``standalone_iam``. Auditing that class turned up **six** copies of "scopes this person is responsible for", written in five modules, giving three different answers about the same person: ============================ ============================================ surface what it used to filter on ============================ ============================================ employee requestable ``!= standalone_iam`` directory offboarding blocker scanner ``deleted_at IS NULL`` only offboarding preview ``deleted_at IS NULL`` only mover / role-change ``deleted_at IS NULL`` only holdings manager team cloud LIST ``deleted_at IS NULL`` only manager team count BADGE ``deleted_at IS NULL`` only leaver-residue reconciler closed + decommissioned (most correct) ============================ ============================================ Measured cost before the fix: 28 ``legacy_billing_unit`` rows + 2 ``azure_resource_group`` rows shown to every employee with a 快速申请 button that ``_refuse_ungrantable_scope_type`` rejects after the tap, plus ``cloud_scope_owner`` blockers raised on rows that are not accounts (a ``closed`` resource group and a ``legacy_billing_unit``), which the T+0 force-transfer path then "resolved" by handing a successor the ownership of a cost-allocation row. Those blockers were not stuck — a first revision of this docstring called them unresolvable, which was inferred rather than checked; prod has zero unresolved blockers of any type. Copies of a predicate drift silently because nothing fails when they disagree. They are now one function each (``owner_governed_scope_criteria`` / ``employee_directory_scope_criteria``) and this file is the thing that fails. """ from __future__ import annotations import uuid from typing import Any import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( SCOPE_STATUS_ACTIVE, SCOPE_STATUS_CLOSED, SCOPE_TYPE_ALIYUN_ACCOUNT, SCOPE_TYPE_AZURE_RESOURCE_GROUP, SCOPE_TYPE_LEGACY_BILLING_UNIT, CmdbCloudScope, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration async def _person( db: AsyncSession, *, status: str = "active", manager_id: uuid.UUID | None = None, ) -> IdentityPerson: row = IdentityPerson( id=uuid.uuid4(), work_email=f"p-{uuid.uuid4().hex[:8]}@pingcap.com", display_name=f"P {uuid.uuid4().hex[:4]}", employment_status=status, source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", manager_person_id=str(manager_id) if manager_id else None, ) db.add(row) await db.commit() return row async def _integration(db: AsyncSession) -> Integration: row = Integration( id=uuid.uuid4(), kind="cloud", provider="aliyun", name=f"integ-{uuid.uuid4().hex[:6]}", slug=f"in-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=[], ) db.add(row) await db.commit() return row async def _scopes_for( db: AsyncSession, integ: Integration, owner: IdentityPerson, ) -> dict[str, CmdbCloudScope]: """One genuinely-owned account + the three shapes that are not. Each junk row is a real production shape, not a synthetic edge case: a decommissioned account, a cost-allocation row from a dead tenant, and an AOAI cost-binding resource group. """ specs = { "real": (SCOPE_TYPE_ALIYUN_ACCOUNT, SCOPE_STATUS_ACTIVE), "closed": (SCOPE_TYPE_ALIYUN_ACCOUNT, SCOPE_STATUS_CLOSED), "legacy": (SCOPE_TYPE_LEGACY_BILLING_UNIT, SCOPE_STATUS_ACTIVE), "rg": (SCOPE_TYPE_AZURE_RESOURCE_GROUP, SCOPE_STATUS_ACTIVE), } out: dict[str, CmdbCloudScope] = {} for key, (scope_type, status) in specs.items(): row = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type=scope_type, external_id=f"{key}-{uuid.uuid4().hex[:8]}", display_name=f"{key}-scope", status=status, environment="dev", owner_person_id=str(owner.id), ) db.add(row) out[key] = row await db.commit() return out def _assert_only_real( ids: set[str], scopes: dict[str, CmdbCloudScope], surface: str, *, junk: tuple[str, ...] = ("closed", "legacy", "rg"), ) -> None: assert str(scopes["real"].id) in ids, ( f"{surface} dropped a real owned account — this filter must not " f"narrow genuine visibility" ) for key in junk: assert str(scopes[key].id) not in ids, ( f"{surface} surfaced the {key!r} scope. Hand-rolled WHERE " f"clause? Use owner_governed_scope_criteria()." ) @pytest.mark.asyncio async def test_offboarding_preview_and_scan_agree_and_skip_dead_scopes( db_session: AsyncSession, ) -> None: """A blocker holds up offboarding, so it must name something resolvable. Prod had two that were not.""" from app.domains.offboarding.models import RES_CLOUD_SCOPE_OWNER from app.domains.offboarding.service import OffboardingService leaver = await _person(db_session) integ = await _integration(db_session) scopes = await _scopes_for(db_session, integ, leaver) preview = await OffboardingService(db_session).get_offboarding_preview( leaver.id, ) ids = { b["resource_id"] for b in preview["blockers"] if b["resource_type"] == RES_CLOUD_SCOPE_OWNER } _assert_only_real(ids, scopes, "offboarding preview") @pytest.mark.asyncio async def test_mover_holdings_skip_dead_scopes( db_session: AsyncSession, ) -> None: """A mover cannot hand over a decommissioned account.""" from app.hr_lifecycle.mover_holdings import _cloud_scope_owner person = await _person(db_session) integ = await _integration(db_session) scopes = await _scopes_for(db_session, integ, person) holdings = await _cloud_scope_owner(db_session, person.id) ids = {str(h.resource_id) for h in holdings} _assert_only_real(ids, scopes, "mover holdings") @pytest.mark.asyncio async def test_team_cloud_list_and_count_badge_agree( db_session: AsyncSession, ) -> None: """The count badge and the list under it are two queries. A badge reading 4 over a list of 1 is its own bug report.""" from app.domains.me_universe.team_service import ( build_team_cloud, build_team_footprint, ) manager = await _person(db_session) sub = await _person(db_session, manager_id=manager.id) integ = await _integration(db_session) scopes = await _scopes_for(db_session, integ, sub) cloud = await build_team_cloud(db_session, manager=manager) ids = {row["id"] for row in cloud["scopes"]} _assert_only_real(ids, scopes, "manager team cloud list") footprint = await build_team_footprint(db_session, manager=manager) rows = [r for r in footprint["rows"] if r["person_id"] == str(sub.id)] assert len(rows) == 1, "subordinate missing from the team footprint" badge = rows[0]["footprint"]["scopes_owned"] assert badge == len(ids), ( f"count badge ({badge}) disagrees with the list ({len(ids)}) — " f"the two queries have drifted apart again" ) @pytest.mark.asyncio async def test_leaver_residue_skips_dead_scopes( db_session: AsyncSession, ) -> None: """The reconciler asks an admin to reassign what it reports.""" from app.domains.reconciliation.service import leaver_residual leaver = await _person(db_session, status="left") integ = await _integration(db_session) scopes = await _scopes_for(db_session, integ, leaver) residue = await leaver_residual(db_session) rows: list[dict[str, Any]] = [ r for r in residue["rows"] if r.get("resource_type") == "cloud_scope_owner" ] ids = {r["resource_id"] for r in rows} _assert_only_real(ids, scopes, "leaver residue reconciler") @pytest.mark.asyncio async def test_owner_orphan_lens_skips_ungovernable_scopes( db_session: AsyncSession, ) -> None: """"Who owns this when nobody owns it" must only ask about rows an admin can actually assign an owner to. Measured before the fix: **all 8** cloud_scope rows this lens reported in prod were ``legacy_billing_unit`` — a lens whose entire output was rows migration 466 typed precisely so that no owner governance applies. (Counted 7 on a first pass that only looked at NULL owners; the lens also reports owners who exist but have left, so the true figure is 8.) An admin could never clear one, which is the same permanent-false-positive shape the GCP-billing-account exclusion next to it already guards against. """ from app.domains.reconciliation.service import ownership_integrity integ = await _integration(db_session) ownerless: dict[str, CmdbCloudScope] = {} for key, scope_type in ( ("real", SCOPE_TYPE_ALIYUN_ACCOUNT), ("legacy", SCOPE_TYPE_LEGACY_BILLING_UNIT), ("rg", SCOPE_TYPE_AZURE_RESOURCE_GROUP), ): row = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type=scope_type, external_id=f"orphan-{key}-{uuid.uuid4().hex[:8]}", display_name=f"orphan-{key}", status=SCOPE_STATUS_ACTIVE, environment="dev", owner_person_id=None, ) db_session.add(row) ownerless[key] = row await db_session.commit() result = await ownership_integrity(db_session) ids = {r["resource_id"] for r in result["issues"]["cloud_scope"]} assert str(ownerless["real"].id) in ids, ( "a genuinely ownerless cloud account must still be reported" ) for key in ("legacy", "rg"): assert str(ownerless[key].id) not in ids, ( f"orphan lens reported the {key!r} scope — nobody can assign " f"it an owner, so the row can never be cleared" ) @pytest.mark.asyncio async def test_employee_directory_and_admin_list_agree_for_one_owner( db_session: AsyncSession, client, admin_headers, employee_user, ) -> None: """The exact shape the user reported: same Owner, two pages, two different numbers. The contract is **containment, not equality**: the employee directory must be a SUBSET of what the admin console shows for that owner. Admin is legitimately wider — it is the ops catalogue and deliberately still lists ``azure_resource_group`` rows (real Azure objects bound to an API key) and ``standalone_iam`` (which the employee page routes to the IAM-users page instead), and it can reveal ``legacy_billing_unit`` on demand via ``include_legacy_billing``. What must never happen again is the reported direction: the employee page showing MORE than the admin console, i.e. handing an employee a row its own admins can't see. """ _, employee_token = employee_user owner = await _person(db_session) integ = await _integration(db_session) scopes = await _scopes_for(db_session, integ, owner) admin_resp = await client.get( f"/api/v1/cmdb/scopes?page_size=500&owner_person_id={owner.id}", headers=admin_headers, ) assert admin_resp.status_code == 200, admin_resp.text admin_ids = {r["id"] for r in admin_resp.json()["data"]} emp_resp = await client.get( "/api/v1/cmdb/my-subscriptions?include_requestable=true", headers={"Authorization": f"Bearer {employee_token}"}, ) assert emp_resp.status_code == 200, emp_resp.text emp_ids = { r["scope_id"] for r in emp_resp.json()["data"] if r["owner_person_id"] == str(owner.id) } # Admin: hides the cost-allocation row (migration 466) and the # closed one (main list is live-only). ``rg`` is deliberately NOT # asserted absent — see the docstring. _assert_only_real( admin_ids, scopes, "admin scope list", junk=("closed", "legacy"), ) # Employee: all three shapes must be gone. _assert_only_real(emp_ids, scopes, "employee requestable directory") assert emp_ids <= admin_ids, ( "employee 我的云订阅 shows scopes the admin 云子账号/订阅 list " f"does not — the exact 7-vs-12 defect. extra={emp_ids - admin_ids}" )