"""Bypass + retry endpoints for 4-eyes-blocked DNS requests.

Covers the two admin-recovery service methods added 2026-05-21:

* ``DnsRequestService.bypass_4_eyes_execute`` — explicit, audited
  override that skips the 4-eyes guard.
* ``DnsRequestService.retry_after_template_fix`` — re-opens a fresh
  Feishu approval instance for a 4-eyes-blocked row.

Both methods gate on (a) ``can_manage_cloud`` (global_admin /
cloud_admin) and (b) the row being in ``failed`` state with a 4-eyes
error_message. Anything else is a 403 / 409 (per the threat model:
bypass and retry must NOT be reachable from a cloud-write failure or
from a non-admin caller).
"""
from __future__ import annotations

import uuid
from unittest.mock import AsyncMock, MagicMock, patch

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

from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent
from app.audit.models import AuditLog
from app.auth.models import AuthRole, AuthUserRole
from app.core.exceptions import AppException, ValidationError
from app.domains.cmdb.dns_request_service import DnsRequestService
from app.domains.cmdb.models import (
    DNS_REQ_ACTION_CREATE,
    DNS_REQ_APPROVED,
    DNS_REQ_FAILED,
    DNS_REQ_PENDING,
    CmdbDnsRecord,
    CmdbDnsRecordRequest,
    CmdbDomain,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


@pytest_asyncio.fixture
async def integ(db_session: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="tencent",
        name=f"pytest-recovery-{uuid.uuid4().hex[:8]}",
        slug=f"pytest-recovery-{uuid.uuid4().hex[:8]}",
        enabled=True,
        is_primary=False,
        capabilities=["dns_records"],  # lowercase: matches ProviderCapability.value
    )
    db_session.add(integ)
    await db_session.flush()

    # Zone allowlist row — retry re-checks ``allow_dns_request`` so we
    # need this for the retry success path.
    domain = CmdbDomain(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        domain_name="example.com",
        registrar=None,
        status="active",
        allow_dns_request=True,
    )
    db_session.add(domain)
    # DNS authority anchor — retry re-resolves the zone via
    # ``CmdbService.resolve_dns_integration_for_zone`` (added
    # 2026-06-01) so the retry path can never silently route to a
    # stale tenant after DNS authority has moved. The resolver needs
    # at least one ``cmdb_dns_records`` row pointing at the same
    # integration to confirm it's still the authoritative provider.
    anchor = CmdbDnsRecord(
        integration_id=str(integ.id),
        zone_name="example.com",
        record_id="zone-anchor",
        name="@",
        record_type="NS",
        value="ns1.example.com",
        ttl=3600,
        status="enable",
    )
    db_session.add(anchor)
    await db_session.commit()
    return integ


async def _make_person(
    db_session: AsyncSession, *, email: str, role: str | None = None,
) -> IdentityPerson:
    """Create an IdentityPerson, optionally tagged with one role.

    Roles are looked up by name in ``auth_roles`` (created by the
    Alembic seed in conftest.py) — we don't insert role rows here.
    """
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="test",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        work_email=email,
        display_name=email.split("@", 1)[0],
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()
    if role is not None:
        role_row = (
            await db_session.execute(
                select(AuthRole).where(AuthRole.name == role),
            )
        ).scalar_one_or_none()
        if role_row is None:
            # Greenfield test DB may not seed every role; insert on the
            # fly so the test is hermetic.
            role_row = AuthRole(id=uuid.uuid4(), name=role, description=role)
            db_session.add(role_row)
            await db_session.flush()
        db_session.add(
            AuthUserRole(user_id=person.id, role_id=role_row.id),
        )
    await db_session.commit()
    # Refresh so .role_names sees the freshly-committed AuthUserRole.
    await db_session.refresh(person, attribute_names=["user_roles"])
    return person


async def _make_failed_row(
    db_session: AsyncSession,
    *,
    integ: Integration,
    requester: IdentityPerson,
    error_message: str = (
        "Blocked: requester appears as an approver on this approval "
        "(4-eyes violation). Re-route the Feishu approval template so "
        "the requester is not in any approver node."
    ),
) -> tuple[CmdbDnsRecordRequest, ApprovalRequest]:
    """Materialize a row that mirrors the post-4-eyes-block durable
    state (status=failed, error_message contains '4-eyes', decided_at
    set, feishu_instance_code present).
    """
    from datetime import UTC, datetime

    row = CmdbDnsRecordRequest(
        requester_person_id=str(requester.id),
        action=DNS_REQ_ACTION_CREATE,
        integration_id=str(integ.id),
        zone_name="example.com",
        name="api",
        record_type="A",
        value="10.0.0.1",
        ttl=600,
        line=None,
        purpose="pytest: admin recovery",
        status=DNS_REQ_FAILED,
        error_message=error_message,
        feishu_instance_code=f"instance-{uuid.uuid4().hex[:12]}",
        decided_at=datetime.now(UTC),
    )
    db_session.add(row)
    await db_session.flush()
    ar = ApprovalRequest(
        id=uuid.uuid4(),
        applicant_id=requester.id,
        approval_type="dns_record_request",
        external_id=row.feishu_instance_code,
        status="approved",
        source_payload={"dns_request_id": str(row.id)},
    )
    db_session.add(ar)
    await db_session.commit()
    return row, ar


# ── bypass_4_eyes_execute ────────────────────────────────────────


@pytest.mark.asyncio
async def test_bypass_rejects_non_admin(db_session, integ):
    requester = await _make_person(
        db_session, email="r1@test.com", role="cloud_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=requester,
    )
    employee = await _make_person(
        db_session, email="e1@test.com", role="employee",
    )

    svc = DnsRequestService(db_session)
    with pytest.raises(AppException) as exc:
        await svc.bypass_4_eyes_execute(
            request_id=str(row.id),
            actor=employee,
            reason="long enough rationale for bypass attestation",
            trace_id="test",
        )
    assert exc.value.status_code == 403


@pytest.mark.asyncio
async def test_bypass_rejects_short_reason(db_session, integ):
    admin = await _make_person(
        db_session, email="a1@test.com", role="global_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=admin,
    )

    svc = DnsRequestService(db_session)
    with pytest.raises(ValidationError):
        await svc.bypass_4_eyes_execute(
            request_id=str(row.id),
            actor=admin,
            reason="too short",
            trace_id="test",
        )


@pytest.mark.asyncio
async def test_bypass_rejects_non_four_eyes_failure(db_session, integ):
    """A cloud-write failure (status=failed but error_message has no
    4-eyes fragment) MUST NOT be reachable via bypass — bypassing would
    double-fire the provider call.

    Round-45 update: requester and bypasser must differ — the
    self-bypass guard (DNS_REQ_BYPASS_SELF_FORBIDDEN) fires before
    the NOT_APPLICABLE check otherwise.
    """
    admin = await _make_person(
        db_session, email="a2@test.com", role="global_admin",
    )
    requester = await _make_person(
        db_session, email="a2-req@test.com", role="employee",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=requester,
        error_message="Route53 returned 500: throttled",
    )

    svc = DnsRequestService(db_session)
    with pytest.raises(AppException) as exc:
        await svc.bypass_4_eyes_execute(
            request_id=str(row.id),
            actor=admin,
            reason="long enough rationale for bypass attestation",
            trace_id="test",
        )
    assert exc.value.status_code == 409


@pytest.mark.asyncio
async def test_bypass_happy_path_flips_status_and_audits(db_session, integ):
    admin = await _make_person(
        db_session, email="a3@test.com", role="cloud_admin",
    )
    # Round-45: requester must differ from the bypasser (self-bypass
    # is now refused with DNS_REQ_BYPASS_SELF_FORBIDDEN 403). The
    # requester is typically an employee / developer who submitted a
    # DNS change; the admin is the one signing off the bypass.
    requester = await _make_person(
        db_session, email="a3-req@test.com", role="employee",
    )
    # Add a peer admin so the notify fan-out has a real recipient to
    # build an envelope for — otherwise the early-out "no recipients"
    # branch skips the CardHighlight construction and we can't catch
    # envelope-shape bugs (prod hit one such bug 2026-05-21 because the
    # earlier version of this test mocked the whole _notify_bypass).
    peer = await _make_person(  # noqa: F841 — created for fan-out target
        db_session, email="peer1@test.com", role="global_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=requester,
    )

    svc = DnsRequestService(db_session)
    # Patch the cloud-write hand-off (don't actually call Tencent) and
    # the network boundary of the notification system (don't actually
    # call Feishu/email), but let _notify_bypass build the full
    # NotificationEnvelope + CardHighlight payload so misnamed fields
    # surface here instead of in prod.
    with (
        patch.object(
            svc, "_execute_approved", new=AsyncMock(return_value=None),
        ) as exec_mock,
        patch(
            "app.notifications.service.NotificationService.dispatch",
            new=AsyncMock(return_value=None),
        ) as dispatch_mock,
    ):
        await svc.bypass_4_eyes_execute(
            request_id=str(row.id),
            actor=admin,
            reason="single-admin shop, no peer reviewer available — "
                   "emergency cert renewal",
            trace_id="test-bypass",
        )

    exec_mock.assert_awaited_once()
    # Notification went to the peer admin (NOT the bypasser).
    assert dispatch_mock.await_count == 1
    sent_env = dispatch_mock.await_args_list[0].args[0]
    assert sent_env.recipient.work_email == "peer1@test.com"
    assert sent_env.card_highlights, (
        "envelope must carry the highlights block — the prod bug we're "
        "regression-pinning was a CardHighlight kwarg name mismatch that "
        "raised TypeError before dispatch was reached"
    )

    # Row should have been flipped to APPROVED before the execute hand-off
    # (the stub didn't progress it further). error_message cleared.
    await db_session.refresh(row)
    assert row.status == DNS_REQ_APPROVED
    assert row.error_message is None

    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == (
                    "cmdb.dns_request.bypassed_4_eyes_execute"
                ),
                AuditLog.resource_id == str(row.id),
            ),
        )
    ).scalars().all()
    assert len(audits) == 1
    detail = audits[0].detail or {}
    assert "bypass_reason" in detail
    assert "emergency cert renewal" in detail["bypass_reason"]
    assert detail.get("original_error_message", "").lower().count("4-eyes") == 1


@pytest.mark.asyncio
async def test_bypass_notify_fan_out_excludes_self(db_session, integ):
    """The bypasser MUST be excluded from the peer-visibility fan-out.
    Otherwise admins see a "X bypassed" notification about their own
    action — noise that erodes the audit signal.
    """
    admin = await _make_person(
        db_session, email="actor@test.com", role="global_admin",
    )
    peer = await _make_person(
        db_session, email="peer@test.com", role="global_admin",
    )
    cloud_peer = await _make_person(
        db_session, email="cloud-peer@test.com", role="cloud_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=admin,
    )

    svc = DnsRequestService(db_session)
    recipients = await svc._list_admin_recipients_excluding(
        exclude_person_id=str(admin.id),
    )
    rec_ids = {str(r.id) for r in recipients}
    assert str(admin.id) not in rec_ids
    assert str(peer.id) in rec_ids
    assert str(cloud_peer.id) in rec_ids


# ── retry_after_template_fix ─────────────────────────────────────


@pytest.mark.asyncio
async def test_retry_rejects_non_admin(db_session, integ):
    requester = await _make_person(
        db_session, email="r2@test.com", role="cloud_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=requester,
    )
    employee = await _make_person(
        db_session, email="e2@test.com", role="employee",
    )

    svc = DnsRequestService(db_session)
    with pytest.raises(AppException) as exc:
        await svc.retry_after_template_fix(
            request_id=str(row.id),
            actor=employee,
            trace_id="test",
        )
    assert exc.value.status_code == 403


@pytest.mark.asyncio
async def test_retry_rejects_non_four_eyes_failure(db_session, integ):
    admin = await _make_person(
        db_session, email="a4@test.com", role="global_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=admin,
        error_message="Aliyun 400: name conflict",
    )

    svc = DnsRequestService(db_session)
    with pytest.raises(AppException) as exc:
        await svc.retry_after_template_fix(
            request_id=str(row.id),
            actor=admin,
            trace_id="test",
        )
    assert exc.value.status_code == 409


@pytest.mark.asyncio
async def test_retry_opens_new_feishu_instance_and_resets_row(
    db_session, integ,
):
    """Happy path: row is reset to ``pending`` with a new
    ``feishu_instance_code``, a new ``ApprovalRequest`` row is created,
    and the old ApprovalRequest row's ``external_id`` is preserved
    (we don't touch it — important for audit).
    """
    admin = await _make_person(
        db_session, email="a5@test.com", role="global_admin",
    )
    row, ar = await _make_failed_row(
        db_session, integ=integ, requester=admin,
    )
    original_instance_code = row.feishu_instance_code

    # Patch the approval flow + Feishu client. The retry path calls:
    #   get_approval_flow(...) → ApprovalFlow with external_code
    #   client.get_approval_definition(approval_code) → {form: "[]"}
    #   client.resolve_user_id_by_email(email) → user_id
    #   client.create_instance(...) → {"instance_code": "..."}
    new_instance_code = f"new-{uuid.uuid4().hex[:12]}"

    fake_flow = MagicMock()
    fake_flow.external_code = "APPROVAL_CODE_X"
    fake_client = MagicMock()
    fake_client.get_approval_definition = AsyncMock(
        return_value={"form": "[]"},
    )
    fake_client.resolve_user_id_by_email = AsyncMock(return_value="u-fake")
    fake_client.create_instance = AsyncMock(
        return_value={"instance_code": new_instance_code},
    )

    svc = DnsRequestService(db_session)
    with (
        patch(
            "app.domains.cmdb.dns_request_service.get_approval_flow",
            new=AsyncMock(return_value=fake_flow),
        ),
        patch.object(
            svc, "_client", new=AsyncMock(return_value=fake_client),
        ),
    ):
        await svc.retry_after_template_fix(
            request_id=str(row.id),
            actor=admin,
            trace_id="test-retry",
        )

    await db_session.refresh(row)
    assert row.status == DNS_REQ_PENDING
    assert row.error_message is None
    assert row.decided_at is None
    assert row.feishu_instance_code == new_instance_code
    assert row.feishu_instance_code != original_instance_code

    # Old ApprovalRequest row left intact for audit.
    await db_session.refresh(ar)
    assert ar.external_id == original_instance_code

    # New ApprovalRequest row exists with the new instance code.
    new_ar = (
        await db_session.execute(
            select(ApprovalRequest).where(
                ApprovalRequest.external_id == new_instance_code,
            ),
        )
    ).scalar_one_or_none()
    assert new_ar is not None
    assert new_ar.source_payload.get("retry_of_instance_code") == (
        original_instance_code
    )

    # Audit row written.
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cmdb.dns_request.retried_after_block",
                AuditLog.resource_id == str(row.id),
            ),
        )
    ).scalars().all()
    assert len(audits) == 1


@pytest.mark.asyncio
async def test_retry_response_loss_reuses_uuid_and_recovers_60012(
    db_session, integ,
):
    admin = await _make_person(
        db_session, email="a5-loss@test.com", role="global_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=admin,
    )
    fake_flow = MagicMock(external_code="APPROVAL_CODE_X")
    fake_client = MagicMock()
    fake_client.get_approval_definition = AsyncMock(return_value={"form": "[]"})
    fake_client.resolve_user_id_by_email = AsyncMock(return_value="u-fake")
    provider_uuids: list[str] = []

    async def _create_instance(**kwargs):
        provider_uuids.append(kwargs["instance_uuid"])
        if len(provider_uuids) == 1:
            raise httpx.ReadTimeout("accepted but response was lost")
        raise AppException(
            code="FEISHU_API_ERROR",
            message="Feishu API error 60012: duplicate uuid",
            status_code=502,
        )

    async def _find_instance(**kwargs):
        assert kwargs["instance_uuid"] == provider_uuids[0]
        assert kwargs["end_time"] >= kwargs["start_time"]
        return {
            "instance_code": "dns-recovered-instance",
            "uuid": kwargs["instance_uuid"],
        }

    fake_client.create_instance = _create_instance
    fake_client.find_instance_by_uuid = _find_instance
    svc = DnsRequestService(db_session)
    with (
        patch(
            "app.domains.cmdb.dns_request_service.get_approval_flow",
            new=AsyncMock(return_value=fake_flow),
        ),
        patch.object(svc, "_client", new=AsyncMock(return_value=fake_client)),
    ):
        with pytest.raises(AppException) as exc_info:
            await svc.retry_after_template_fix(
                request_id=str(row.id), actor=admin, trace_id="dns-loss-1",
            )
        assert exc_info.value.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN"

        result = await svc.retry_after_template_fix(
            request_id=str(row.id), actor=admin, trace_id="dns-loss-2",
        )

    assert result.feishu_instance_code == "dns-recovered-instance"
    assert provider_uuids == [provider_uuids[0], provider_uuids[0]]
    intent = await db_session.scalar(select(ApprovalSubmissionIntent).where(
        ApprovalSubmissionIntent.scope == f"cmdb_dns_retry:{row.id}",
    ))
    assert intent is not None and intent.state == "completed"


@pytest.mark.asyncio
async def test_retry_rejects_when_zone_revoked(db_session, integ):
    """If the admin revoked ``allow_dns_request`` on the zone after the
    original submit, retry should refuse — silently routing to a now-
    closed zone would surprise everyone.
    """
    admin = await _make_person(
        db_session, email="a6@test.com", role="global_admin",
    )
    row, _ar = await _make_failed_row(
        db_session, integ=integ, requester=admin,
    )
    # Flip the zone closed.
    zone = (
        await db_session.execute(
            select(CmdbDomain).where(
                CmdbDomain.integration_id == str(integ.id),
                CmdbDomain.domain_name == row.zone_name,
            ),
        )
    ).scalar_one()
    zone.allow_dns_request = False
    await db_session.commit()

    svc = DnsRequestService(db_session)
    with pytest.raises(ValidationError):
        await svc.retry_after_template_fix(
            request_id=str(row.id),
            actor=admin,
            trace_id="test",
        )