"""Tests for the asset_admin auto-derived role + location region gate. Migration 530 moved asset authority off offices: ``asset_admin`` role membership is now exactly ``SELECT DISTINCT person_id FROM asset_location_admins`` (filtered to active, non-deleted persons), unioned with the ``asset_global_admin`` co-derivation, and the region gate compares an asset's resolved *location* against the caller's administered set. This module carries forward every invariant the office-based suite asserted, and adds the ones the new model introduces: an office's 当地 Admin list must grant nothing, a deactivated location must stay in scope for the write gate, and a location's single accountable administrator must resolve the way the notification flows expect. """ from __future__ import annotations import uuid import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthRole, AuthUserRole from app.domains.asset.location_admin_sync import ( get_asset_location_id, get_location_ids_managed_by, get_person_location_id, list_location_admin_ids_bulk, list_location_admins, person_can_manage_asset, reconcile_asset_admin_role, replace_location_admins, resolve_location_primary_admins_bulk, sync_asset_admin_role_for_person, ) from app.domains.asset.models import ( AssetItem, AssetLocation, AssetLocationAdmin, AssetOfficeAddress, ) from app.identity.person_models import IdentityPerson async def _make_person(db: AsyncSession, email: str) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=email, display_name=email.split("@")[0], employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.commit() return p async def _make_location( db: AsyncSession, name: str, *, is_active: bool = True, ) -> AssetLocation: loc = AssetLocation( id=uuid.uuid4(), code=f"T{uuid.uuid4().hex[:8]}", name=name, is_active=is_active, ) db.add(loc) await db.commit() return loc async def _make_asset(db: AsyncSession, **kwargs) -> AssetItem: asset = AssetItem( asset_tag=f"R-{uuid.uuid4().hex[:6]}", category="laptop", status=kwargs.pop("status", "claimed"), **kwargs, ) db.add(asset) await db.commit() return asset async def _asset_admin_role_id(db: AsyncSession) -> uuid.UUID: """Ensure the auth_roles row exists; return its id.""" row = await db.execute( select(AuthRole).where(AuthRole.name == "asset_admin") ) role = row.scalar_one_or_none() if role is None: role = AuthRole(id=uuid.uuid4(), name="asset_admin", is_system=True) db.add(role) await db.commit() return role.id async def _grant_row(db: AsyncSession, person_id, role_id): rows = await db.execute( select(AuthUserRole).where( AuthUserRole.user_id == person_id, AuthUserRole.role_id == role_id, ) ) return rows.scalar_one_or_none() # ── Role derivation ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_sync_role_grants_when_person_joins_location( db_session: AsyncSession, ) -> None: role_id = await _asset_admin_role_id(db_session) person = await _make_person(db_session, "alice@pingcap.com") location = await _make_location(db_session, "LocBeijing") assert await _grant_row(db_session, person.id, role_id) is None await replace_location_admins(db_session, location.id, [person.id]) await db_session.commit() assert await _grant_row(db_session, person.id, role_id) is not None @pytest.mark.asyncio async def test_sync_role_revokes_when_removed_from_last_location( db_session: AsyncSession, ) -> None: role_id = await _asset_admin_role_id(db_session) person = await _make_person(db_session, "bob@pingcap.com") location = await _make_location(db_session, "LocShanghai") await replace_location_admins(db_session, location.id, [person.id]) await db_session.commit() assert await _grant_row(db_session, person.id, role_id) is not None await replace_location_admins(db_session, location.id, []) await db_session.commit() assert await _grant_row(db_session, person.id, role_id) is None @pytest.mark.asyncio async def test_role_survives_while_another_location_remains( db_session: AsyncSession, ) -> None: """Removal from one location must not revoke a still-earned role.""" role_id = await _asset_admin_role_id(db_session) person = await _make_person(db_session, "two@pingcap.com") first = await _make_location(db_session, "LocFirst") second = await _make_location(db_session, "LocSecond") await replace_location_admins(db_session, first.id, [person.id]) await replace_location_admins(db_session, second.id, [person.id]) await db_session.commit() await replace_location_admins(db_session, first.id, []) await db_session.commit() assert await _grant_row(db_session, person.id, role_id) is not None assert await get_location_ids_managed_by(db_session, person.id) == {second.id} @pytest.mark.asyncio async def test_reconcile_recovers_arbitrary_drift( db_session: AsyncSession, ) -> None: role_id = await _asset_admin_role_id(db_session) person_should = await _make_person(db_session, "carol@pingcap.com") person_phantom = await _make_person(db_session, "drift@pingcap.com") location = await _make_location(db_session, "LocSingapore") # Drift state: A is in the join but missing from grants; # B has a grant but no join row. db_session.add( AssetLocationAdmin(location_id=location.id, person_id=person_should.id), ) db_session.add( AuthUserRole(user_id=person_phantom.id, role_id=role_id), ) await db_session.commit() summary = await reconcile_asset_admin_role(db_session) await db_session.commit() assert summary["granted"] >= 1 assert summary["revoked"] >= 1 assert await _grant_row(db_session, person_should.id, role_id) is not None assert await _grant_row(db_session, person_phantom.id, role_id) is None @pytest.mark.asyncio async def test_office_local_admin_grants_nothing( db_session: AsyncSession, ) -> None: """The whole point of the split: 当地 Admin confers no authority. Before migration 535 this exact row was the grant. If a future change re-couples them, this fails. """ from app.domains.asset.office_local_admins import ( replace_office_local_admins, ) role_id = await _asset_admin_role_id(db_session) person = await _make_person(db_session, "localadmin@pingcap.com") office = AssetOfficeAddress(id=uuid.uuid4(), city="LocalOnlyCity") db_session.add(office) await db_session.commit() await replace_office_local_admins(db_session, office.id, [person.id]) await db_session.commit() assert await _grant_row(db_session, person.id, role_id) is None assert await get_location_ids_managed_by(db_session, person.id) == set() # The reconciler must agree — a 当地 Admin is not a grant source. await reconcile_asset_admin_role(db_session) await db_session.commit() assert await _grant_row(db_session, person.id, role_id) is None # ── Admin list management ─────────────────────────────────────────── @pytest.mark.asyncio async def test_replace_location_admins_sets_primary( db_session: AsyncSession, ) -> None: await _asset_admin_role_id(db_session) p1 = await _make_person(db_session, "p1@pingcap.com") p2 = await _make_person(db_session, "p2@pingcap.com") location = await _make_location(db_session, "LocTokyo") # Default: the first submitted person becomes primary. await replace_location_admins(db_session, location.id, [p1.id, p2.id]) await db_session.commit() admins = await list_location_admins(db_session, location.id) primaries = [a for a in admins if a["is_primary"]] assert len(primaries) == 1 assert primaries[0]["person_id"] == p1.id # An explicit primary switch with an unchanged membership list must # still land — the flag update is not gated on membership churn. await replace_location_admins( db_session, location.id, [p1.id, p2.id], primary_person_id=p2.id, ) await db_session.commit() admins = await list_location_admins(db_session, location.id) primaries = [a for a in admins if a["is_primary"]] assert len(primaries) == 1 assert primaries[0]["person_id"] == p2.id @pytest.mark.asyncio async def test_replace_location_admins_rejects_primary_outside_list( db_session: AsyncSession, ) -> None: from app.core.exceptions import ValidationError await _asset_admin_role_id(db_session) p1 = await _make_person(db_session, "inlist@pingcap.com") outsider = await _make_person(db_session, "outsider@pingcap.com") location = await _make_location(db_session, "LocPrimaryGuard") with pytest.raises(ValidationError): await replace_location_admins( db_session, location.id, [p1.id], primary_person_id=outsider.id, ) @pytest.mark.asyncio async def test_replace_location_admins_rejects_terminated_person( db_session: AsyncSession, ) -> None: """A departed person must not be writable into the admin list. Uses ``'leaver_pending'`` rather than ``'left'``: flipping to ``'left'`` triggers the LeaverBlocked listener, which writes to ``leaver_block_audit``. ``'leaver_pending'`` is equally non-active and exercises the same rejection branch. """ from app.core.exceptions import ValidationError await _asset_admin_role_id(db_session) leaver = await _make_person(db_session, "leaver@pingcap.com") leaver.employment_status = "leaver_pending" await db_session.commit() location = await _make_location(db_session, "LocSydney") with pytest.raises(ValidationError): await replace_location_admins(db_session, location.id, [leaver.id]) @pytest.mark.asyncio async def test_list_location_admins_filters_departed( db_session: AsyncSession, ) -> None: """A row that leaks in (added while active, person later leaves) must not surface as a ghost admin, and the role must drop.""" role_id = await _asset_admin_role_id(db_session) active_admin = await _make_person(db_session, "active@pingcap.com") later_leaver = await _make_person(db_session, "willleave@pingcap.com") location = await _make_location(db_session, "LocBangalore") await replace_location_admins( db_session, location.id, [active_admin.id, later_leaver.id], ) await db_session.commit() later_leaver.employment_status = "leaver_pending" await db_session.commit() admins = await list_location_admins(db_session, location.id) admin_ids = {a["person_id"] for a in admins} assert active_admin.id in admin_ids assert later_leaver.id not in admin_ids bulk = await list_location_admin_ids_bulk(db_session, [location.id]) bulk_ids = {a["person_id"] for a in bulk[location.id]} assert active_admin.id in bulk_ids assert later_leaver.id not in bulk_ids await sync_asset_admin_role_for_person(db_session, later_leaver.id) await db_session.commit() assert await _grant_row(db_session, later_leaver.id, role_id) is None @pytest.mark.asyncio async def test_primary_admin_resolution_rules( db_session: AsyncSession, ) -> None: """The single-recipient rule the inventory fan-out depends on. Primary when flagged; the sole admin when there is exactly one and no flag; nothing when several admins share a location with no primary — an arbitrary pick would tell the wrong person they are accountable. """ await _asset_admin_role_id(db_session) p1 = await _make_person(db_session, "prim1@pingcap.com") p2 = await _make_person(db_session, "prim2@pingcap.com") flagged = await _make_location(db_session, "LocFlagged") await replace_location_admins( db_session, flagged.id, [p1.id, p2.id], primary_person_id=p2.id, ) solo = await _make_location(db_session, "LocSolo") db_session.add( AssetLocationAdmin(location_id=solo.id, person_id=p1.id, is_primary=False), ) ambiguous = await _make_location(db_session, "LocAmbiguous") db_session.add( AssetLocationAdmin( location_id=ambiguous.id, person_id=p1.id, is_primary=False, ), ) db_session.add( AssetLocationAdmin( location_id=ambiguous.id, person_id=p2.id, is_primary=False, ), ) await db_session.commit() resolved = await resolve_location_primary_admins_bulk( db_session, [flagged.id, solo.id, ambiguous.id], ) assert resolved[flagged.id] == p2.id assert resolved[solo.id] == p1.id assert resolved[ambiguous.id] is None # ── Person → location resolution ──────────────────────────────────── @pytest.mark.asyncio async def test_person_location_resolution_chain( db_session: AsyncSession, ) -> None: """``work_location_name`` then ``city``; ``work_location_code`` never. The code is an ISO country code (CN/US/SG) and cannot legitimately name a location, so admitting it as a candidate would only ever produce a false positive against a location that happens to share the string. """ beijing = await _make_location(db_session, "ResolveBeijing") shanghai = await _make_location(db_session, "ResolveShanghai") trap = await _make_location(db_session, "CN") p1 = await _make_person(db_session, "r1@pingcap.com") p1.work_location_name = "ResolveBeijing" await db_session.commit() assert await get_person_location_id(db_session, p1.id) == beijing.id p2 = await _make_person(db_session, "r2@pingcap.com") p2.work_location_name = None p2.city = "ResolveShanghai" await db_session.commit() assert await get_person_location_id(db_session, p2.id) == shanghai.id p3 = await _make_person(db_session, "r3@pingcap.com") p3.work_location_code = "CN" p3.work_location_name = None p3.city = None await db_session.commit() assert await get_person_location_id(db_session, p3.id) is None _ = trap # exists only to make case 3 a real trap @pytest.mark.asyncio async def test_person_location_resolves_across_shi_suffix( db_session: AsyncSession, ) -> None: """Regression (2026-07-15): HRIS writes mainland-China cities with the ``市`` suffix (``杭州市``) while the master stores the bare name (``杭州``). Exact equality failed, so every China employee resolved to nothing and their own regional administrator was told the target was outside their region (real case: 王静文 P10012 @ 杭州市, administrator Thalie Zhuang covering 杭州). Covered twice over now: ``杭州市`` has an explicit normalization-map entry, and any other ``…市`` value falls back to the generic suffix strip. The second case below exercises the generic path with a name the map has never heard of. """ hangzhou = await _make_location(db_session, "杭州") p = await _make_person(db_session, "jingwen@pingcap.com") p.work_location_name = "杭州市" p.city = "杭州市" await db_session.commit() assert await get_person_location_id(db_session, p.id) == hangzhou.id # Generic suffix strip: a China city the normalization map does not # list, added to the master by an operator. suzhou = await _make_location(db_session, "苏州") p2 = await _make_person(db_session, "suzhou@pingcap.com") p2.work_location_name = "苏州市" p2.city = None await db_session.commit() assert await get_person_location_id(db_session, p2.id) == suzhou.id # ── Region gate ───────────────────────────────────────────────────── @pytest.mark.asyncio async def test_region_permission_gate(db_session: AsyncSession) -> None: await _asset_admin_role_id(db_session) asset_admin = await _make_person(db_session, "regional@pingcap.com") other_admin = await _make_person(db_session, "other@pingcap.com") beijing = await _make_location(db_session, "GateBeijing") shanghai = await _make_location(db_session, "GateShanghai") await replace_location_admins(db_session, beijing.id, [asset_admin.id]) await replace_location_admins(db_session, shanghai.id, [other_admin.id]) await db_session.commit() in_region = await _make_asset(db_session, location="GateBeijing") out_of_region = await _make_asset(db_session, location="GateShanghai") managed = await get_location_ids_managed_by(db_session, asset_admin.id) assert managed == {beijing.id} assert await get_asset_location_id(db_session, in_region) == beijing.id assert await person_can_manage_asset( db_session, asset_admin.id, in_region, role_names=["asset_admin"], ) assert not await person_can_manage_asset( db_session, asset_admin.id, out_of_region, role_names=["asset_admin"], ) # global_admin bypasses regardless. assert await person_can_manage_asset( db_session, asset_admin.id, out_of_region, role_names=["global_admin"], ) @pytest.mark.asyncio async def test_unresolvable_asset_location_is_global_only( db_session: AsyncSession, ) -> None: """An asset whose location names nothing managed belongs to no region — a regional admin must not reach it, the global tier must.""" await _asset_admin_role_id(db_session) admin = await _make_person(db_session, "orphangate@pingcap.com") location = await _make_location(db_session, "GateOrphanHome") await replace_location_admins(db_session, location.id, [admin.id]) await db_session.commit() orphan = await _make_asset(db_session, location="NoSuchPlaceAtAll") assert await get_asset_location_id(db_session, orphan) is None assert not await person_can_manage_asset( db_session, admin.id, orphan, role_names=["asset_admin"], ) assert await person_can_manage_asset( db_session, admin.id, orphan, role_names=["asset_global_admin"], ) @pytest.mark.asyncio async def test_asset_falls_back_to_assignee_location( db_session: AsyncSession, ) -> None: """An asset with no usable location of its own resolves through its holder, so a claim that never stamped the field stays in region.""" await _asset_admin_role_id(db_session) admin = await _make_person(db_session, "fallback.admin@pingcap.com") location = await _make_location(db_session, "FallbackCity") await replace_location_admins(db_session, location.id, [admin.id]) holder = await _make_person(db_session, "fallback.holder@pingcap.com") holder.work_location_name = "FallbackCity" await db_session.commit() asset = await _make_asset(db_session, location=None, assigned_to=holder.id) assert await get_asset_location_id(db_session, asset) == location.id assert await person_can_manage_asset( db_session, admin.id, asset, role_names=["asset_admin"], ) @pytest.mark.asyncio async def test_deactivated_location_stays_in_region( db_session: AsyncSession, ) -> None: """Retiring a location must not strand the assets left there. The write gate loads the resolver with inactive rows included, so an administrator keeps authority over what is still sitting at a location an operator has closed. Without this the assets would resolve to nothing and only the global tier could clean them up. """ await _asset_admin_role_id(db_session) admin = await _make_person(db_session, "retired.admin@pingcap.com") location = await _make_location(db_session, "RetiredCity") await replace_location_admins(db_session, location.id, [admin.id]) await db_session.commit() asset = await _make_asset(db_session, location="RetiredCity") location.is_active = False await db_session.commit() assert await get_asset_location_id(db_session, asset) == location.id assert await person_can_manage_asset( db_session, admin.id, asset, role_names=["asset_admin"], ) @pytest.mark.asyncio async def test_country_location_resolves_directly( db_session: AsyncSession, ) -> None: """A country-level location is just a location. The office model needed a special "only if this country has exactly one office" rule to make ``location='美国'`` safe. Countries are managed locations in their own right now, so the value resolves directly and the special case is gone. """ await _asset_admin_role_id(db_session) admin = await _make_person(db_session, "us.admin@pingcap.com") usa = await _make_location(db_session, "CountryUSA") await replace_location_admins(db_session, usa.id, [admin.id]) await db_session.commit() asset = await _make_asset(db_session, location="CountryUSA") assert await get_asset_location_id(db_session, asset) == usa.id assert await person_can_manage_asset( db_session, admin.id, asset, role_names=["asset_admin"], ) @pytest.mark.asyncio async def test_pre_hire_target_bypass_actually_fires( db_session: AsyncSession, ) -> None: """The new-hire bypass must let a regional administrator issue an asset to a target whose HRIS sync has not landed (``employment_status='pre_hire'``, no work location yet). Written against the real gate, not the enum value: an earlier revision of this rule spelled the status ``'pending'``, which is not a valid value and would be rejected by the ``@validates`` hook, so the bypass silently never fired. """ from app.core.exceptions import ForbiddenError from app.domains.asset.region_gates import assert_target_person_in_region await _asset_admin_role_id(db_session) admin = await _make_person(db_session, "asset.admin@pingcap.com") location = await _make_location(db_session, "PreHireCity") await replace_location_admins(db_session, location.id, [admin.id]) await db_session.commit() class _Role: def __init__(self, name): self.name = name class _UR: def __init__(self, role_name): self.role = _Role(role_name) class _U: def __init__(self, pid): self.id = pid self.user_roles = [_UR("asset_admin")] caller = _U(admin.id) new_hire = await _make_person(db_session, "newhire@pingcap.com") new_hire.employment_status = "pre_hire" new_hire.work_location_name = None new_hire.city = None await db_session.commit() await assert_target_person_in_region(db_session, caller, new_hire.id) # Negative: an active target with no work location is NOT covered. foreign = await _make_person(db_session, "foreign@pingcap.com") foreign.work_location_name = None foreign.city = None await db_session.commit() with pytest.raises(ForbiddenError): await assert_target_person_in_region(db_session, caller, foreign.id) @pytest.mark.asyncio async def test_explicit_dispatch_location_can_establish_region( db_session: AsyncSession, ) -> None: """When HR has no location for the target, the location chosen on the dispatch form may establish it — but only within the caller's own region, never as a way out of it.""" from app.core.exceptions import ForbiddenError from app.domains.asset.region_gates import assert_target_person_in_region await _asset_admin_role_id(db_session) admin = await _make_person(db_session, "dispatch.admin@pingcap.com") mine = await _make_location(db_session, "DispatchMine") await _make_location(db_session, "DispatchTheirs") await replace_location_admins(db_session, mine.id, [admin.id]) await db_session.commit() class _Role: def __init__(self, name): self.name = name class _UR: def __init__(self, role_name): self.role = _Role(role_name) class _U: def __init__(self, pid): self.id = pid self.user_roles = [_UR("asset_admin")] caller = _U(admin.id) target = await _make_person(db_session, "nolocation@pingcap.com") target.work_location_name = None target.city = None await db_session.commit() await assert_target_person_in_region( db_session, caller, target.id, target_location="DispatchMine", ) with pytest.raises(ForbiddenError): await assert_target_person_in_region( db_session, caller, target.id, target_location="DispatchTheirs", )