"""Managed-location resolver tests.

Covers:
  * Direct ``ManagedLocationResolver.resolve()`` matrix — exact, map,
    miss, whitespace, non-string, NFKC.
  * Master-state coupling — disabled master rows are NOT returned even
    when the normalization map points at them.
  * Boundary integration — claim / return / borrow / change_assignee
    and create / intake / update all route raw values through the
    resolver before writing ``asset_items.location``.
  * Audit trail — operation ``detail`` carries ``location_raw`` when
    the resolver rewrites the input.
  * Regression — every dirty value from the historical 2026-05-30
    backfill (Santa Clara → 美国; 北京市 → 北京; Würzburg → 德国; …)
    still resolves the same way today.

The test pattern mirrors ``test_brand_model_service.py``: direct
service-layer drive against ``db_session``, no HTTP client roundtrip —
the resolver is plain enough that boundary tests don't need full
auth/RBAC stubbing.
"""
from __future__ import annotations

import uuid
from datetime import date, timedelta

import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.asset.location_resolver import (
    ManagedLocationResolver,
    get_normalization_map_snapshot,
    resolve_managed_location,
)
from app.domains.asset.models import AssetItem, AssetLocation
from app.domains.asset.schemas import (
    AssetBorrowRequest,
    AssetChangeAssigneeRequest,
    AssetClaimRequest,
    AssetCreateRequest,
    AssetReturnRequest,
    AssetUpdateRequest,
)
from app.domains.asset.service import AssetService
from app.identity.person_models import IdentityPerson

# ── Fixtures ───────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def location_master(db_session: AsyncSession):
    """Seed the master rows the resolver expects.

    Mirrors the prod ``asset_locations`` rows shipped by migrations
    1-226 (subset chosen so the resolver matrix has both China-city
    canonicals and country canonicals to land on).
    """
    rows = [
        ("P-CN03", "北京", 1),
        ("P-CN06", "杭州", 3),
        ("P-CN07", "上海", 5),
        ("P-US", "美国", 7),
        ("P-JP", "日本", 8),
        ("P-MY", "马来西亚", 10),
        ("NL", "荷兰", 16),
        ("DE", "德国", 21),
        ("AU", "澳大利亚", 25),
        ("CA", "加拿大", 26),
    ]
    for code, name, order in rows:
        db_session.add(
            AssetLocation(
                code=code,
                name=name,
                sort_order=order,
                is_active=True,
            )
        )
    await db_session.flush()
    return [name for _, name, _ in rows]


@pytest_asyncio.fixture
async def operator_person(db_session: AsyncSession) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"op-{uuid.uuid4().hex[:8]}@test.com",
        display_name="Operator",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-op-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(p)
    await db_session.flush()
    return p


@pytest_asyncio.fixture
async def holder_person(db_session: AsyncSession) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"holder-{uuid.uuid4().hex[:8]}@test.com",
        display_name="Holder",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-hd-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(p)
    await db_session.flush()
    return p


@pytest_asyncio.fixture
async def second_holder(db_session: AsyncSession) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"sec-{uuid.uuid4().hex[:8]}@test.com",
        display_name="Second Holder",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-h2-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(p)
    await db_session.flush()
    return p


@pytest_asyncio.fixture
async def idle_asset(db_session: AsyncSession) -> AssetItem:
    row = AssetItem(
        asset_tag=f"TST-{uuid.uuid4().hex[:8]}",
        asset_name="resolver-boundary",
        category="laptop",
        status="idle",
    )
    db_session.add(row)
    await db_session.flush()
    return row


# ── Direct resolver matrix ─────────────────────────────────────────


@pytest.mark.asyncio
async def test_exact_match_returns_canonical(
    db_session: AsyncSession, location_master,
):
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve("北京") == "北京"
    assert resolver.resolve("美国") == "美国"


@pytest.mark.asyncio
async def test_china_city_suffix_normalized(
    db_session: AsyncSession, location_master,
):
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve("北京市") == "北京"
    assert resolver.resolve("杭州市") == "杭州"
    assert resolver.resolve("上海市") == "上海"


@pytest.mark.asyncio
async def test_us_cities_collapse_to_country(
    db_session: AsyncSession, location_master,
):
    resolver = await ManagedLocationResolver.load(db_session)
    for city in [
        "Santa Clara", "Sunnyvale", "California", "San Jose",
        "Seattle", "King County", "Chapel Hill",
    ]:
        assert resolver.resolve(city) == "美国", f"failed: {city}"


@pytest.mark.asyncio
async def test_other_country_city_mappings(
    db_session: AsyncSession, location_master,
):
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve("Tokyo") == "日本"
    assert resolver.resolve("Kuala Lumpur") == "马来西亚"
    assert resolver.resolve("Würzburg") == "德国"
    assert resolver.resolve("Sydney") == "澳大利亚"
    assert resolver.resolve("Toronto") == "加拿大"
    assert resolver.resolve("Rotterdam") == "荷兰"


@pytest.mark.asyncio
async def test_empty_and_none_yield_none(
    db_session: AsyncSession, location_master,
):
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve(None) is None
    assert resolver.resolve("") is None
    assert resolver.resolve("   ") is None
    assert resolver.resolve("\t\n") is None


@pytest.mark.asyncio
async def test_non_string_input_yields_none(
    db_session: AsyncSession, location_master,
):
    """Defensive: callers occasionally pass an int / UUID / None."""
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve(42) is None  # type: ignore[arg-type]
    assert resolver.resolve(uuid.uuid4()) is None  # type: ignore[arg-type]


@pytest.mark.asyncio
async def test_unknown_value_returns_none(
    db_session: AsyncSession, location_master,
):
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve("Mars Colony") is None
    assert resolver.resolve("天王星") is None


@pytest.mark.asyncio
async def test_nfkc_and_whitespace_normalization(
    db_session: AsyncSession, location_master,
):
    """Fullwidth / half-width / surrounding whitespace all collapse."""
    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve("  北京 ") == "北京"
    assert resolver.resolve("Ｓａｎｔａ Ｃｌａｒａ") == "美国"


@pytest.mark.asyncio
async def test_disabled_master_row_breaks_resolution(
    db_session: AsyncSession, location_master,
):
    """If an admin deactivates ``美国``, ``Santa Clara`` stops
    resolving — the resolver re-checks the map result against the
    *active* master snapshot rather than trusting the map blindly."""
    row = (await db_session.execute(
        select(AssetLocation).where(AssetLocation.name == "美国")
    )).scalar_one()
    row.is_active = False
    await db_session.flush()

    resolver = await ManagedLocationResolver.load(db_session)
    assert resolver.resolve("美国") is None
    assert resolver.resolve("Santa Clara") is None


def test_normalization_map_pure_python_invariant():
    """Guard against map-typo regressions.

    Pure-python — no DB / fixture coupling — so it runs even when the
    full ``location_master`` fixture is unavailable.
    """
    snap = get_normalization_map_snapshot()
    assert snap, "normalization map must not be empty"
    for raw, canon in snap.items():
        assert canon, f"empty canonical for {raw!r}"
        assert canon == canon.strip(), f"unstripped canonical {canon!r}"
        assert canon != raw, (
            f"map entry {raw!r}→{canon!r} would loop (identity)"
        )
        assert not (canon.endswith("市") and len(canon) > 2), (
            f"canonical {canon!r} still carries 市 suffix"
        )


# ── One-shot convenience wrapper ───────────────────────────────────


@pytest.mark.asyncio
async def test_one_shot_resolve_managed_location(
    db_session: AsyncSession, location_master,
):
    assert await resolve_managed_location(db_session, "北京市") == "北京"
    assert await resolve_managed_location(db_session, "unknown") is None
    assert await resolve_managed_location(db_session, None) is None


# ── Boundary integration ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_claim_resolves_dirty_location(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset, op = await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(
            to_user_id=holder_person.id,
            location="Santa Clara",
        ),
        operator_person.id,
    )
    assert asset.location == "美国"
    assert op.detail["location"] == "美国"
    assert op.detail["location_raw"] == "Santa Clara"


@pytest.mark.asyncio
async def test_claim_unresolvable_protects_existing_location(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    idle_asset.location = "北京"
    await db_session.flush()
    svc = AssetService(db_session)
    asset, op = await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(
            to_user_id=holder_person.id,
            location="Mars Colony",
        ),
        operator_person.id,
    )
    assert asset.location == "北京"
    assert "location" not in (op.detail or {})
    assert (op.detail or {}).get("location_raw") == "Mars Colony"


@pytest.mark.asyncio
async def test_claim_exact_match_passes_through(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset, op = await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(to_user_id=holder_person.id, location="北京"),
        operator_person.id,
    )
    assert asset.location == "北京"
    assert op.detail["location"] == "北京"
    assert "location_raw" not in op.detail


@pytest.mark.asyncio
async def test_return_resolves_dirty_location(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(to_user_id=holder_person.id, location="北京"),
        operator_person.id,
    )
    asset, op = await svc.return_asset(
        idle_asset.id,
        AssetReturnRequest(location="杭州市"),
        operator_person.id,
    )
    assert asset.location == "杭州"
    assert op.detail["location"] == "杭州"
    assert op.detail["location_raw"] == "杭州市"


@pytest.mark.asyncio
async def test_borrow_resolves_dirty_location(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset, op = await svc.borrow_asset(
        idle_asset.id,
        AssetBorrowRequest(
            to_user_id=holder_person.id,
            expected_return_date=date.today() + timedelta(days=14),
            location="Würzburg",
        ),
        operator_person.id,
    )
    assert asset.location == "德国"
    assert op.detail["location"] == "德国"
    assert op.detail["location_raw"] == "Würzburg"


@pytest.mark.asyncio
async def test_change_assignee_resolves_dirty_location(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    second_holder: IdentityPerson,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(to_user_id=holder_person.id, location="北京"),
        operator_person.id,
    )
    asset, op = await svc.change_assignee(
        idle_asset.id,
        AssetChangeAssigneeRequest(
            to_user_id=second_holder.id,
            location="Toronto",
        ),
        operator_person.id,
    )
    assert asset.location == "加拿大"
    assert op.detail["location"] == "加拿大"


# ── create / update boundary ───────────────────────────────────────


@pytest.mark.asyncio
async def test_create_asset_drops_unresolvable_location(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"CR-{uuid.uuid4().hex[:8]}",
            asset_name="create-unresolvable",
            category="laptop",
            status="idle",
            location="Mars Colony",
        ),
        operator_person.id,
    )
    assert asset.location is None


@pytest.mark.asyncio
async def test_create_asset_normalizes_dirty_location(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"CR-{uuid.uuid4().hex[:8]}",
            asset_name="create-normalize",
            category="laptop",
            status="idle",
            location="Sunnyvale",
        ),
        operator_person.id,
    )
    assert asset.location == "美国"


@pytest.mark.asyncio
async def test_update_asset_unresolvable_protects_existing(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"UP-{uuid.uuid4().hex[:8]}",
            asset_name="update-unresolvable",
            category="laptop",
            status="idle",
            location="北京",
        ),
        operator_person.id,
    )
    assert asset.location == "北京"

    await svc.update_asset(
        asset.id,
        AssetUpdateRequest(location="Mars Colony"),
        operator_person.id,
    )
    refreshed = await db_session.get(AssetItem, asset.id)
    assert refreshed.location == "北京"


@pytest.mark.asyncio
async def test_update_asset_normalizes(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
):
    svc = AssetService(db_session)
    asset = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"UP-{uuid.uuid4().hex[:8]}",
            asset_name="update-normalize",
            category="laptop",
            status="idle",
            location="北京",
        ),
        operator_person.id,
    )

    await svc.update_asset(
        asset.id,
        AssetUpdateRequest(location="Kuala Lumpur"),
        operator_person.id,
    )
    refreshed = await db_session.get(AssetItem, asset.id)
    assert refreshed.location == "马来西亚"


# ── Regression — historical 2026-05-30 backfill cases ──────────────


@pytest.mark.asyncio
async def test_historical_2026_05_30_backfill_still_resolves(
    db_session: AsyncSession, location_master,
):
    """The trace_id ``f63a1c64-…`` + ``f5ebbdd4-…`` data fix sent
    these dirty values to specific canonical names. The runtime
    resolver MUST agree with that decision so we don't reintroduce
    drift via the live write path."""
    resolver = await ManagedLocationResolver.load(db_session)
    expected = {
        "Santa Clara": "美国",
        "Sunnyvale": "美国",
        "California": "美国",
        "北京市": "北京",
        "杭州市": "杭州",
        "上海市": "上海",
        "Kuala Lumpur": "马来西亚",
        "Rotterdam": "荷兰",
        "Würzburg": "德国",
    }
    for raw, canon in expected.items():
        got = resolver.resolve(raw)
        assert got == canon, f"{raw!r}: got {got!r}, expected {canon!r}"


# ── Round-3 coverage closes ───────────────────────────────────────


@pytest.mark.asyncio
async def test_update_asset_rejected_location_records_history_without_bumping_updated_at(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
):
    """Round-3 review medium: an update_asset PATCH with ONLY an
    unresolvable location must (a) leave the column untouched,
    (b) NOT bump asset.updated_at (the asset list page surfaces
    "recently updated" rows; lying violates the contract), AND
    (c) still write an AssetHistory row that names the rejected
    raw input + the rejection reason so admins can audit the
    intent."""
    from sqlalchemy import select as _select

    from app.domains.asset.models import AssetHistory

    svc = AssetService(db_session)
    asset = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"UP-{uuid.uuid4().hex[:8]}",
            asset_name="reject-audit",
            category="laptop",
            status="idle",
            location="北京",
        ),
        operator_person.id,
    )
    # Snapshot the freshness signal so we can compare after the
    # PATCH — the row was just created so updated_at == created_at,
    # but we want to confirm the rejection does NOT bump it.
    await db_session.refresh(asset)
    snapshot_updated_at = asset.updated_at

    await svc.update_asset(
        asset.id,
        AssetUpdateRequest(location="Mars Colony"),
        operator_person.id,
    )
    await db_session.flush()

    refreshed = await db_session.get(AssetItem, asset.id)
    assert refreshed.location == "北京"  # column preserved
    assert refreshed.updated_at == snapshot_updated_at  # freshness honest

    rows = (await db_session.execute(
        _select(AssetHistory)
        .where(AssetHistory.asset_id == asset.id)
        .where(AssetHistory.action == "updated")
    )).scalars().all()
    assert len(rows) == 1, "exactly one rejection-audit history row"
    notes = rows[0].notes or ""
    # The rejection marker IS present so admins reading the asset
    # detail timeline see that a PATCH was rejected.
    assert "rejected" in notes
    assert "managed asset_locations" in notes
    # The raw HRIS-derived submitted value MUST NOT be in the notes
    # (Round-4 review high) — it's exposed via /assets/{id} history
    # to non-admin viewers in scope, and via /assets/history to the
    # broader management-tier read role. The raw value lives only in
    # the structured logs (logger.warning above) for admin
    # correlation.
    assert "Mars Colony" not in notes


@pytest.mark.asyncio
async def test_bulk_claim_uses_shared_resolver_snapshot(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
    holder_person: IdentityPerson,
):
    """Round-3 review blocker: the resolver master snapshot is
    documented as per-request. A bulk operation must therefore see
    the same master across every iteration, even if the admin
    edits ``asset_locations`` mid-loop (which we simulate here by
    deactivating a row after the first call). Without the
    instance-cached resolver, the second call's resolver would
    re-load and miss the deactivated name."""
    svc = AssetService(db_session)

    asset1 = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"BLK-{uuid.uuid4().hex[:8]}",
            asset_name="bulk-1",
            category="laptop",
            status="idle",
        ),
        operator_person.id,
    )
    asset2 = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"BLK-{uuid.uuid4().hex[:8]}",
            asset_name="bulk-2",
            category="laptop",
            status="idle",
        ),
        operator_person.id,
    )

    # First claim warms the instance-cached resolver.
    await svc.claim_asset(
        asset1.id,
        AssetClaimRequest(to_user_id=holder_person.id, location="北京"),
        operator_person.id,
    )

    # Mid-bulk admin action: deactivate the "北京" master row. A
    # naive per-call resolver would now miss the row on asset2;
    # the cached resolver still has the warm snapshot.
    bj_row = (await db_session.execute(
        select(AssetLocation).where(AssetLocation.name == "北京")
    )).scalar_one()
    bj_row.is_active = False
    await db_session.flush()

    # Second claim — same service instance, same resolver snapshot.
    await svc.claim_asset(
        asset2.id,
        AssetClaimRequest(to_user_id=holder_person.id, location="北京"),
        operator_person.id,
    )

    refreshed2 = await db_session.get(AssetItem, asset2.id)
    assert refreshed2.location == "北京", (
        "bulk-loop resolver must use the snapshot loaded at first "
        "use, not re-read the (now-disabled) master mid-loop"
    )


@pytest.mark.asyncio
async def test_recycle_to_office_resolver_normalizes_office_city(
    db_session: AsyncSession,
    location_master,
    operator_person: IdentityPerson,
):
    """Round-3 review high (test gap): the recycle_to_office path
    routes ``office.city`` through the resolver. Verify a dirty
    office.city (``北京市``) is normalised to canonical ``北京``
    before landing on ``asset.location``."""
    from app.domains.asset.models import (
        AssetOfficeAddress,
    )

    # Seed an office with a "dirty" city string (operator-managed,
    # not FK-bound to asset_locations).
    office = AssetOfficeAddress(
        id=uuid.uuid4(),
        city="北京市",
        office_name="Beijing HQ",
        country_or_region="中国",
        is_active=True,
    )
    db_session.add(office)
    await db_session.flush()

    svc = AssetService(db_session)
    asset = await svc.create_asset(
        AssetCreateRequest(
            asset_tag=f"RC-{uuid.uuid4().hex[:8]}",
            asset_name="recycle-normalize",
            category="laptop",
            status="idle",
            location="北京",
        ),
        operator_person.id,
    )
    # Re-route through recycle_to_office with the dirty-city office.
    result = await svc.recycle_to_office(
        asset.id,
        office_id=office.id,
        reason="resolver-coverage-test",
        operator_id=operator_person.id,
        source="manual",
    )
    assert result is not None
    refreshed = await db_session.get(AssetItem, asset.id)
    assert refreshed.location == "北京", (
        "office.city '北京市' must normalize to managed '北京' on recycle"
    )


@pytest.mark.asyncio
async def test_serialize_operations_strips_location_raw_for_non_admin(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    """Round-5 review high (PII leak): ``serialize_operations`` is the
    list-endpoint twin of ``enrich_operations_with_users``. Without
    the same ``admin_view`` gate, ``GET /assets/operations?to_user_id=<self>``
    (employee scope) would leak ``detail.location_raw`` to the
    employee — bypassing the detail-endpoint gate the Round-3 fix
    added. This test pins both branches."""
    from app.domains.asset.service import serialize_operations
    svc = AssetService(db_session)
    _asset, op = await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(
            to_user_id=holder_person.id,
            location="Santa Clara",
        ),
        operator_person.id,
    )

    admin_resp = await serialize_operations(
        db_session, [op], admin_view=True,
    )
    assert admin_resp[0].detail["location_raw"] == "Santa Clara"
    assert admin_resp[0].detail["location"] == "美国"

    employee_resp = await serialize_operations(
        db_session, [op], admin_view=False,
    )
    assert "location_raw" not in employee_resp[0].detail
    assert employee_resp[0].detail["location"] == "美国"
    # Source ORM row is not mutated.
    assert op.detail["location_raw"] == "Santa Clara"


@pytest.mark.asyncio
async def test_non_admin_view_strips_location_raw_from_operation_detail(
    db_session: AsyncSession,
    location_master,
    idle_asset: AssetItem,
    holder_person: IdentityPerson,
    operator_person: IdentityPerson,
):
    """Round-3 review medium (PII leak): a non-admin viewer of an
    asset's operation log must NOT see ``detail.location_raw`` —
    that field carries the operator's raw HRIS-derived input
    (``Santa Clara``, ``北京市``, ``Würzburg``) which is an
    admin-only audit signal. The enrich helper gates this on
    ``admin_view``."""
    svc = AssetService(db_session)
    _asset, op = await svc.claim_asset(
        idle_asset.id,
        AssetClaimRequest(
            to_user_id=holder_person.id,
            location="Santa Clara",
        ),
        operator_person.id,
    )
    # Admin view: location_raw is present.
    admin = await svc.enrich_operations_with_users([op], admin_view=True)
    assert admin[0]["detail"]["location_raw"] == "Santa Clara"
    assert admin[0]["detail"]["location"] == "美国"

    # Non-admin view: same op, but location_raw stripped.
    nonadmin = await svc.enrich_operations_with_users([op], admin_view=False)
    assert "location_raw" not in nonadmin[0]["detail"]
    # The canonical location stays visible — it's already the
    # managed name and matches what the dropdown carries.
    assert nonadmin[0]["detail"]["location"] == "美国"
    # Source object MUST NOT be mutated.
    assert op.detail["location_raw"] == "Santa Clara"