"""HTTP surface of the internal system registry: RBAC, CRUD, ownership
trail, and the reads other domains depend on."""
from __future__ import annotations

import uuid

import pytest

from app.identity.person_models import IdentityPerson

BASE = "/api/v1/systems"


async def _person(db_session, email: str, *, name: str = "Owner"):
    person = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=name,
        employment_status="active", 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 = {
        "name": "Clinic",
        "slug": f"clinic-{uuid.uuid4().hex[:6]}",
        "kind": "support",
        "exposure": "authenticated",
        "hostnames": ["clinic.pingcap.com"],
        "in_bounty_scope": True,
    }
    payload.update(overrides)
    resp = await client.post(BASE, json=payload, headers=headers)
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


@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 set(data["external_exposures"]) == {"public", "authenticated"}
    assert data["budget_tracks"] == ["product", "it"]
    assert data["can_manage"] is True
    assert "coverage_pct" in data["summary"]


@pytest.mark.asyncio
async def test_employee_cannot_read_the_registry(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_auditor_reads_but_cannot_write(client, auditor_headers):
    """The security lane consumes ownership to route a finding; it does
    not decide who owns a system. IT maintains the register."""
    assert (
        await client.get(BASE, headers=auditor_headers)
    ).status_code == 200
    resp = await client.post(
        BASE, json={"name": "Sneaky"}, headers=auditor_headers,
    )
    assert resp.status_code == 403


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


@pytest.mark.asyncio
async def test_create_normalizes_slug_and_hostnames(client, admin_headers):
    system = await _create(
        client, admin_headers,
        slug="TiDB_Cloud--Prod",
        hostnames=[
            "https://TiDBCloud.com/health?x=1", "tidbcloud.com",
            "  TIDBAPI.com:443  ", "",
        ],
    )
    assert system["slug"] == "tidb-cloud-prod"
    # Scheme / path / port / case stripped, duplicates collapsed — an
    # operator pasting a URL out of the researcher's mail must still
    # produce a hostname a later report can match.
    assert system["hostnames"] == ["tidbcloud.com", "tidbapi.com"]


@pytest.mark.asyncio
async def test_duplicate_slug_is_a_conflict(client, admin_headers):
    system = await _create(client, admin_headers)
    resp = await client.post(
        BASE, json={"name": "Other", "slug": system["slug"]},
        headers=admin_headers,
    )
    assert resp.status_code == 409
    assert resp.json()["errors"][0]["code"] == "SYSTEM_SLUG_TAKEN"


@pytest.mark.asyncio
async def test_invalid_enum_is_rejected(client, admin_headers):
    resp = await client.post(
        BASE, json={"name": "X", "slug": "x-sys", "exposure": "maybe"},
        headers=admin_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_patch_cannot_change_ownership(client, admin_headers, db_session):
    """Ownership has its own endpoint so no code path changes an owner
    without writing an event."""
    owner = await _person(db_session, f"own-{uuid.uuid4().hex[:6]}@t.com")
    system = await _create(client, admin_headers)
    resp = await client.patch(
        f"{BASE}/{system['id']}",
        json={"owner_person_id": str(owner.id)},
        headers=admin_headers,
    )
    # ``extra`` is not forbidden on this schema, but the field is simply
    # not writable here — the row must come back unowned.
    assert resp.status_code in (200, 422)
    fresh = await client.get(f"{BASE}/{system['id']}", headers=admin_headers)
    assert fresh.json()["data"]["owner"] is None


@pytest.mark.asyncio
async def test_owner_transfer_writes_an_event_trail(
    client, admin_headers, db_session,
):
    first = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com",
                          name="First Owner")
    second = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com",
                           name="Second Owner")
    system = await _create(client, admin_headers)

    resp = await client.post(
        f"{BASE}/{system['id']}/owner",
        json={"owner_person_id": str(first.id)},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["owner"]["name"] == "First Owner"
    assert resp.json()["data"]["owner_assigned_at"] is not None

    resp = await client.post(
        f"{BASE}/{system['id']}/owner",
        json={"owner_person_id": str(second.id), "note": "轮换"},
        headers=admin_headers,
    )
    assert resp.json()["data"]["owner"]["name"] == "Second Owner"

    events = (
        await client.get(f"{BASE}/{system['id']}/events", headers=admin_headers)
    ).json()["data"]
    kinds = [e["event_type"] for e in events]
    assert kinds.count("owner_assigned") == 2
    # The trail must name BOTH sides — "who took over from whom" is the
    # question an incident review asks. Selected by content rather than
    # by position: ``created_at`` is second-granular, so two transfers
    # inside one second have no defined order and asserting on "the
    # newest row" would be a flake, not a check.
    transfers = [e for e in events if e["event_type"] == "owner_assigned"]
    handover = next(
        e for e in transfers if e["meta"]["to_name"] == "Second Owner"
    )
    assert handover["meta"]["from_name"] == "First Owner"
    first = next(e for e in transfers if e["meta"]["to_name"] == "First Owner")
    assert first["meta"]["from_name"] is None


@pytest.mark.asyncio
async def test_promoting_the_backup_clears_the_backup_slot(
    client, admin_headers, db_session,
):
    owner = await _person(db_session, f"o-{uuid.uuid4().hex[:6]}@t.com")
    backup = await _person(db_session, f"k-{uuid.uuid4().hex[:6]}@t.com")
    system = await _create(
        client, admin_headers,
        owner_person_id=str(owner.id),
        backup_owner_person_id=str(backup.id),
    )
    assert system["backup_owner"]["id"] == str(backup.id)

    resp = await client.post(
        f"{BASE}/{system['id']}/owner",
        json={"owner_person_id": str(backup.id)},
        headers=admin_headers,
    )
    data = resp.json()["data"]
    assert data["owner"]["id"] == str(backup.id)
    # Same person cannot hold both slots — otherwise the next handover
    # has no successor to propose.
    assert data["backup_owner"] is None


@pytest.mark.asyncio
async def test_owner_and_backup_cannot_be_the_same_at_creation(
    client, admin_headers, db_session,
):
    person = await _person(db_session, f"same-{uuid.uuid4().hex[:6]}@t.com")
    resp = await client.post(
        BASE,
        json={
            "name": "Dup", "slug": f"dup-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(person.id),
            "backup_owner_person_id": str(person.id),
        },
        headers=admin_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_ownerless_filter_and_coverage_exclude_retired(
    client, admin_headers, db_session,
):
    owner = await _person(db_session, f"c-{uuid.uuid4().hex[:6]}@t.com")
    owned = await _create(
        client, admin_headers, slug=f"owned-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(owner.id),
    )
    bare = await _create(client, admin_headers,
                         slug=f"bare-{uuid.uuid4().hex[:6]}")

    listed = (
        await client.get(f"{BASE}?ownerless=true&page_size=100",
                         headers=admin_headers)
    ).json()["data"]
    ids = {row["id"] for row in listed}
    assert bare["id"] in ids
    assert owned["id"] not in ids

    before = (
        await client.get(f"{BASE}/meta", headers=admin_headers)
    ).json()["data"]["summary"]
    resp = await client.delete(f"{BASE}/{bare['id']}", headers=admin_headers)
    assert resp.status_code == 200
    assert resp.json()["data"]["status"] == "retired"
    after = (
        await client.get(f"{BASE}/meta", headers=admin_headers)
    ).json()["data"]["summary"]
    # Switching something off correctly must not look like a governance
    # gap: the ownerless count drops, it does not stay flat.
    assert after["ownerless"] == before["ownerless"] - 1
    assert after["live"] == before["live"] - 1


@pytest.mark.asyncio
async def test_suggest_matches_hostname_and_skips_retired(
    client, admin_headers,
):
    live = await _create(
        client, admin_headers, slug=f"live-{uuid.uuid4().hex[:6]}",
        hostnames=["clinic.pingcap.com"],
    )
    gone = await _create(
        client, admin_headers, slug=f"gone-{uuid.uuid4().hex[:6]}",
        hostnames=["clinic-old.pingcap.com"],
    )
    await client.delete(f"{BASE}/{gone['id']}", headers=admin_headers)

    hits = (
        await client.get(f"{BASE}/suggest?host=https://clinic.pingcap.com/x",
                         headers=admin_headers)
    ).json()["data"]
    ids = {row["id"] for row in hits}
    assert live["id"] in ids
    assert gone["id"] not in ids


@pytest.mark.asyncio
async def test_my_systems_is_scoped_to_the_caller(
    client, db_session, employee_user, employee_headers, admin_headers,
):
    me, _token = employee_user
    mine = await _create(
        client, admin_headers, slug=f"mine-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(me.id),
    )
    other_owner = await _person(db_session, f"z-{uuid.uuid4().hex[:6]}@t.com")
    theirs = await _create(
        client, admin_headers, slug=f"theirs-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(other_owner.id),
    )
    backup_of = await _create(
        client, admin_headers, slug=f"bk-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(other_owner.id),
        backup_owner_person_id=str(me.id),
    )

    data = (
        await client.get("/api/v1/me/systems", headers=employee_headers)
    ).json()["data"]
    by_id = {row["id"]: row for row in data["items"]}
    assert mine["id"] in by_id
    assert backup_of["id"] in by_id
    assert theirs["id"] not in by_id
    assert by_id[mine["id"]]["my_role"] == "owner"
    assert by_id[backup_of["id"]]["my_role"] == "backup_owner"
    assert data["owned"] == 1
    assert data["backup"] == 1


@pytest.mark.asyncio
async def test_export_carries_owner_columns(client, admin_headers):
    await _create(client, admin_headers, slug=f"exp-{uuid.uuid4().hex[:6]}")
    resp = await client.get(f"{BASE}/export", headers=admin_headers)
    assert resp.status_code == 200
    assert "spreadsheet" in resp.headers["content-type"]


@pytest.mark.asyncio
async def test_soft_deleted_person_can_still_act(
    client, admin_headers, db_session,
):
    """A null actor is recorded rather than 500-ing the write."""
    system = await _create(client, admin_headers,
                           slug=f"act-{uuid.uuid4().hex[:6]}")
    resp = await client.patch(
        f"{BASE}/{system['id']}", json={"notes": "hello"},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    events = (
        await client.get(f"{BASE}/{system['id']}/events", headers=admin_headers)
    ).json()["data"]
    assert any(e["event_type"] == "updated" for e in events)


@pytest.mark.asyncio
async def test_a_departed_person_cannot_be_named_owner(
    client, admin_headers, db_session,
):
    """The bounty domain already refuses to assign work to a leaver.
    Ownership is the stronger claim and was letting one through."""
    gone = await _person(db_session, f"gone-{uuid.uuid4().hex[:6]}@t.com")
    # ``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.employment_status = "leaver_pending"
    await db_session.commit()

    resp = await client.post(
        BASE,
        json={
            "name": "Ghost", "slug": f"ghost-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(gone.id),
        },
        headers=admin_headers,
    )
    assert resp.status_code == 409
    assert resp.json()["errors"][0]["code"] == "SYSTEM_OWNER_NOT_ACTIVE"

    system = await _create(client, admin_headers,
                           slug=f"live-{uuid.uuid4().hex[:6]}")
    resp = await client.post(
        f"{BASE}/{system['id']}/owner",
        json={"owner_person_id": str(gone.id)},
        headers=admin_headers,
    )
    assert resp.status_code == 409


@pytest.mark.asyncio
async def test_a_system_whose_owner_left_is_not_counted_as_covered(
    client, admin_headers, db_session,
):
    """A 100% coverage number that coexists with nobody being accountable
    is the failure this metric exists to prevent. The owner leaves AFTER
    being named, so the write-time guard above cannot catch it."""
    owner = await _person(db_session, f"leaver-{uuid.uuid4().hex[:6]}@t.com")
    system = await _create(
        client, admin_headers, slug=f"cov-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(owner.id),
    )
    before = (
        await client.get(f"{BASE}/meta", headers=admin_headers)
    ).json()["data"]["summary"]
    assert before["owned"] >= 1

    owner.employment_status = "leaver_pending"
    await db_session.commit()

    after = (
        await client.get(f"{BASE}/meta", headers=admin_headers)
    ).json()["data"]["summary"]
    assert after["owned"] == before["owned"] - 1
    assert after["owner_departed"] == before["owner_departed"] + 1
    # NOT folded into ownerless: the two need different action.
    assert after["ownerless"] == before["ownerless"]
    assert after["coverage_pct"] < before["coverage_pct"]

    # The tile's filter must select the same rows the tile counts.
    resp = await client.get(
        f"{BASE}?ownerless=true&page_size=100", headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    assert system["id"] in {row["id"] for row in resp.json()["data"]}


@pytest.mark.asyncio
async def test_only_the_named_person_can_dispute_their_ownership(
    client, admin_headers, db_session, employee_user, employee_headers,
):
    """The employee cannot re-point the row — ownership is IT's record —
    but they can contest one that names them, and only one that does."""
    me, _token = employee_user
    stranger = await _person(db_session, f"s-{uuid.uuid4().hex[:6]}@t.com")

    mine = await _create(
        client, admin_headers, slug=f"mine-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(me.id),
    )
    theirs = await _create(
        client, admin_headers, slug=f"not-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(stranger.id),
    )

    resp = await client.post(
        f"/api/v1/me/systems/{theirs['id']}/dispute",
        json={"note": "not mine"}, headers=employee_headers,
    )
    assert resp.status_code == 403

    resp = await client.post(
        f"/api/v1/me/systems/{mine['id']}/dispute",
        json={"note": "这个系统 2025 年就交给平台组了"},
        headers=employee_headers,
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["recorded"] is True

    # The claim lands on the system's own trail, next to the ownership
    # history it disputes — not in a separate inbox IT has to remember.
    events = (
        await client.get(f"{BASE}/{mine['id']}/events", headers=admin_headers)
    ).json()["data"]
    disputed = [e for e in events if e["event_type"] == "owner_disputed"]
    assert len(disputed) == 1
    assert disputed[0]["meta"]["role"] == "owner"
    assert "平台组" in disputed[0]["body"]

    # Ownership itself is unchanged — contesting is not editing.
    fresh = (
        await client.get(f"{BASE}/{mine['id']}", headers=admin_headers)
    ).json()["data"]
    assert fresh["owner"]["id"] == str(me.id)


@pytest.mark.asyncio
async def test_a_backup_owner_can_dispute_too(
    client, admin_headers, db_session, employee_user, employee_headers,
):
    me, _token = employee_user
    owner = await _person(db_session, f"o-{uuid.uuid4().hex[:6]}@t.com")
    system = await _create(
        client, admin_headers, slug=f"bk-{uuid.uuid4().hex[:6]}",
        owner_person_id=str(owner.id), backup_owner_person_id=str(me.id),
    )
    resp = await client.post(
        f"/api/v1/me/systems/{system['id']}/dispute", json={},
        headers=employee_headers,
    )
    assert resp.status_code == 200
    events = (
        await client.get(f"{BASE}/{system['id']}/events", headers=admin_headers)
    ).json()["data"]
    disputed = next(e for e in events if e["event_type"] == "owner_disputed")
    assert disputed["meta"]["role"] == "backup_owner"