"""Onboarding cockpit queue endpoint — Day-1 readiness aggregation.

Covers:
  * RBAC — employee 403, auditor 200, admin 200
  * window: every OPEN process with hire_date <= today+14d — future
    hires AND already-arrived/incomplete ones (no backward cutoff);
    only > +14d future is excluded
  * inactive + soft-deleted excluded
  * each of the 7 readiness dimensions flips correctly
  * Feishu is a global signal (target health)
  * sorting (hire_date asc, least-ready first) + fully_ready + KPIs
  * regression: a hire who already started but isn't fully provisioned
    stays on the board and counts in needs_attention (pre-2026-06-29 the
    hire_date>=today + status-excludes-live query silently dropped them)
"""
import uuid
from datetime import datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.asset.models import (
    AssetItem,
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.identity_sync.models import (
    IdentitySyncState,
    IdentitySyncTarget,
)
from app.domains.vpn_application.models import (
    VPN_APP_STATUS_APPROVED,
    VPN_APP_STATUS_PENDING,
    VpnApplication,
)
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityPerson,
)

QUEUE_URL = "/api/v1/onboarding/cockpit/queue"


def _today():
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


async def _mk_person(
    db: AsyncSession,
    *,
    hire_offset_days: int,
    work_email: str,
    status: str = "active",
    deleted: bool = False,
):
    pid = uuid.uuid4()
    p = IdentityPerson(
        id=pid,
        work_email=work_email,
        display_name=work_email.split("@")[0],
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{pid.hex[:8]}",
        hire_date=_today() + timedelta(days=hire_offset_days),
        job_title="Engineer",
        company_name="PingCAP",
        deleted_at=datetime.now(ZoneInfo("UTC")) if deleted else None,
    )
    db.add(p)
    await db.flush()
    # 2026-06-01 PR-1: the cockpit now reads from onboarding_processes
    # + onboarding_blockers (instead of recomputing 7 dimensions per
    # request), so every test that wants a person to appear in the
    # queue must also have a matching process row. Skip seeding the
    # process for deleted / non-active people — those are explicitly
    # excluded from the cockpit window upstream.
    if not deleted and status == "active":
        from app.domains.onboarding.service import OnboardingService

        await OnboardingService(db).start_process(
            pid,
            p.hire_date,
            trace_id=f"test:{pid.hex[:8]}",
        )
    return p


async def _give_sso(db, person_id, *, is_login=True, provider="google"):
    """Seed an external identity. ``is_login=True`` (default) mints a real
    LOGIN account (google/feishu) — the only kind that greens 账号/SSO/邮箱.
    Pass ``is_login=False`` to simulate the successfactors HRIS-source row
    (directory record, NOT a login) which must NOT green those dims."""
    db.add(
        IdentityExternalIdentity(
            id=uuid.uuid4(),
            person_id=person_id,
            provider=provider,
            provider_subject=f"sub-{uuid.uuid4().hex[:8]}",
            identity_status="active",
            is_login_identity=is_login,
        )
    )
    await db.flush()


async def _give_seat(db, person_id):
    cat = AssetSoftwareCatalog(
        id=uuid.uuid4(),
        name=f"Tool-{uuid.uuid4().hex[:6]}",
        offboarding_policy="revoke",
        category="collaboration",
    )
    db.add(cat)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(),
        software_id=cat.id,
        name="Standard",
        price=Decimal("10.00"),
        billing_cycle="monthly",
        over_quota_policy="block",
        currency="USD",
    )
    db.add(plan)
    await db.flush()
    db.add(
        SoftwareSeatAllocation(
            id=uuid.uuid4(),
            plan_id=plan.id,
            user_id=person_id,
            status="active",
        )
    )
    await db.flush()


async def _give_computer(db, person_id):
    db.add(
        AssetItem(
            id=uuid.uuid4(),
            asset_tag=f"PC-{uuid.uuid4().hex[:8]}",
            category="laptop",
            status="claimed",  # was "in_use" — never a valid asset status (2026-08-07)
            assigned_to=person_id,
        )
    )
    await db.flush()


async def _give_vpn(db, person_id, *, status=VPN_APP_STATUS_APPROVED):
    db.add(
        VpnApplication(
            id=str(uuid.uuid4()),
            person_id=str(person_id),
            vpn_node="cn-north",
            usage_description="dev",
            resource_checklist="db",
            status=status,
        )
    )
    await db.flush()


async def _set_feishu_target(db, *, status="ok", enabled=True):
    t = IdentitySyncTarget(
        id=uuid.uuid4(),
        name="Feishu Contact",
        slug="feishu-contact",
        kind="feishu_contact",
        enabled=enabled,
        on_offboard="deactivate",
        mapping_config={},
        filter_config={},
        schedule_config={},
        last_run_status=status,
    )
    db.add(t)
    await db.flush()
    return t


async def _give_feishu_sync(db, person_id, *, status="success", target=None):
    """Seed a per-person identity_sync_state row for the feishu_contact
    target — i.e. record that this hire actually landed in Feishu. Creates
    a healthy target on the fly when one isn't supplied."""
    if target is None:
        target = await _set_feishu_target(db, status="ok", enabled=True)
    db.add(
        IdentitySyncState(
            id=uuid.uuid4(),
            target_id=target.id,
            person_id=person_id,
            last_status=status,
        )
    )
    await db.flush()
    return target


def _hdr(token):
    return {"Authorization": f"Bearer {token}"}


async def _refresh_blockers(db):
    """Reconcile every active process + commit — re-scan dimensions so
    blockers reflect the resource grants seeded by ``_give_*`` helpers.

    The PR-1 cockpit reads from ``onboarding_processes`` +
    ``onboarding_blockers`` (no per-request recompute), so tests that
    grant resources AFTER ``_mk_person`` must refresh + commit before
    querying. Mirrors the 5-min ``onboarding.reconcile_processes`` cron
    in prod.
    """
    from app.domains.onboarding.service import OnboardingService

    await OnboardingService(db).reconcile_all_active_processes()
    await db.commit()


# ── RBAC ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_employee_forbidden(client, employee_user):
    _, token = employee_user
    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_auditor_allowed(client, auditor_user):
    _, token = auditor_user
    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    assert resp.status_code == 200


@pytest.mark.asyncio
async def test_admin_empty_window(client, admin_user):
    """No upcoming hires → empty rows + zeroed KPIs."""
    _, token = admin_user
    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["rows"] == []
    assert data["kpis"]["total"] == 0
    assert data["kpis"]["avg_readiness_pct"] == 0


# ── Window boundaries ─────────────────────────────────────────────


@pytest.mark.asyncio
async def test_window_boundaries(client, admin_user, db_session):
    """Forward bound only: today / +14 in, +15 out. Yesterday's hire
    (already arrived, still onboarding) is now IN — there is no backward
    cutoff, so an arrived-but-incomplete process never drops off."""
    _, token = admin_user
    await _mk_person(db_session, hire_offset_days=0, work_email="d0@x.com")
    await _mk_person(db_session, hire_offset_days=14, work_email="d14@x.com")
    await _mk_person(db_session, hire_offset_days=15, work_email="d15@x.com")
    await _mk_person(db_session, hire_offset_days=-1, work_email="dm1@x.com")
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    emails = {r["work_email"] for r in resp.json()["data"]["rows"]}
    assert emails == {"d0@x.com", "d14@x.com", "dm1@x.com"}


@pytest.mark.asyncio
async def test_arrived_incomplete_visible_and_counted(
    client, admin_user, db_session
):
    """Regression: a hire whose start date already passed but who is NOT
    fully provisioned (status ``preparing``, past-due) must stay on the
    cockpit and count in ``needs_attention``.

    Pre-2026-06-29 the queue filtered ``hire_date >= today`` AND status
    in (scanning/preparing/ready) only — so this person disappeared and
    the KPI strip read 0 while the analytics scorecard still flagged the
    open hard-gap. That contradiction is the bug this locks down."""
    _, token = admin_user
    # bare hire, started 2 days ago → pending blockers, never promoted to
    # ``live`` (the scanner only goes live when blockers clear), so it
    # sits ``preparing`` past its hire_date.
    await _mk_person(db_session, hire_offset_days=-2, work_email="late@x.com")
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    data = resp.json()["data"]
    row = next(
        (r for r in data["rows"] if r["work_email"] == "late@x.com"), None
    )
    assert row is not None, "arrived-but-incomplete hire must be visible"
    assert row["status"] == "preparing"
    assert row["days_to_hire"] == -2
    assert row["fully_ready"] is False
    assert data["kpis"]["total"] >= 1
    assert data["kpis"]["needs_attention"] >= 1


@pytest.mark.asyncio
async def test_live_fully_provisioned_arrival_visible(
    client, admin_user, db_session
):
    """A hire who arrived and got fully provisioned flips to ``live`` and
    stays on the board (status ``live`` is no longer excluded)."""
    _, token = admin_user
    p = await _mk_person(
        db_session, hire_offset_days=-1, work_email="livefull@x.com",
    )
    await _give_sso(db_session, p.id)
    await _give_seat(db_session, p.id)
    await _give_computer(db_session, p.id)
    await _give_vpn(db_session, p.id)
    await _give_feishu_sync(db_session, p.id, status="success")
    await db_session.commit()
    await _refresh_blockers(db_session)

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    data = resp.json()["data"]
    row = next(
        (r for r in data["rows"] if r["work_email"] == "livefull@x.com"),
        None,
    )
    assert row is not None, "a live (arrived) process must be visible"
    assert row["status"] == "live"
    assert row["fully_ready"] is True
    assert data["kpis"]["fully_ready"] >= 1


@pytest.mark.asyncio
async def test_departed_after_process_started_excluded(
    client, admin_user, db_session
):
    """Regression for the backward-unbounded window: a person who had an
    OPEN onboarding process and THEN left (or was soft-deleted) must not
    surface on the cockpit. Nothing cancels an onboarding process on
    offboard, so without an explicit guard the leaver would render with a
    broken ``account`` dim and inflate ``needs_attention``."""
    from sqlalchemy import update

    _, token = admin_user
    leaver = await _mk_person(
        db_session, hire_offset_days=-3, work_email="became_leaver@x.com",
    )
    deleted = await _mk_person(
        db_session, hire_offset_days=-3, work_email="became_deleted@x.com",
    )
    await _mk_person(
        db_session, hire_offset_days=-3, work_email="still_here@x.com",
    )
    # Flip status AFTER the process exists (the helper created it while
    # they were active). Use a Core UPDATE so we exercise ONLY the cockpit
    # query filter — going through the ORM attribute set would fire the
    # Leaver-Blocked before_flush listener, which is a different subsystem.
    await db_session.execute(
        update(IdentityPerson)
        .where(IdentityPerson.id == leaver.id)
        .values(employment_status="left")
    )
    await db_session.execute(
        update(IdentityPerson)
        .where(IdentityPerson.id == deleted.id)
        .values(deleted_at=datetime.now(ZoneInfo("UTC")))
    )
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    emails = {r["work_email"] for r in resp.json()["data"]["rows"]}
    assert "became_leaver@x.com" not in emails
    assert "became_deleted@x.com" not in emails
    assert "still_here@x.com" in emails


@pytest.mark.asyncio
async def test_inactive_and_deleted_excluded(client, admin_user, db_session):
    _, token = admin_user
    await _mk_person(
        db_session, hire_offset_days=3, work_email="active@x.com",
    )
    await _mk_person(
        db_session, hire_offset_days=3, work_email="leaver@x.com",
        status="left",
    )
    await _mk_person(
        db_session, hire_offset_days=3, work_email="gone@x.com", deleted=True,
    )
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    emails = {r["work_email"] for r in resp.json()["data"]["rows"]}
    assert emails == {"active@x.com"}


# ── Dimensions ────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_baseline_dimensions(client, admin_user, db_session):
    """A bare active hire with only the reserved SF address and no login
    account: account / sso NOT ready (no account minted), work_email in
    the RESERVED middle state (planned address, no live mailbox). The
    three policy/checklist dims are vacuously ready.

    Regression for the P10002 false-green bug: pre-2026-07-06 account +
    work_email greened off employment_status + the reserved address, so a
    hire with no Google account read as ready."""
    _, token = admin_user
    await _mk_person(db_session, hire_offset_days=2, work_email="bare@x.com")
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    row = next(
        r for r in resp.json()["data"]["rows"]
        if r["work_email"] == "bare@x.com"
    )
    dims = row["dimensions"]
    states = row["dimension_states"]
    # No login identity → account/sso not ready, work_email is RESERVED.
    assert dims["account"] is False
    assert states["account"] == "pending"
    assert dims["sso"] is False
    assert states["sso"] == "pending"
    assert dims["work_email"] is False
    assert states["work_email"] == "reserved"
    assert dims["seat"] is False
    assert dims["computer"] is False
    assert dims["vpn"] is False
    assert dims["feishu"] is False
    # The three policy/checklist dims are vacuously ready (no policies
    # / no ack items in this fixture) → 3 of 10.
    assert dims["email_group"] is True
    assert dims["ai_tool"] is True
    assert dims["training"] is True
    assert row["ready_count"] == 3
    assert row["fully_ready"] is False


@pytest.mark.asyncio
async def test_all_dimensions_ready(client, admin_user, db_session):
    _, token = admin_user
    p = await _mk_person(
        db_session, hire_offset_days=5, work_email="ready@x.com",
    )
    await _give_sso(db_session, p.id)
    await _give_seat(db_session, p.id)
    await _give_computer(db_session, p.id)
    await _give_vpn(db_session, p.id)
    await _give_feishu_sync(db_session, p.id, status="success")
    await db_session.commit()
    await _refresh_blockers(db_session)

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    row = next(
        r for r in resp.json()["data"]["rows"]
        if r["work_email"] == "ready@x.com"
    )
    assert all(row["dimensions"].values()), row["dimensions"]
    # 10 dims since 2026-06-11: email_group / ai_tool / training are
    # vacuously ready here (no policies / no ack items configured).
    assert row["ready_count"] == 10
    assert row["fully_ready"] is True


@pytest.mark.asyncio
async def test_successfactors_only_identity_not_ready(
    client, admin_user, db_session
):
    """Core P10002 regression: a hire whose ONLY external identity is the
    successfactors HRIS-source row (is_login_identity=False) must read
    account / SSO NOT ready and work_email RESERVED — that row is a
    directory record, not a login account. Before the fix its presence
    (any active identity) greened SSO, and employment_status greened
    account, so a hire with no Google account looked fully provisioned."""
    _, token = admin_user
    p = await _mk_person(
        db_session, hire_offset_days=3, work_email="sfonly@x.com",
    )
    # The HRIS-source row: active, but NOT a login identity.
    await _give_sso(
        db_session, p.id, is_login=False, provider="successfactors",
    )
    await db_session.commit()
    await _refresh_blockers(db_session)

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    row = next(
        r for r in resp.json()["data"]["rows"]
        if r["work_email"] == "sfonly@x.com"
    )
    assert row["dimensions"]["account"] is False
    assert row["dimensions"]["sso"] is False
    assert row["dimensions"]["work_email"] is False
    assert row["dimension_states"]["work_email"] == "reserved"


@pytest.mark.asyncio
async def test_login_identity_greens_account_sso_email(
    client, admin_user, db_session
):
    """A hire with an active Google LOGIN identity reads account / SSO
    ready and work_email as a live mailbox (ready). This is the positive
    side of the P10002 fix — a real account greens all three."""
    _, token = admin_user
    p = await _mk_person(
        db_session, hire_offset_days=3, work_email="haslogin@x.com",
    )
    await _give_sso(db_session, p.id, is_login=True, provider="google")
    await db_session.commit()
    await _refresh_blockers(db_session)

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    row = next(
        r for r in resp.json()["data"]["rows"]
        if r["work_email"] == "haslogin@x.com"
    )
    assert row["dimensions"]["account"] is True
    assert row["dimension_states"]["account"] == "ready"
    assert row["dimensions"]["sso"] is True
    assert row["dimensions"]["work_email"] is True
    assert row["dimension_states"]["work_email"] == "ready"


@pytest.mark.asyncio
async def test_vpn_pending_not_ready(client, admin_user, db_session):
    """A pending (un-approved) VPN application does NOT count as ready."""
    _, token = admin_user
    p = await _mk_person(db_session, hire_offset_days=4, work_email="v@x.com")
    await _give_vpn(db_session, p.id, status=VPN_APP_STATUS_PENDING)
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    row = next(
        r for r in resp.json()["data"]["rows"] if r["work_email"] == "v@x.com"
    )
    assert row["dimensions"]["vpn"] is False


@pytest.mark.asyncio
async def test_feishu_global_signal(client, admin_user, db_session):
    """The cockpit ``feishu`` meta block reports channel health (target
    enabled + last_run_status), surfaced for the inline "trigger sync"
    action. The per-row ``feishu`` dimension is separate — it follows
    per-person sync state (see test below); with no per-person rows here
    every row's dimension is False regardless of channel health."""
    _, token = admin_user
    await _mk_person(db_session, hire_offset_days=1, work_email="a@x.com")
    await _mk_person(db_session, hire_offset_days=2, work_email="b@x.com")
    await _set_feishu_target(db_session, status="failed", enabled=True)
    await db_session.commit()

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    data = resp.json()["data"]
    assert data["feishu"]["ready"] is False
    assert data["feishu"]["last_run_status"] == "failed"
    assert all(r["dimensions"]["feishu"] is False for r in data["rows"])
    # target_id is surfaced for the inline "trigger sync" action
    assert data["feishu"]["target_id"] is not None


@pytest.mark.asyncio
async def test_feishu_dimension_follows_per_person_sync(
    client, admin_user, db_session
):
    """Regression: the per-row Feishu dimension reflects whether THIS hire
    actually landed in Feishu (identity_sync_state), not channel health.
    A 'success' or 'skipped' projection row is ready; 'failed' / missing
    is not — so the advisory auto-resolves once the person is in Feishu
    instead of sitting pending while a healthy channel hasn't pushed them."""
    _, token = admin_user
    synced = await _mk_person(
        db_session, hire_offset_days=3, work_email="synced@x.com",
    )
    skipped = await _mk_person(
        db_session, hire_offset_days=3, work_email="skipped@x.com",
    )
    await _mk_person(
        db_session, hire_offset_days=3, work_email="missing@x.com",
    )
    failed = await _mk_person(
        db_session, hire_offset_days=3, work_email="failed@x.com",
    )
    target = await _set_feishu_target(db_session, status="ok", enabled=True)
    await _give_feishu_sync(
        db_session, synced.id, status="success", target=target,
    )
    await _give_feishu_sync(
        db_session, skipped.id, status="skipped", target=target,
    )
    await _give_feishu_sync(
        db_session, failed.id, status="failed", target=target,
    )
    await db_session.commit()
    await _refresh_blockers(db_session)

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    by_email = {
        r["work_email"]: r["dimensions"]["feishu"]
        for r in resp.json()["data"]["rows"]
    }
    assert by_email["synced@x.com"] is True
    assert by_email["skipped@x.com"] is True
    assert by_email["missing@x.com"] is False
    assert by_email["failed@x.com"] is False


@pytest.mark.asyncio
async def test_sorting_and_kpis(client, admin_user, db_session):
    """hire_date asc, then least-ready first within a day; KPIs tally."""
    _, token = admin_user
    # same hire day: one fully bare (2 dims), one with sso (3 dims)
    _p_bare = await _mk_person(
        db_session, hire_offset_days=1, work_email="bare1@x.com",
    )
    p_more = await _mk_person(
        db_session, hire_offset_days=1, work_email="more1@x.com",
    )
    await _give_sso(db_session, p_more.id)
    # Initial scan happened at _mk_person time; refresh so the post-
    # mk_person SSO grant lands in the blocker set the cockpit reads.
    # a later hire
    await _mk_person(db_session, hire_offset_days=8, work_email="late@x.com")
    await db_session.commit()
    await _refresh_blockers(db_session)

    resp = await client.get(QUEUE_URL, headers=_hdr(token))
    data = resp.json()["data"]
    rows = data["rows"]
    # day-1 hires come first; within day-1 the least-ready (bare) is first
    assert rows[0]["work_email"] == "bare1@x.com"
    assert rows[1]["work_email"] == "more1@x.com"
    assert rows[2]["work_email"] == "late@x.com"
    kpis = data["kpis"]
    assert kpis["total"] == 3
    assert kpis["this_week"] == 2  # the two day-1 hires
    assert kpis["fully_ready"] == 0
    assert kpis["needs_attention"] == 3
    assert kpis["dimension_ready"]["sso"] == 1
    # account now tracks a real login identity too — only p_more has one
    # (the SSO grant), so account is ready for exactly that one hire (was
    # 3 when account greened off employment_status alone).
    assert kpis["dimension_ready"]["account"] == 1