"""Self-service DNS / server batch owner transfer (2026-06-02).

Covers ``POST /cmdb/dns-records/batch-owner-transfer`` and
``POST /cmdb/servers/batch-owner-transfer`` — the employee-facing
handoff path that lets the current owner reassign records / servers to
an active teammate without IT involvement. Key contracts verified:

* employee can transfer rows they own; new owner is updated; audit row
  + owner_history rows land
* employee cannot transfer rows owned by someone else (403)
* admin bypass: global_admin can transfer mixed-ownership batch
* recipient must be an active employee (leaver_pending → 400)
* recipient cannot equal the caller (no-op self-transfer rejected)
* idempotent — re-running the same payload after a successful transfer
  is a no-op (transferred=0)
* DNS path clears ``last_recertify_nag_at`` + ``last_owner_review_at``
  so the new owner enters the next recertify cycle
"""
from __future__ import annotations

import uuid

import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.domains.cmdb.models import CmdbDnsRecord, CmdbServer
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _make_person(
    db: AsyncSession, *, email: str, status: str = "active",
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    return p


@pytest_asyncio.fixture
async def integ(db_session: AsyncSession) -> Integration:
    i = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="tencent",
        name="pytest-batch-transfer",
        slug=f"pytest-batch-transfer-{uuid.uuid4().hex[:8]}",
        enabled=True,
        is_primary=False,
        capabilities=["DNS_RECORDS", "SERVERS"],
    )
    db_session.add(i)
    await db_session.commit()
    return i


async def _make_dns(
    db: AsyncSession, integ: Integration, *, owner_id: str,
    name: str = "rec",
) -> CmdbDnsRecord:
    r = CmdbDnsRecord(
        integration_id=integ.id,
        zone_name="transfer.example.com",
        record_id=f"prov-{uuid.uuid4().hex[:8]}",
        name=name,
        record_type="A",
        value="1.2.3.4",
        ttl=600,
        status="enable",
        owner_person_id=owner_id,
    )
    db.add(r)
    await db.flush()
    return r


async def _make_server(
    db: AsyncSession, integ: Integration, *, owner_id: str,
    name: str = "vm", last_seen_sync_job_id: str | None = "seeded-sync-job",
) -> CmdbServer:
    # Default to a sync-confirmed row (``last_seen_sync_job_id`` set) — that
    # is the real-world default for every live server. Pass ``None`` to
    # simulate a never-confirmed stale row that transfer must refuse.
    s = CmdbServer(
        integration_id=integ.id,
        instance_id=f"i-{uuid.uuid4().hex[:8]}",
        name=name,
        status="running",
        region="ap-southeast-1",
        owner_person_id=owner_id,
        last_seen_sync_job_id=last_seen_sync_job_id,
    )
    db.add(s)
    await db.flush()
    return s


# ─── DNS ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_dns_employee_can_transfer_owned_records(
    client, employee_user, integ, db_session,
):
    """Happy path: employee transfers 2 records they own; new owner
    updated; audit + owner_history land; recertify dedup cleared."""
    caller, token = employee_user
    teammate = await _make_person(db_session, email="teammate@test.com")
    r1 = await _make_dns(db_session, integ, owner_id=str(caller.id), name="r1")
    r2 = await _make_dns(db_session, integ, owner_id=str(caller.id), name="r2")
    # Seed a stale review timestamp so we can prove the transfer clears it.
    from datetime import UTC, datetime, timedelta
    r1.last_owner_review_at = datetime.now(UTC) - timedelta(days=10)
    r1.last_recertify_nag_at = datetime.now(UTC) - timedelta(days=10)
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r1.id), str(r2.id)],
            "new_owner_person_id": str(teammate.id),
            "reason": "Handing off the auth-experiments stack",
        },
    )
    assert resp.status_code == 200, resp.text
    summary = resp.json()["data"]
    assert summary["transferred"] == 2
    assert summary["not_owned"] == 0
    assert summary["not_found"] == 0

    for r in (r1, r2):
        await db_session.refresh(r)
        assert r.owner_person_id == str(teammate.id)
        # Recertify dedup must reset so the new owner enters next cycle.
        assert r.last_owner_review_at is None
        assert r.last_recertify_nag_at is None

    audit = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "cmdb.dns.self_service_transfer")
    )).scalars().all()
    assert len(audit) == 1
    assert audit[0].actor_user_id == caller.id


@pytest.mark.asyncio
async def test_dns_employee_cannot_transfer_others_records(
    client, employee_user, integ, db_session,
):
    """An employee selecting a record they don't own gets 403 and no
    changes commit. The ownership check is a single-shot rejection —
    the batch is atomic."""
    caller, token = employee_user
    stranger = await _make_person(db_session, email="stranger@test.com")
    teammate = await _make_person(db_session, email="teammate2@test.com")
    mine = await _make_dns(db_session, integ, owner_id=str(caller.id), name="mine")
    theirs = await _make_dns(db_session, integ, owner_id=str(stranger.id), name="theirs")
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(mine.id), str(theirs.id)],
            "new_owner_person_id": str(teammate.id),
        },
    )
    assert resp.status_code == 403, resp.text
    # The "mine" record should NOT have changed owner — atomic rejection.
    await db_session.refresh(mine)
    await db_session.refresh(theirs)
    assert mine.owner_person_id == str(caller.id)
    assert theirs.owner_person_id == str(stranger.id)


@pytest.mark.asyncio
async def test_dns_admin_bypasses_ownership_check(
    client, admin_user, integ, db_session,
):
    """Admin can transfer rows they don't personally own — the
    ``can_manage_cloud`` carve-out exists so an admin can re-bind on
    someone's behalf without first becoming the owner."""
    _, token = admin_user
    a = await _make_person(db_session, email="a@test.com")
    b = await _make_person(db_session, email="b@test.com")
    r = await _make_dns(db_session, integ, owner_id=str(a.id))
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r.id)],
            "new_owner_person_id": str(b.id),
        },
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["transferred"] == 1
    await db_session.refresh(r)
    assert r.owner_person_id == str(b.id)


@pytest.mark.asyncio
async def test_dns_transfer_to_leaver_rejected(
    client, employee_user, integ, db_session,
):
    """Recipient must be ``employment_status='active'`` — handing off
    to a leaver_pending teammate would silently re-orphan once T+0
    fires."""
    caller, token = employee_user
    leaver = await _make_person(
        db_session, email="leaver@test.com", status="leaver_pending",
    )
    r = await _make_dns(db_session, integ, owner_id=str(caller.id))
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r.id)],
            "new_owner_person_id": str(leaver.id),
        },
    )
    assert resp.status_code in (400, 422), resp.text


@pytest.mark.asyncio
async def test_dns_transfer_to_self_rejected(
    client, employee_user, integ, db_session,
):
    """Self-transfer is meaningless — reject so the audit log doesn't
    fill with no-op rows from misclicks."""
    caller, token = employee_user
    r = await _make_dns(db_session, integ, owner_id=str(caller.id))
    await db_session.commit()
    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r.id)],
            "new_owner_person_id": str(caller.id),
        },
    )
    assert resp.status_code in (400, 422), resp.text


@pytest.mark.asyncio
async def test_dns_idempotent_on_replay(
    client, employee_user, integ, db_session,
):
    """Replay of a successful transfer is a no-op: the second call
    returns transferred=0 because both rows already belong to the
    target. Important for the FE retry behaviour on slow networks."""
    caller, token = employee_user
    teammate = await _make_person(db_session, email="teammate3@test.com")
    r = await _make_dns(db_session, integ, owner_id=str(caller.id))
    await db_session.commit()

    payload = {
        "record_ids": [str(r.id)],
        "new_owner_person_id": str(teammate.id),
    }
    headers = {"Authorization": f"Bearer {token}"}

    r1 = await client.post("/api/v1/cmdb/dns-records/batch-owner-transfer",
                           headers=headers, json=payload)
    assert r1.status_code == 200
    assert r1.json()["data"]["transferred"] == 1

    # Second call — caller no longer owns the record, so the
    # ownership check rejects with 403. That's the right answer: the
    # FE should refresh and re-render after a successful transfer.
    r2 = await client.post("/api/v1/cmdb/dns-records/batch-owner-transfer",
                           headers=headers, json=payload)
    assert r2.status_code == 403, r2.text


# ─── Server ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_server_employee_can_transfer_owned_servers(
    client, employee_user, integ, db_session,
):
    caller, token = employee_user
    teammate = await _make_person(db_session, email="teammate4@test.com")
    s1 = await _make_server(db_session, integ, owner_id=str(caller.id), name="s1")
    s2 = await _make_server(db_session, integ, owner_id=str(caller.id), name="s2")
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/servers/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "server_ids": [str(s1.id), str(s2.id)],
            "new_owner_person_id": str(teammate.id),
            "reason": "Off-boarding the K8s nodes I was babysitting",
        },
    )
    assert resp.status_code == 200, resp.text
    summary = resp.json()["data"]
    assert summary["transferred"] == 2

    for s in (s1, s2):
        await db_session.refresh(s)
        assert s.owner_person_id == str(teammate.id)

    audit = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "cmdb.server.self_service_transfer")
    )).scalars().all()
    assert len(audit) == 1


@pytest.mark.asyncio
async def test_server_employee_cannot_transfer_others_servers(
    client, employee_user, integ, db_session,
):
    caller, token = employee_user
    stranger = await _make_person(db_session, email="stranger2@test.com")
    teammate = await _make_person(db_session, email="teammate5@test.com")
    mine = await _make_server(db_session, integ, owner_id=str(caller.id), name="mine")
    theirs = await _make_server(db_session, integ, owner_id=str(stranger.id), name="theirs")
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/servers/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "server_ids": [str(mine.id), str(theirs.id)],
            "new_owner_person_id": str(teammate.id),
        },
    )
    assert resp.status_code == 403, resp.text
    await db_session.refresh(mine)
    assert mine.owner_person_id == str(caller.id)


@pytest.mark.asyncio
async def test_server_never_confirmed_row_refused(
    client, employee_user, integ, db_session,
):
    """A live-but-never-sync-confirmed row (``last_seen_sync_job_id IS
    NULL``) may already be gone from the cloud (2026-07 Oncall_bot
    dangling-transfer incident). Refuse the hand-off instead of shipping a
    misleading "you are now the owner" notification for a ghost."""
    caller, token = employee_user
    teammate = await _make_person(db_session, email="teammate6@test.com")
    ghost = await _make_server(
        db_session, integ, owner_id=str(caller.id), name="ghost",
        last_seen_sync_job_id=None,
    )
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/servers/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "server_ids": [str(ghost.id)],
            "new_owner_person_id": str(teammate.id),
        },
    )
    assert resp.status_code in (400, 422), resp.text
    await db_session.refresh(ghost)
    assert ghost.owner_person_id == str(caller.id)  # unchanged

    # No transfer audit row was written for the ghost.
    audit = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "cmdb.server.self_service_transfer"
        )
    )).scalars().all()
    assert audit == []


@pytest.mark.asyncio
async def test_server_mixed_batch_transfers_confirmed_skips_ghost(
    client, employee_user, integ, db_session,
):
    """A batch mixing a real (sync-confirmed) server with a ghost transfers
    the real one and reports the ghost as ``unconfirmed`` — partial success,
    not an all-or-nothing failure."""
    caller, token = employee_user
    teammate = await _make_person(db_session, email="teammate7@test.com")
    real = await _make_server(
        db_session, integ, owner_id=str(caller.id), name="real",
    )
    ghost = await _make_server(
        db_session, integ, owner_id=str(caller.id), name="ghost2",
        last_seen_sync_job_id=None,
    )
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/servers/batch-owner-transfer",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "server_ids": [str(real.id), str(ghost.id)],
            "new_owner_person_id": str(teammate.id),
        },
    )
    assert resp.status_code == 200, resp.text
    summary = resp.json()["data"]
    assert summary["transferred"] == 1
    assert summary["unconfirmed"] == 1

    await db_session.refresh(real)
    await db_session.refresh(ghost)
    assert real.owner_person_id == str(teammate.id)
    assert ghost.owner_person_id == str(caller.id)  # ghost untouched