import os
import re
import uuid
from collections.abc import AsyncGenerator

# pytest-xdist isolation: every worker MUST get its own Redis logical DB
# before any ``app.*`` import runs. ``app.core.config.settings`` snapshots
# REDIS_URL at import time and is then read by ``redis_client`` and the
# rate-limit middleware in ``app.main`` — once those modules bind to a
# pool they ignore later env mutations. Doing the rewrite here, at the
# very top of conftest, means the first ``from app.<x>`` below picks up
# the worker-scoped URL. Single-process runs (no PYTEST_XDIST_WORKER) are
# untouched, so local ``pytest`` keeps using DB 0.
_xdist_worker = os.environ.get("PYTEST_XDIST_WORKER", "")
if _xdist_worker:
    _digits = re.search(r"\d+", _xdist_worker)
    _redis_db = (int(_digits.group()) % 16) if _digits else 0
    _existing_redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
    os.environ["REDIS_URL"] = re.sub(r"/\d+$", f"/{_redis_db}", _existing_redis_url)

import pytest
import pytest_asyncio
import sqlalchemy as sa
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.approval.models import *  # noqa: F401, F403
from app.audit.models import *  # noqa: F401, F403

# Import all models so Base.metadata knows about them
from app.auth.models import *  # noqa: F401, F403
from app.compliance.models import *  # noqa: F401, F403
from app.core.database import Base, get_db
from app.domains.asset.models import *  # noqa: F401, F403
from app.domains.automation_exception.models import *  # noqa: F401, F403
from app.domains.bailian.models import *  # noqa: F401, F403
from app.domains.cmdb.models import *  # noqa: F401, F403
from app.domains.data_access.models import *  # noqa: F401, F403
from app.domains.email_group.models import *  # noqa: F401, F403
from app.domains.feed.models import *  # noqa: F401, F403
from app.domains.hr.models import *  # noqa: F401, F403
from app.identity.person_models import *  # noqa: F401, F403
from app.main import app
from app.notifications.models import *  # noqa: F401, F403
from app.scheduled_jobs.models import *  # noqa: F401, F403
from tests.support.database_bootstrap import resolve_test_database_config

pytest_plugins = ("tests.support.hang_diagnostics",)

_SYSTEM_ROLES: list[tuple[str, str]] = [
    ("global_admin", "Global administrator"),
    ("employee", "Regular user"),
    ("auditor", "Audit access"),
    ("cloud_admin", "Cloud administrator"),
    ("asset_admin", "Asset administrator"),
    # Seeded by alembic migration 275 in real envs. Global asset admin:
    # manually granted, office-region bypass inside the asset domain;
    # holders co-derive ``asset_admin`` via location_admin_sync.
    ("asset_global_admin", "Global asset administrator"),
    # Seeded by alembic migration 043 in real envs; tests need it
    # explicitly so hr_tools / HR-router endpoints can be exercised
    # under the ``hr`` role.
    ("hr", "HR administrator"),
    # Seeded by alembic migration in real envs; tests need it so the
    # finance read surface (software cost / billing / procurement) can
    # be exercised under the ``finance`` role.
    ("finance", "Finance read-only"),
    # Seeded by alembic migration in real envs; tests need it so the
    # leadership read-only surfaces (insights impact scorecard, admin
    # cockpits) can be exercised under the ``management`` role.
    ("management", "Leadership read-only"),
    # Seeded by alembic migration 567 in real envs. GitHub 组织管理员:
    # GitHub 组织域内读写(组织登记、自动加入规则、成员台账、对账处置),
    # 域外无任何权限。
    ("github_admin", "GitHub organization administrator"),
]

# (asset_category, code, name) — asset-tag category segment. Seeded by
# alembic migration 205 in real envs; tests build the schema via
# ``create_all`` (no migrations), so asset-intake tag generation needs
# these rows seeded explicitly or it 400s with CATEGORY_TAG_CODE_MISSING.
_ASSET_CATEGORIES: list[tuple[str, str, str]] = [
    ("laptop", "NB", "笔记本"),
    ("desktop", "DT", "台式机"),
    ("monitor", "MT", "显示器"),
    ("printer", "PR", "打印机"),
    ("server", "SV", "服务器"),
    ("network_device", "ND", "网络设备"),
    ("accessory", "AC", "配件"),
    ("conference_display", "CD", "会议电视"),
]

_TEST_ASYNC_BASE = os.environ.get(
    "TEST_DATABASE_URL",
    "mysql+aiomysql://helix:helix@localhost:3306/helix",
)
_TEST_SYNC_BASE = os.environ.get(
    "TEST_DATABASE_URL_SYNC",
    "mysql+pymysql://helix:helix@localhost:3306/helix",
)
_TEST_DB_CONFIG = resolve_test_database_config(
    test_async_base=_TEST_ASYNC_BASE,
    test_sync_base=_TEST_SYNC_BASE,
    admin_url=os.environ.get("TEST_DATABASE_ADMIN_URL"),
)
TEST_DB_URL = _TEST_DB_CONFIG.async_url
_TEST_SYNC_URL = TEST_DB_URL.replace("mysql+aiomysql://", "mysql+pymysql://").replace(
    "mysql+asyncmy://", "mysql+pymysql://"
)

# Sorted so we wipe child tables before parents. FOREIGN_KEY_CHECKS=0 makes
# this order non-critical, but the explicit ordering is cheap insurance.
_TABLE_NAMES_FOR_TRUNCATE: list[str] = [
    table.name for table in reversed(Base.metadata.sorted_tables)
]

# Tables that carry an ``AUTO_INCREMENT`` column. The per-test reset wipes
# these with ``TRUNCATE`` (which resets the counter to 1) and every other
# table with ``DELETE`` (which does not). Rationale below in ``db_engine``;
# this set is populated once per session by ``_prepare_test_schema`` from
# ``information_schema`` so it self-maintains when a new auto-increment
# table is added — no hand-kept list to drift.
_AUTOINCREMENT_TABLES: set[str] = set()


@pytest.fixture(scope="session", autouse=True)
def _prepare_test_schema():
    """Drop and recreate the test schema ONCE per pytest session.

    Previously `db_engine` did drop_all + create_all on every test function,
    which cost ~1-2s of DDL per test against MySQL (40+ tables × 281 tests =
    most of CI runtime). Schema is stable across a single test run, so we
    build it once here and rely on TRUNCATE between tests for data isolation.

    Under pytest-xdist, this runs once per worker against the worker's own
    ``helix_test_gwN`` schema, plus a one-shot Redis FLUSHDB so any leftover
    keys from a previous CI run on the same logical DB don't leak into this
    session.
    """
    sync_engine = sa.create_engine(_TEST_SYNC_URL)
    try:
        with sync_engine.begin() as conn:
            conn.execute(sa.text("SET FOREIGN_KEY_CHECKS = 0"))
            inspector = sa.inspect(conn)
            for table_name in inspector.get_table_names():
                conn.execute(sa.text(f"DROP TABLE IF EXISTS `{table_name}`"))
            conn.execute(sa.text("SET FOREIGN_KEY_CHECKS = 1"))
        # FK checks OFF during create_all too. ``Base.metadata.sorted_tables``
        # sorts by FK dependency, but TiDB / asyncmy occasionally trip on a
        # CREATE TABLE whose FK target was just emitted in the SAME multi-
        # statement transaction (race observed locally 2026-05-29: `1824
        # Failed to open the referenced table 'approval_requests'` from a
        # `CREATE TABLE approval_steps` step). Disabling FK_CHECKS during
        # creation defers the constraint resolution to first row insert,
        # which is well after every table is materialised — matches the
        # `mysqldump --no-data` recipe most ops teams use. Same envelope
        # the DROP block above uses, just on the other side.
        with sync_engine.begin() as conn:
            conn.execute(sa.text("SET FOREIGN_KEY_CHECKS = 0"))
            Base.metadata.create_all(conn)
            conn.execute(sa.text("SET FOREIGN_KEY_CHECKS = 1"))

        # Discover which tables carry an AUTO_INCREMENT column so the
        # per-test reset (``db_engine``) can TRUNCATE just those (to reset
        # the counter) and DELETE the rest. Queried once here so the
        # distinction self-maintains as the schema evolves.
        with sync_engine.begin() as conn:
            rows = conn.execute(
                sa.text(
                    "SELECT DISTINCT table_name FROM information_schema.columns "
                    "WHERE table_schema = DATABASE() "
                    "AND extra LIKE '%auto_increment%'"
                )
            ).scalars().all()
        _AUTOINCREMENT_TABLES.clear()
        _AUTOINCREMENT_TABLES.update(rows)
    finally:
        sync_engine.dispose()

    # Wipe the worker-scoped Redis DB once — defends against leftover state
    # if a prior xdist run on the same logical DB crashed mid-test. Using a
    # fresh sync client here so we don't have to reach into the asyncio
    # event-loop machinery just for a flush.
    try:
        import redis as _redis_sync

        _r = _redis_sync.Redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379/0"))
        _r.flushdb()
        _r.close()
    except Exception:
        # Tests that don't actually need Redis still pass; failure here
        # shouldn't block the whole session.
        pass

    yield


async def _get_role_id(db_session: AsyncSession, role_name: str) -> uuid.UUID:
    result = await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = :role_name"),
        {"role_name": role_name},
    )
    role_id = result.scalar_one()
    return uuid.UUID(role_id) if isinstance(role_id, str) else role_id


@pytest.fixture(autouse=True)
def _no_route_leaks_onto_the_shared_app():
    """No test may leave anything mounted on the process-global ``app``.

    ``app.main:app`` is built once and shared by every test in the session.
    A few gate tests mount a throwaway probe router to prove a sweep isn't
    vacuous; each has to take it back off. The pattern that used to be
    written for that — filtering ``app.router.routes`` by ``.path`` — is a
    no-op under FastAPI 0.141, where ``include_router`` appends one deferred
    ``_IncludedRouter`` wrapper with no ``.path``. The probe survived the
    whole session and made every later ``app.openapi()`` raise on its
    unresolved annotation, which is what broke
    ``test_knowledge_loop``/``test_partner_directory``'s
    ``test_the_endpoint_list_matches_the_router`` in full runs while both
    passed alone.

    Comparing route identity (not just count) is ~150 ``id()`` calls per
    test. Use ``tests.support.router_probe.mounted_router`` to mount a probe
    and this stays quiet.
    """
    before = [id(r) for r in app.router.routes]
    yield
    after = [id(r) for r in app.router.routes]
    assert after == before, (
        "this test changed the shared app's route table and did not put it "
        f"back ({len(before)} routes -> {len(after)}). Mount probe routers "
        "with tests.support.router_probe.mounted_router — filtering "
        "app.router.routes by `.path` does NOT remove an included router "
        "under FastAPI 0.141."
    )


@pytest.fixture(autouse=True)
def _disable_impact_cache():
    """The insights impact endpoint carries a per-process TTL cache
    (``INSIGHTS_IMPACT_CACHE_SECONDS``, default 60 s) so the projection
    board's polling can't amplify DB load. The cache dict is process-
    global — ANY test that touches ``/dashboard/insights/impact``
    (directly or via the export dispatcher) would otherwise read a
    stale payload cached by an earlier test. Session-wide autouse:
    disable + clear around every test; the dedicated TTL test re-enables
    it locally.
    """
    from app.core.config import settings
    from app.dashboard import insights_router

    prev = settings.INSIGHTS_IMPACT_CACHE_SECONDS
    settings.INSIGHTS_IMPACT_CACHE_SECONDS = 0
    insights_router._IMPACT_CACHE.clear()
    yield
    settings.INSIGHTS_IMPACT_CACHE_SECONDS = prev
    insights_router._IMPACT_CACHE.clear()


@pytest.fixture(autouse=True)
def _clear_kb_vector_cache():
    """The KB's embedding corpus is cached per process, keyed on
    ``(count, MAX(updated_at))`` over the published+embedded rows.

    In production that signature moves whenever the corpus does. Across TESTS it
    does not: two tests that each seed two embedded articles inside the same
    second produce the SAME signature, so the second would rank its query
    against the first test's vectors. Same precedent as
    ``_disable_impact_cache`` — clear around every test; the dedicated caching
    test drives the cache explicitly.
    """
    from app.domains.support import kb_service

    kb_service.reset_vector_cache()
    yield
    kb_service.reset_vector_cache()


@pytest.fixture(autouse=True)
def _disable_cost_read_cache():
    """成本口径读面(财务瀑布 / methodology 对数区)带 Redis 响应缓存
    (``COST_READ_CACHE_TTL_SECONDS``)。Redis 只在会话开头 FLUSHDB、
    不随每测清理——任何触碰这两个端点的测试都可能读到前一个测试
    缓存的聚合。同 ``_disable_impact_cache`` 先例:全程置 0,专门的
    缓存测试再局部拉起。
    """
    from app.core.config import settings

    prev = settings.COST_READ_CACHE_TTL_SECONDS
    settings.COST_READ_CACHE_TTL_SECONDS = 0
    yield
    settings.COST_READ_CACHE_TTL_SECONDS = prev


@pytest_asyncio.fixture(autouse=True)
async def _reset_redis_between_tests():
    """Give every test an empty worker-scoped Redis DB and a fresh pool.

    pytest-asyncio uses function-scoped event loops by default, so a cached
    Redis pool from one test becomes useless in the next. Drop the singleton
    before each test so get_redis() rebuilds against the current loop.

    The xdist bootstrap at the top of this file assigns each worker its own
    logical Redis DB, but tests within one ``loadscope`` worker still share
    that DB.  In particular, every ASGI test client has the same source IP;
    login-rate-limit attempts therefore leaked between otherwise independent
    auth tests and caused order-dependent false 429s.  Flush only this
    worker's DB before the test starts.  Production key names and limits stay
    untouched, and concurrently running xdist workers remain isolated.
    """
    from app.core import redis_client

    redis_client._redis = None
    try:
        await redis_client.get_redis().flushdb()
    except Exception:
        if _xdist_worker:
            # Parallel CI requires Redis. Silently continuing here would
            # reintroduce the order-dependent state leak this fixture guards.
            raise
        # Keep the existing test contract: suites that do not require Redis
        # can still run without xdist when a local Redis service is absent.
        # Tests which do require it will fail at their own assertion boundary.
        pass
    yield
    existing = redis_client._redis
    redis_client._redis = None
    if existing is not None:
        try:
            await existing.aclose()
        except Exception:
            pass


@pytest_asyncio.fixture
async def db_engine(_prepare_test_schema):
    """Reset data between tests (schema created once per session).

    Most tables are wiped with ``DELETE`` rather than ``TRUNCATE``. On
    InnoDB, ``TRUNCATE`` is a DDL that drops and recreates the table's
    tablespace on disk *even when the table is empty* — wiping all 140+
    tables this way cost ~1.3s per test (measured), which dominated CI
    wall-clock (140 tables × hundreds of tests). ``DELETE`` on an empty
    table is a near-free metadata check (~0.05s for the whole set), and
    test tables only ever hold a handful of rows, so it stays cheap when
    they aren't empty. ``FOREIGN_KEY_CHECKS`` is disabled so order does
    not matter; the test schema is built via ``create_all`` (no triggers),
    so ``DELETE`` has no trigger side-effects that ``TRUNCATE`` would skip.

    The two AUTO_INCREMENT tables are still ``TRUNCATE``-d so their counter
    resets to 1 each test (``DELETE`` would not), preserving prior
    behaviour for any test that relies on a deterministic first id.

    System roles / category seeds are re-inserted after the wipe because
    tests expect them to be present.
    """
    engine = create_async_engine(TEST_DB_URL, echo=False)

    async with engine.begin() as conn:
        await conn.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
        for table_name in _TABLE_NAMES_FOR_TRUNCATE:
            if table_name in _AUTOINCREMENT_TABLES:
                await conn.execute(text(f"TRUNCATE TABLE `{table_name}`"))
            else:
                await conn.execute(text(f"DELETE FROM `{table_name}`"))
        await conn.execute(text("SET FOREIGN_KEY_CHECKS = 1"))

        # System-role seed uses ``INSERT … ON DUPLICATE KEY UPDATE`` so
        # a crashed prior session that left ``auth_roles`` rows behind
        # — TRUNCATE auto-commits in MySQL, so a SIGKILL between
        # TRUNCATE-of-children and TRUNCATE-of-auth_roles is enough to
        # leave a stray ``global_admin`` row — doesn't poison every
        # subsequent run with ``IntegrityError: Duplicate entry``.
        # That failure mode hit during the AK-20260509-EF8A diagnostic
        # session; codified self-healing to prevent recurrence.
        for role_name, role_desc in _SYSTEM_ROLES:
            await conn.execute(
                text(
                    "INSERT INTO auth_roles (id, name, description, is_system) "
                    "VALUES (:id, :name, :desc, 1) "
                    "ON DUPLICATE KEY UPDATE "
                    "description = VALUES(description), "
                    "is_system = VALUES(is_system)"
                ),
                {
                    "id": str(uuid.uuid4()),
                    "name": role_name,
                    "desc": role_desc,
                },
            )

        for asset_category, code, name in _ASSET_CATEGORIES:
            await conn.execute(
                text(
                    "INSERT INTO asset_category_configs "
                    "(id, code, name, asset_category, sort_order, is_active, "
                    " created_at, updated_at) "
                    "VALUES (:id, :code, :name, :key, 0, 1, NOW(), NOW()) "
                    "ON DUPLICATE KEY UPDATE "
                    "asset_category = VALUES(asset_category), "
                    "name = VALUES(name), is_active = 1"
                ),
                {
                    "id": str(uuid.uuid4()),
                    "code": code,
                    "name": name,
                    "key": asset_category,
                },
            )

        # Seed procurement_categories — mirrors migration 222's starter
        # set. Tests build the schema via ``create_all`` (no migrations),
        # so any code that expects the four Phase-1 codes (software /
        # hardware / cloud / service) to exist needs them seeded here.
        for code, name_zh, name_en, semantics, days, sort in (
            ("software", "软件 / SaaS", "Software / SaaS", "software_activation", 60, 10),
            ("hardware", "硬件", "Hardware", "hardware_provision", None, 20),
            ("cloud", "云服务", "Cloud Services", "software_activation", 30, 30),
            ("service", "服务采购", "Service", "service_acceptance", None, 40),
        ):
            await conn.execute(
                text(
                    "INSERT INTO procurement_categories "
                    "(id, code, name_zh, name_en, receipt_semantics, "
                    " default_reminder_days, is_active, sort_order, "
                    " created_at, updated_at) "
                    "VALUES (:id, :code, :nz, :ne, :rs, :rd, 1, :so, NOW(), NOW()) "
                    "ON DUPLICATE KEY UPDATE name_zh = VALUES(name_zh)"
                ),
                {
                    "id": str(uuid.uuid4()),
                    "code": code,
                    "nz": name_zh,
                    "ne": name_en,
                    "rs": semantics,
                    "rd": days,
                    "so": sort,
                },
            )
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def db_session(db_engine) -> AsyncGenerator[AsyncSession, None]:
    """Provide a DB session for each test."""
    session_factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with session_factory() as session:
        yield session


@pytest_asyncio.fixture
async def client(db_engine, db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """HTTP test client with dependency override for DB session.

    Two layers of session-injection are wired here:

    1. ``get_db`` is overridden so FastAPI route handlers receive the
       test session.
    2. ``app.core.database.async_session_factory`` is monkeypatched to
       a factory bound to the test engine. ASGI middleware that runs
       before the FastAPI dependency stack (notably the MCP PAT auth
       middleware) and any background-task code that opens its own
       session via the module-level factory therefore also lands on
       the test database. Without this second hook the MCP endpoint
       would query the production DB and never find tokens written by
       a test.
    """
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    from app.core import database as _db
    test_factory = async_sessionmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    original_factory = _db.async_session_factory
    _db.async_session_factory = test_factory

    # FastMCP's Streamable-HTTP transport needs its session manager
    # running to serve any request. ``httpx.ASGITransport`` doesn't
    # invoke lifespan, so we drive it via ``asgi-lifespan``.
    #
    # Wrinkle: ``StreamableHTTPSessionManager.run()`` raises
    # ``RuntimeError: can only be called once per instance`` on second
    # entry. The session_manager itself is created lazily by
    # ``streamable_http_app()`` (which we already called at app build
    # time), so we can't simply reset it to None and let it lazy-init.
    # Instead, flip the private ``_has_started`` flag back to False
    # before each test so the same instance can be re-run. This is the
    # least-invasive way to share the registered-tool state across
    # tests while still getting a clean lifespan per test.
    from asgi_lifespan import LifespanManager

    from app.mcp.server import mcp_server
    sm = mcp_server.session_manager
    sm._has_started = False
    try:
        async with LifespanManager(app):
            async with AsyncClient(
                transport=ASGITransport(app=app),
                base_url="http://test",
            ) as ac:
                yield ac
    finally:
        _db.async_session_factory = original_factory
        app.dependency_overrides.clear()


@pytest_asyncio.fixture
async def admin_user(db_session: AsyncSession):
    """Create a global admin user and return (user, access_token).

    Attaches BOTH ``global_admin`` and ``employee`` roles to mirror
    production HR sync (``HrService._grant_default_employee_role``
    auto-attaches ``employee`` to every active person). Tests for the
    ``X-View-Role`` self-downgrade rely on the admin actually holding
    ``employee`` so the narrowing path matches the production carve-out
    that refuses to grant ``employee`` via header to a person who
    doesn't already hold it.
    """
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="admin@test.com",
        display_name="Admin User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-admin-001",
    )
    db_session.add(user)
    await db_session.flush()

    admin_role_id = await _get_role_id(db_session, "global_admin")
    employee_role_id = await _get_role_id(db_session, "employee")

    db_session.add(AuthUserRole(user_id=user_id, role_id=admin_role_id))
    db_session.add(AuthUserRole(user_id=user_id, role_id=employee_role_id))
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["global_admin"])
    return user, token


@pytest_asyncio.fixture
async def employee_user(db_session: AsyncSession):
    """Create an employee user and return (user, access_token)."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="employee@test.com",
        display_name="Employee User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-emp-001",
    )
    db_session.add(user)
    await db_session.flush()

    role_id = await _get_role_id(db_session, "employee")

    user_role = AuthUserRole(user_id=user_id, role_id=role_id)
    db_session.add(user_role)
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee"])
    return user, token


@pytest_asyncio.fixture
async def manager_and_subordinate(db_session: AsyncSession):
    """Create a manager with a subordinate, return (manager, sub, manager_token)."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    manager_id = uuid.uuid4()
    manager = IdentityPerson(
        id=manager_id,
        work_email="manager@test.com",
        display_name="Manager User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-mgr-001",
    )
    db_session.add(manager)
    await db_session.flush()

    role_id = await _get_role_id(db_session, "employee")
    db_session.add(AuthUserRole(user_id=manager_id, role_id=role_id))

    sub_id = uuid.uuid4()
    subordinate = IdentityPerson(
        id=sub_id,
        work_email="subordinate@test.com",
        display_name="Subordinate User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-sub-001",
        manager_person_id=manager_id,
    )
    db_session.add(subordinate)

    db_session.add(AuthUserRole(user_id=sub_id, role_id=role_id))
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(manager_id), ["employee"])
    return manager, subordinate, token


@pytest_asyncio.fixture
async def auditor_user(db_session: AsyncSession):
    """Create an auditor user and return (user, access_token)."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="auditor@test.com",
        display_name="Auditor User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-aud-001",
    )
    db_session.add(user)
    await db_session.flush()

    # Assign both employee and auditor roles
    emp_role_id = await _get_role_id(db_session, "employee")
    db_session.add(AuthUserRole(user_id=user_id, role_id=emp_role_id))

    aud_role_id = await _get_role_id(db_session, "auditor")
    db_session.add(AuthUserRole(user_id=user_id, role_id=aud_role_id))

    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee", "auditor"])
    return user, token


@pytest_asyncio.fixture
async def asset_admin_user(db_session: AsyncSession):
    """Create an asset admin user and return (user, access_token).

    ``asset_admin`` is a region-scoped role: callers must administer at
    least one managed location to perform writes (the region gates in
    ``app.domains.asset.region_gates``). Since migration 535 that scope
    comes from ``asset_location_admins``, not from the office. The
    fixture therefore:

    1. Creates a managed location ``TestCity`` and lists the user as its
       administrator, so the standard scenarios (create asset / claim /
       return / …) work without each test bootstrapping region
       membership.
    2. Sets ``work_location_name`` / ``city`` to the same name, so the
       target-person gate resolves the user's own region for self-claim.
    3. Also creates a ``TestCity`` office with the user as its 当地
       Admin — that list grants nothing, but the office-held and
       recycle-to-office scenarios need a real office row to point at.

    Tests exercising the "asset_admin with no region" rejection path
    should NOT use this fixture — they should build a bare user.
    """
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.domains.asset.models import (
        AssetLocation,
        AssetLocationAdmin,
        AssetOfficeAddress,
        AssetOfficeLocalAdmin,
    )
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="asset-admin@test.com",
        display_name="Asset Admin User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-asset-admin-001",
        work_location_name="TestCity",
        city="TestCity",
    )
    db_session.add(user)
    await db_session.flush()

    employee_role_id = await _get_role_id(db_session, "employee")
    asset_admin_role_id = await _get_role_id(db_session, "asset_admin")

    db_session.add(AuthUserRole(user_id=user_id, role_id=employee_role_id))
    db_session.add(AuthUserRole(user_id=user_id, role_id=asset_admin_role_id))

    # Managed location + admin row so region-scoped writes succeed.
    location_id = uuid.uuid4()
    db_session.add(AssetLocation(
        id=location_id,
        code="TESTCITY",
        name="TestCity",
        sort_order=999,
    ))
    await db_session.flush()
    db_session.add(AssetLocationAdmin(
        location_id=location_id,
        person_id=user_id,
        is_primary=True,
    ))

    # Office row for the office-held / recycle-to-office scenarios. Its
    # 当地 Admin list grants no authority — the location row above is
    # what makes this user a regional administrator.
    office_id = uuid.uuid4()
    db_session.add(AssetOfficeAddress(
        id=office_id,
        city="TestCity",
        office_name="Test Office",
    ))
    await db_session.flush()
    db_session.add(AssetOfficeLocalAdmin(
        office_id=office_id,
        person_id=user_id,
        is_primary=True,
    ))
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee", "asset_admin"])
    return user, token


@pytest_asyncio.fixture
async def finance_user(db_session: AsyncSession):
    """Create a finance user and return (user, access_token)."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="finance@test.com",
        display_name="Finance User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-finance-001",
    )
    db_session.add(user)
    await db_session.flush()

    employee_role_id = await _get_role_id(db_session, "employee")
    finance_role_id = await _get_role_id(db_session, "finance")

    db_session.add(AuthUserRole(user_id=user_id, role_id=employee_role_id))
    db_session.add(AuthUserRole(user_id=user_id, role_id=finance_role_id))
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee", "finance"])
    return user, token


@pytest_asyncio.fixture
async def hr_user(db_session: AsyncSession):
    """Create an HR user and return (user, access_token)."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="hr@test.com",
        display_name="HR User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-hr-001",
    )
    db_session.add(user)
    await db_session.flush()

    employee_role_id = await _get_role_id(db_session, "employee")
    hr_role_id = await _get_role_id(db_session, "hr")

    db_session.add(AuthUserRole(user_id=user_id, role_id=employee_role_id))
    db_session.add(AuthUserRole(user_id=user_id, role_id=hr_role_id))
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee", "hr"])
    return user, token


@pytest_asyncio.fixture
async def github_admin_user(db_session: AsyncSession):
    """GitHub 组织管理员 —— 域内读写,域外一无所有。

    ``github_admin`` 的价值恰恰在于它有边界:测试既要证明它能改
    GitHub 组织登记,也要证明它进不了别的管理面。所以这个 fixture
    只授予 ``employee`` + ``github_admin``,不搭任何别的角色。
    """
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="github-admin@test.com",
        display_name="GitHub Admin User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-gha-001",
    )
    db_session.add(user)
    await db_session.flush()

    for role_name in ("employee", "github_admin"):
        role_id = await _get_role_id(db_session, role_name)
        db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))

    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee", "github_admin"])
    return user, token


@pytest_asyncio.fixture
async def cloud_admin_user(db_session: AsyncSession):
    """Create a cloud_admin user and return (user, access_token)."""
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="cloud-admin@test.com",
        display_name="Cloud Admin User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-cloud-admin-001",
    )
    db_session.add(user)
    await db_session.flush()

    employee_role_id = await _get_role_id(db_session, "employee")
    cloud_admin_role_id = await _get_role_id(db_session, "cloud_admin")

    db_session.add(AuthUserRole(user_id=user_id, role_id=employee_role_id))
    db_session.add(AuthUserRole(user_id=user_id, role_id=cloud_admin_role_id))
    await db_session.flush()
    await db_session.commit()

    token = create_access_token(str(user_id), ["employee", "cloud_admin"])
    return user, token


@pytest_asyncio.fixture
async def legacy_cloud_admin_token_user(db_session: AsyncSession):
    """Create a user with ``employee`` DB role + a token that ALSO
    claims ``cloud_admin``.

    Despite the historical fixture name, the property under test here
    is the broader "JWT claims do NOT elevate" invariant: the request
    pipeline derives effective roles from the DB (auth_user_roles),
    not from token claims. Migration 080 (2026-04-29) brought the
    cloud_admin role row back as a perpetual-grant marker, so the row
    DOES exist now in ``auth_roles`` — but this user is not assigned
    that DB role, so the token claim alone must not carry asset-admin
    or any other authority.
    """
    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="cloud-admin@test.com",
        display_name="Cloud Admin User (legacy)",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-cloud-admin-001",
    )
    db_session.add(user)
    await db_session.flush()

    employee_role_id = await _get_role_id(db_session, "employee")
    db_session.add(AuthUserRole(user_id=user_id, role_id=employee_role_id))
    await db_session.flush()
    await db_session.commit()

    # Token carries the legacy claim so the alias path has something
    # to upgrade. No corresponding DB row.
    token = create_access_token(str(user_id), ["employee", "cloud_admin"])
    return user, token


@pytest.fixture
def admin_headers(admin_user):
    """Helper fixture for admin authorization headers."""
    _, token = admin_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def employee_headers(employee_user):
    """Helper fixture for employee authorization headers."""
    _, token = employee_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def auditor_headers(auditor_user):
    """Helper fixture for auditor authorization headers."""
    _, token = auditor_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def asset_admin_headers(asset_admin_user):
    """Helper fixture for asset_admin authorization headers."""
    _, token = asset_admin_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def finance_headers(finance_user):
    """Helper fixture for finance authorization headers."""
    _, token = finance_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def hr_headers(hr_user):
    """Helper fixture for hr authorization headers."""
    _, token = hr_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def github_admin_headers(github_admin_user):
    """Helper fixture for github_admin authorization headers."""
    _, token = github_admin_user
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def cloud_admin_headers(cloud_admin_user):
    """Helper fixture for cloud_admin authorization headers."""
    _, token = cloud_admin_user
    return {"Authorization": f"Bearer {token}"}