"""HTTP surface of the approval-flow registry: RBAC, CRUD, the
published all-staff read, the ownership trail, and the import."""
from __future__ import annotations

import uuid

import pytest

from app.identity.person_models import IdentityPerson

BASE = "/api/v1/approval-flows"
SELF = "/api/v1/me/approval-flows"


async def _person(
    db_session, email: str, *, name: str = "Owner", status: str = "active",
):
    person = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=name,
        employment_status=status, source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db_session.add(person)
    await db_session.flush()
    await db_session.commit()
    return person


async def _create(client, headers, **overrides):
    payload = {
        "approval_code": f"AAAA{uuid.uuid4().hex[:8].upper()}",
        "name": "差旅报销申请",
        "category": "Finance",
    }
    payload.update(overrides)
    resp = await client.post(BASE, json=payload, headers=headers)
    assert resp.status_code == 201, resp.text
    return resp.json()["data"]


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


@pytest.mark.asyncio
async def test_meta_exposes_vocabulary_and_coverage(client, admin_headers):
    resp = await client.get(f"{BASE}/meta", headers=admin_headers)
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert "active" in data["statuses"]
    assert "confirmed" in data["owner_sources"]
    assert data["inference"]["min_tasks"] >= 1
    assert "coverage_pct" in data["summary"]


@pytest.mark.asyncio
async def test_employee_cannot_read_the_admin_surface(client, employee_headers):
    assert (
        await client.get(f"{BASE}/meta", headers=employee_headers)
    ).status_code == 403
    assert (
        await client.get(BASE, headers=employee_headers)
    ).status_code == 403


@pytest.mark.asyncio
async def test_employee_can_read_the_published_list(
    client, admin_headers, employee_headers,
):
    """The whole point of the register is that anyone can find out which
    approval to use and who owns it."""
    await _create(client, admin_headers, name="全员公告申请")
    resp = await client.get(f"{BASE}/published", headers=employee_headers)
    assert resp.status_code == 200
    data = resp.json()["data"]
    names = [f["name"] for g in data["groups"] for f in g["flows"]]
    assert "全员公告申请" in names


@pytest.mark.asyncio
async def test_auditor_reads_but_cannot_write(client, admin_headers, auditor_headers):
    flow = await _create(client, admin_headers)
    assert (await client.get(BASE, headers=auditor_headers)).status_code == 200
    resp = await client.patch(
        f"{BASE}/{flow['id']}", json={"name": "改名"}, headers=auditor_headers,
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_hr_can_read_the_registry(client, admin_headers, hr_headers):
    await _create(client, admin_headers)
    assert (await client.get(BASE, headers=hr_headers)).status_code == 200


# ── CRUD ────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_code_is_upper_cased_and_unique(client, admin_headers):
    code = f"bbbb{uuid.uuid4().hex[:8]}"
    flow = await _create(client, admin_headers, approval_code=code)
    assert flow["approval_code"] == code.upper()
    # The same code in another case must not register a second row —
    # two rows would mean two owners and two confirmation cards for one
    # flow.
    dup = await client.post(
        BASE,
        json={"approval_code": code.upper(), "name": "又一个"},
        headers=admin_headers,
    )
    assert dup.status_code == 409


@pytest.mark.asyncio
async def test_patch_clears_category_only_when_explicitly_null(
    client, admin_headers,
):
    flow = await _create(client, admin_headers, category="Finance")
    # Key absent → untouched.
    resp = await client.patch(
        f"{BASE}/{flow['id']}", json={"description": "报销用"},
        headers=admin_headers,
    )
    assert resp.json()["data"]["category"] == "Finance"
    # Explicit null → cleared.
    resp = await client.patch(
        f"{BASE}/{flow['id']}", json={"category": None}, headers=admin_headers,
    )
    assert resp.json()["data"]["category"] is None


@pytest.mark.asyncio
async def test_update_cannot_change_ownership(client, admin_headers, db_session):
    """``PATCH`` has no ownership field at all, so there is no path that
    changes accountability without leaving an event."""
    owner = await _person(db_session, f"o-{uuid.uuid4().hex[:6]}@test.com")
    flow = await _create(client, admin_headers)
    resp = await client.patch(
        f"{BASE}/{flow['id']}",
        json={"owner_person_id": str(owner.id), "name": "新名字"},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["owner"] is None


@pytest.mark.asyncio
async def test_retire_removes_it_from_the_published_list(
    client, admin_headers, employee_headers,
):
    flow = await _create(client, admin_headers, name="已下线的流程")
    await client.post(
        f"{BASE}/{flow['id']}/retire", json={"note": "不再使用"},
        headers=admin_headers,
    )
    resp = await client.get(f"{BASE}/published", headers=employee_headers)
    names = [f["name"] for g in resp.json()["data"]["groups"] for f in g["flows"]]
    assert "已下线的流程" not in names


@pytest.mark.asyncio
async def test_unpublished_flow_is_hidden_from_staff_but_not_from_it(
    client, admin_headers, employee_headers,
):
    flow = await _create(client, admin_headers, name="内部专用流程")
    await client.patch(
        f"{BASE}/{flow['id']}", json={"published": False}, headers=admin_headers,
    )
    staff = await client.get(f"{BASE}/published", headers=employee_headers)
    names = [f["name"] for g in staff.json()["data"]["groups"] for f in g["flows"]]
    assert "内部专用流程" not in names
    admin = await client.get(
        f"{BASE}?q=内部专用流程", headers=admin_headers,
    )
    assert admin.json()["data"][0]["name"] == "内部专用流程"


# ── Ownership ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_owner_assignment_writes_an_event(client, admin_headers, db_session):
    owner = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@test.com")
    flow = await _create(client, admin_headers)
    resp = await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(owner.id)}, headers=admin_headers,
    )
    assert resp.status_code == 200
    body = resp.json()["data"]
    assert body["owner"]["id"] == str(owner.id)
    assert body["owner_source"] == "manual"
    # Assigned but not confirmed by the person themselves.
    assert body["owner_state"] == "unconfirmed"

    events = (
        await client.get(f"{BASE}/{flow['id']}/events", headers=admin_headers)
    ).json()["data"]
    assert any(e["event_type"] == "owner_assigned" for e in events)


@pytest.mark.asyncio
async def test_transfer_records_both_sides(client, admin_headers, db_session):
    first = await _person(db_session, f"f-{uuid.uuid4().hex[:6]}@test.com", name="甲")
    second = await _person(db_session, f"s-{uuid.uuid4().hex[:6]}@test.com", name="乙")
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(first.id)}, headers=admin_headers,
    )
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(second.id)}, headers=admin_headers,
    )
    events = (
        await client.get(f"{BASE}/{flow['id']}/events", headers=admin_headers)
    ).json()["data"]
    transfer = next(e for e in events if e["event_type"] == "owner_transferred")
    assert transfer["meta"]["from_name"] == "甲"
    assert transfer["meta"]["to_name"] == "乙"


@pytest.mark.asyncio
async def test_cannot_name_a_departed_person_as_owner(
    client, admin_headers, db_session,
):
    """A flow owned by someone who has left satisfies every "has an
    owner" check while nobody is actually answering for it."""
    # ``leaver_pending``, not ``left``: the LeaverBlocked before-flush
    # listener refuses a direct flip to ``left``. Both are non-active, so
    # the guard under test sees the same thing either way.
    gone = await _person(
        db_session, f"g-{uuid.uuid4().hex[:6]}@test.com",
        status="leaver_pending",
    )
    flow = await _create(client, admin_headers)
    resp = await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(gone.id)}, headers=admin_headers,
    )
    assert resp.status_code == 409
    assert resp.json()["errors"][0]["code"] == "APPROVAL_FLOW_OWNER_NOT_ACTIVE"


@pytest.mark.asyncio
async def test_ownerless_filter_matches_the_coverage_tile(
    client, admin_headers, db_session,
):
    """Clicking the KPI number must show exactly the rows it counted."""
    owner = await _person(db_session, f"c-{uuid.uuid4().hex[:6]}@test.com")
    owned = await _create(client, admin_headers, name="有主的")
    await client.post(
        f"{BASE}/{owned['id']}/owner",
        json={"owner_person_id": str(owner.id)}, headers=admin_headers,
    )
    await _create(client, admin_headers, name="没主的")

    summary = (
        await client.get(f"{BASE}/meta", headers=admin_headers)
    ).json()["data"]["summary"]
    listed = (
        await client.get(f"{BASE}?ownerless=true&page_size=100", headers=admin_headers)
    ).json()
    assert listed["meta"]["total"] == summary["ownerless"] + summary["owner_departed"]


@pytest.mark.asyncio
async def test_departed_owner_counts_as_a_gap_not_as_coverage(
    client, admin_headers, db_session,
):
    person = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@test.com")
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(person.id)}, headers=admin_headers,
    )
    person.employment_status = "leaver_pending"
    db_session.add(person)
    await db_session.commit()

    summary = (
        await client.get(f"{BASE}/meta", headers=admin_headers)
    ).json()["data"]["summary"]
    assert summary["owner_departed"] >= 1

    row = (
        await client.get(f"{BASE}/{flow['id']}", headers=admin_headers)
    ).json()["data"]
    assert row["owner_state"] == "departed"


# ── Self-service ────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_owner_confirms_their_own_flow(
    client, admin_headers, employee_headers, employee_user,
):
    person, _token = employee_user
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(person.id)}, headers=admin_headers,
    )
    mine = (await client.get(SELF, headers=employee_headers)).json()["data"]
    assert mine["total"] == 1
    assert mine["needs_confirmation"] == 1

    resp = await client.post(
        f"{SELF}/{flow['id']}/confirm", json={}, headers=employee_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["owner_source"] == "confirmed"
    assert resp.json()["data"]["owner_state"] == "confirmed"

    mine = (await client.get(SELF, headers=employee_headers)).json()["data"]
    assert mine["needs_confirmation"] == 0


@pytest.mark.asyncio
async def test_confirming_someone_elses_flow_is_404_not_403(
    client, admin_headers, employee_headers, db_session,
):
    """A 403 would confirm the row exists and that somebody else owns
    it — information this endpoint has no reason to hand out."""
    other = await _person(db_session, f"x-{uuid.uuid4().hex[:6]}@test.com")
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(other.id)}, headers=admin_headers,
    )
    resp = await client.post(
        f"{SELF}/{flow['id']}/confirm", json={}, headers=employee_headers,
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_the_owner_can_write_what_their_flow_is_for(
    client, admin_headers, employee_headers, employee_user,
):
    """All 49 flows in this tenant shipped with an empty purpose, so the
    register could name a person to ask and never say what the thing was
    for. IT cannot credibly write 49 of these; the owner can write one."""
    person, _token = employee_user
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(person.id)}, headers=admin_headers,
    )

    resp = await client.patch(
        f"{SELF}/{flow['id']}/description",
        json={"description": "出差前订机票酒店走这条"},
        headers=employee_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["description"] == "出差前订机票酒店走这条"

    # It reaches the ALL-STAFF page, which is the entire point.
    pub = (await client.get(f"{BASE}/published", headers=employee_headers)).json()
    rows = [f for g in pub["data"]["groups"] for f in g["flows"]
            if f["id"] == flow["id"]]
    assert rows and rows[0]["description"] == "出差前订机票酒店走这条"

    # And it is on the trail with a name against it — 「这句话是谁写的」
    # stays answerable.
    events = (await client.get(
        f"{BASE}/{flow['id']}/events", headers=admin_headers,
    )).json()["data"]
    assert any(e["event_type"] == "updated"
               and (e["meta"] or {}).get("field") == "description"
               for e in events)

    # Empty clears it: a purpose that turned out wrong should be
    # removable without asking IT.
    resp = await client.patch(
        f"{SELF}/{flow['id']}/description", json={"description": ""},
        headers=employee_headers,
    )
    assert resp.status_code == 200
    assert not resp.json()["data"]["description"]


@pytest.mark.asyncio
async def test_describing_someone_elses_flow_is_404_not_403(
    client, admin_headers, employee_headers, db_session,
):
    """Same gate as every other verb on this router. A description is
    published to the whole company, so 「谁能写」 is a real question."""
    other = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@test.com")
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(other.id)}, headers=admin_headers,
    )
    resp = await client.patch(
        f"{SELF}/{flow['id']}/description", json={"description": "我乱写的"},
        headers=employee_headers,
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_dispute_records_but_does_not_clear_the_owner(
    client, admin_headers, employee_headers, employee_user,
):
    person, _token = employee_user
    flow = await _create(client, admin_headers)
    await client.post(
        f"{BASE}/{flow['id']}/owner",
        json={"owner_person_id": str(person.id)}, headers=admin_headers,
    )
    resp = await client.post(
        f"{SELF}/{flow['id']}/dispute", json={"reason": "这是财务在管"},
        headers=employee_headers,
    )
    assert resp.status_code == 200

    row = (
        await client.get(f"{BASE}/{flow['id']}", headers=admin_headers)
    ).json()["data"]
    # Still owned: letting one person's say-so blank the register would
    # make the contested flow vanish from the coverage read.
    assert row["owner"]["id"] == str(person.id)
    events = (
        await client.get(f"{BASE}/{flow['id']}/events", headers=admin_headers)
    ).json()["data"]
    assert any(e["event_type"] == "owner_disputed" for e in events)


# ── Import ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_import_is_partial_and_reports_every_rejection(
    client, admin_headers, db_session,
):
    owner = await _person(db_session, f"i-{uuid.uuid4().hex[:6]}@test.com")
    good_code = f"CCCC{uuid.uuid4().hex[:8].upper()}"
    resp = await client.post(
        f"{BASE}/import",
        json={
            "rows": [
                {
                    "approval_code": good_code, "name": "入职申请",
                    "category": "HR", "owner_email": owner.work_email,
                },
                {
                    "approval_code": f"DDDD{uuid.uuid4().hex[:8].upper()}",
                    "name": "没有这个人", "owner_email": "nobody@test.com",
                },
            ],
        },
        headers=admin_headers,
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["created"] == 1
    assert data["failed"] == 1
    assert data["errors"][0]["row"] == 2
    assert "nobody@test.com" in data["errors"][0]["error"]

    listed = (
        await client.get(f"{BASE}?q={good_code}", headers=admin_headers)
    ).json()["data"]
    assert listed[0]["owner"]["id"] == str(owner.id)
    assert listed[0]["owner_source"] == "imported"


@pytest.mark.asyncio
async def test_export_is_readable_by_the_governance_lane(client, admin_headers):
    await _create(client, admin_headers)
    resp = await client.get(f"{BASE}/export", headers=admin_headers)
    assert resp.status_code == 200
    assert "spreadsheetml" in resp.headers["content-type"]