"""Migration 535 — move asset authority from offices to locations.

The backfill is the whole risk of this migration. It copies every
office's admin list onto the managed location that office sits at, and
then rebuilds the ``asset_admin`` grant set from the new table. If the
copy misses a row, the rebuild silently revokes a real administrator's
access and nobody finds out until they hit a 403 — so the migration
refuses to continue when any office admin cannot be mapped, and this
file pins that behaviour rather than trusting it.

Also pinned: the primary flag survives the copy (single-recipient flows
read it), the role rebuild is by set difference so re-running is a
no-op, and ``asset_global_admin`` keeps co-deriving ``asset_admin``
without any location of its own.
"""
from __future__ import annotations

import importlib.util
import pathlib
import uuid

import pytest
import sqlalchemy as sa
from alembic.migration import MigrationContext
from alembic.operations import Operations

from ..conftest import _TEST_SYNC_URL

_MIGRATION = (
    pathlib.Path(__file__).resolve().parents[2]
    / "alembic" / "versions" / "535_asset_location_admins.py"
)


def _load():
    spec = importlib.util.spec_from_file_location("m535", _MIGRATION)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# Every row this file seeds is prefixed so teardown can find it. The
# async ``db_session`` fixture truncates between tests; this file talks
# to a raw sync engine that does not, and a leaked office with no
# location makes the very next test raise the "cannot map" error for
# somebody else's data.
_PREFIX = "m535"


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        _ensure_bridge_column(c)
        try:
            yield c
        finally:
            _cleanup(c)
    engine.dispose()


def _ensure_bridge_column(conn) -> None:
    """Recreate the schema state migration 535 actually runs against.

    In the real chain 525 adds ``asset_office_addresses.asset_location``
    and 536 drops it, so 535 always sees the column. The test schema is
    built by ``create_all`` from the CURRENT models, where 536 has
    already removed it — a state 535 never encounters in production.
    Without this the test fails on a column the migration is right to
    depend on, which would push us to weaken the migration to satisfy a
    fixture that does not reflect reality.
    """
    if _has_bridge_column(conn):
        return
    conn.execute(
        sa.text(
            "ALTER TABLE asset_office_addresses "
            "ADD COLUMN asset_location VARCHAR(128) NULL"
        )
    )


def _cleanup(conn) -> None:
    legacy = _legacy_table(conn)
    conn.execute(sa.text(
        "DELETE FROM asset_location_admins WHERE person_id IN "
        "(SELECT id FROM identity_persons WHERE work_email LIKE :p)"
    ), {"p": f"{_PREFIX}-%"})
    conn.execute(sa.text(
        f"DELETE FROM {legacy} WHERE person_id IN "
        "(SELECT id FROM identity_persons WHERE work_email LIKE :p)"
    ), {"p": f"{_PREFIX}-%"})
    conn.execute(sa.text(
        "DELETE FROM auth_user_roles WHERE user_id IN "
        "(SELECT id FROM identity_persons WHERE work_email LIKE :p)"
    ), {"p": f"{_PREFIX}-%"})
    conn.execute(sa.text(
        "DELETE FROM asset_office_addresses WHERE city LIKE :p"
    ), {"p": f"{_PREFIX.upper()}%"})
    conn.execute(sa.text(
        "DELETE FROM asset_location_admins WHERE location_id IN "
        "(SELECT id FROM asset_locations WHERE code LIKE :p)"
    ), {"p": f"{_PREFIX.upper()}%"})
    conn.execute(sa.text(
        "DELETE FROM asset_locations WHERE code LIKE :p OR name IN ('巴西', '瑞典')"
    ), {"p": f"{_PREFIX.upper()}%"})
    conn.execute(sa.text(
        "DELETE FROM identity_persons WHERE work_email LIKE :p"
    ), {"p": f"{_PREFIX}-%"})


def _run(conn, direction: str = "upgrade") -> None:
    ctx = MigrationContext.configure(conn)
    with Operations.context(ctx):
        getattr(_load(), direction)()


def _person(conn, email: str) -> str:
    pid = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO identity_persons "
            "(id, source_system, hris_person_id, work_email, display_name, "
            " employment_status, is_china_employee, hris_record_state, "
            " last_synced_at, created_at, updated_at) "
            "VALUES (:id, 'successfactors', :hris, :email, :name, "
            " 'active', 0, 'present', NOW(), NOW(), NOW())"
        ),
        {
            "id": pid,
            "hris": f"sf-{pid[:8]}",
            "email": email,
            "name": email.split("@")[0],
        },
    )
    return pid


def _location(conn, code: str, name: str) -> str:
    lid = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO asset_locations "
            "(id, code, name, sort_order, is_active, created_at, updated_at) "
            "VALUES (:id, :code, :name, 900, 1, NOW(), NOW())"
        ),
        {"id": lid, "code": code, "name": name},
    )
    return lid


def _office(conn, city: str, asset_location: str | None) -> str:
    oid = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO asset_office_addresses "
            "(id, city, office_name, is_active, created_at, updated_at) "
            "VALUES (:id, :city, :city, 1, NOW(), NOW())"
        ),
        {"id": oid, "city": city},
    )
    # The bridge column still exists at this point in the chain; a
    # deployment that has already dropped it falls back to the city
    # match, which is exactly the second tier under test.
    if asset_location is not None and _has_bridge_column(conn):
        conn.execute(
            sa.text(
                "UPDATE asset_office_addresses SET asset_location = :loc "
                "WHERE id = :id"
            ),
            {"loc": asset_location, "id": oid},
        )
    return oid


def _has_bridge_column(conn) -> bool:
    return bool(
        conn.execute(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.COLUMNS "
                "WHERE TABLE_SCHEMA = DATABASE() "
                "  AND TABLE_NAME = 'asset_office_addresses' "
                "  AND COLUMN_NAME = 'asset_location'"
            )
        ).scalar()
    )


def _legacy_table(conn) -> str:
    """The office admin join, whichever name this schema carries."""
    exists = conn.execute(
        sa.text(
            "SELECT COUNT(*) FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'asset_office_admins'"
        )
    ).scalar()
    return "asset_office_admins" if exists else "asset_office_local_admins"


def _office_admin(conn, office_id: str, person_id: str, *, primary: bool) -> None:
    conn.execute(
        sa.text(
            f"INSERT INTO {_legacy_table(conn)} "
            "(office_id, person_id, is_primary, created_at) "
            "VALUES (:o, :p, :prim, NOW())"
        ),
        {"o": office_id, "p": person_id, "prim": 1 if primary else 0},
    )


def _role_id(conn, name: str) -> str:
    rid = conn.execute(
        sa.text("SELECT id FROM auth_roles WHERE name = :n"), {"n": name},
    ).scalar()
    if rid:
        return rid
    rid = str(uuid.uuid4())
    conn.execute(
        sa.text(
            "INSERT INTO auth_roles (id, name, description, is_system, created_at) "
            "VALUES (:id, :n, :n, 1, NOW())"
        ),
        {"id": rid, "n": name},
    )
    return rid


def _holds(conn, person_id: str, role_name: str) -> bool:
    return bool(
        conn.execute(
            sa.text(
                "SELECT COUNT(*) FROM auth_user_roles ur "
                "JOIN auth_roles r ON r.id = ur.role_id "
                "WHERE ur.user_id = :p AND r.name = :n"
            ),
            {"p": person_id, "n": role_name},
        ).scalar()
    )


def _location_admins(conn, location_id: str) -> dict[str, int]:
    rows = conn.execute(
        sa.text(
            "SELECT person_id, is_primary FROM asset_location_admins "
            "WHERE location_id = :l"
        ),
        {"l": location_id},
    ).fetchall()
    return {r[0]: int(r[1]) for r in rows}


def test_backfill_copies_admins_and_rebuilds_the_role(conn) -> None:
    _role_id(conn, "asset_admin")
    beijing = _location(conn, "M535BJ", "M535北京")
    singapore = _location(conn, "M535SG", "M535新加坡")

    # Office mapped through the explicit bridge column.
    bj_office = _office(conn, "M535北京", "M535北京")
    bj_admin = _person(conn, "m535-bj@test.com")
    _office_admin(conn, bj_office, bj_admin, primary=True)

    # Two admins, one flagged primary — the flag has to survive.
    sg_office = _office(conn, "M535新加坡", "M535新加坡")
    sg_primary = _person(conn, "m535-sg-primary@test.com")
    sg_second = _person(conn, "m535-sg-second@test.com")
    _office_admin(conn, sg_office, sg_primary, primary=True)
    _office_admin(conn, sg_office, sg_second, primary=False)

    _run(conn)

    assert _location_admins(conn, beijing) == {bj_admin: 1}
    assert _location_admins(conn, singapore) == {sg_primary: 1, sg_second: 0}
    for pid in (bj_admin, sg_primary, sg_second):
        assert _holds(conn, pid, "asset_admin"), pid


def test_office_without_bridge_column_maps_by_city_name(conn) -> None:
    """An office whose bridge column was never filled in still maps,
    as long as its city names a managed location."""
    _role_id(conn, "asset_admin")
    location = _location(conn, "M535CD", "M535成都")
    office = _office(conn, "M535成都", None)
    admin = _person(conn, "m535-cd@test.com")
    _office_admin(conn, office, admin, primary=True)

    _run(conn)

    assert _location_admins(conn, location) == {admin: 1}
    assert _holds(conn, admin, "asset_admin")


def test_unmappable_office_admin_stops_the_migration(conn) -> None:
    """Refuse rather than silently revoke.

    An office admin we cannot place on a location would be dropped by
    the role rebuild with no trace. Failing the deploy is loud, and
    recoverable in a minute by pointing the office at a location.
    """
    _role_id(conn, "asset_admin")
    office = _office(conn, "M535NoSuchCity", None)
    admin = _person(conn, "m535-orphan@test.com")
    _office_admin(conn, office, admin, primary=True)

    with pytest.raises(RuntimeError, match="cannot map every office admin"):
        _run(conn)


def test_rerun_is_a_noop(conn) -> None:
    _role_id(conn, "asset_admin")
    location = _location(conn, "M535HZ", "M535杭州")
    office = _office(conn, "M535杭州", "M535杭州")
    admin = _person(conn, "m535-hz@test.com")
    _office_admin(conn, office, admin, primary=True)

    _run(conn)
    first = _location_admins(conn, location)
    granted_at = conn.execute(
        sa.text(
            "SELECT granted_at FROM auth_user_roles ur "
            "JOIN auth_roles r ON r.id = ur.role_id "
            "WHERE ur.user_id = :p AND r.name = 'asset_admin'"
        ),
        {"p": admin},
    ).scalar()

    _run(conn)

    assert _location_admins(conn, location) == first
    # The grant is left alone, not deleted and re-inserted — the
    # access-cycle ledger reads ``granted_at`` as the start of the
    # holding interval, and churning it would fabricate a gap.
    assert conn.execute(
        sa.text(
            "SELECT granted_at FROM auth_user_roles ur "
            "JOIN auth_roles r ON r.id = ur.role_id "
            "WHERE ur.user_id = :p AND r.name = 'asset_admin'"
        ),
        {"p": admin},
    ).scalar() == granted_at


def test_global_asset_admin_keeps_the_role_without_a_location(conn) -> None:
    """The location-less grant source has to survive the rebuild.

    Without the union in the rebuild SQL, every global asset admin who
    is not also listed against a location would be stripped of
    ``asset_admin`` the moment this migration runs.
    """
    asset_admin_role = _role_id(conn, "asset_admin")
    global_role = _role_id(conn, "asset_global_admin")
    person = _person(conn, "m535-global@test.com")
    conn.execute(
        sa.text(
            "INSERT INTO auth_user_roles (user_id, role_id, granted_at) "
            "VALUES (:p, :r, NOW())"
        ),
        {"p": person, "r": global_role},
    )
    conn.execute(
        sa.text(
            "INSERT INTO auth_user_roles (user_id, role_id, granted_at) "
            "VALUES (:p, :r, NOW())"
        ),
        {"p": person, "r": asset_admin_role},
    )

    _run(conn)

    assert _holds(conn, person, "asset_admin")


def test_departed_office_admin_is_not_granted(conn) -> None:
    """The rebuild filters on employment status, so a leaver whose join
    row was never swept does not come back as an administrator."""
    _role_id(conn, "asset_admin")
    location = _location(conn, "M535GZ", "M535广州")
    office = _office(conn, "M535广州", "M535广州")
    leaver = _person(conn, "m535-leaver@test.com")
    conn.execute(
        sa.text(
            "UPDATE identity_persons SET employment_status = 'leaver_pending' "
            "WHERE id = :p"
        ),
        {"p": leaver},
    )
    _office_admin(conn, office, leaver, primary=True)

    _run(conn)

    # The row is copied (it is a faithful copy of the office list) …
    assert leaver in _location_admins(conn, location)
    # … but it grants nothing.
    assert not _holds(conn, leaver, "asset_admin")


def test_seeds_the_missing_managed_locations(conn) -> None:
    """巴西 / 瑞典 exist afterwards, so the resolver map entries added in
    the same PR have a master row to resolve to."""
    _role_id(conn, "asset_admin")
    _run(conn)

    for name in ("巴西", "瑞典"):
        assert conn.execute(
            sa.text("SELECT COUNT(*) FROM asset_locations WHERE name = :n"),
            {"n": name},
        ).scalar() == 1