"""Regression: zone → authoritative-DNS-provider resolution.

Prod incident 2026-06-01: the admin DNS picker showed
"AliCloud (海外)" for ``tikv.io`` even though tikv.io has no
records on AliCloud at all. Root cause was a layered set of bugs
covered by this PR:

1. ``GET /cmdb/dns-zones`` returned ``list[str]`` — no
   integration mapping. The FE compensated with a buggy
   per-zone ``listDnsRecords?page_size=1`` round-trip; when the
   zone had zero records, the FE fell back to "first provider
   with ``dns_records`` capability", silently picking whichever
   integration happened to come first in the providers array.
2. ``GET /cmdb/domains/selectable-for-dns`` returned
   ``cmdb_domains.integration_id``, which is the **registrar**
   (populated by ``CloudProvider.list_domains`` →
   ``route53domains:ListDomains`` / Tencent 域名注册 /
   UCloud 域名注册), NOT the authoritative DNS provider. Self-
   service requests then carried the wrong ``integration_id``
   into the Feishu approval flow.
3. ``DnsRequestService.submit()`` validated the request against
   ``cmdb_domains.integration_id == body.integration_id`` — the
   registrar check — which let the wrong-tenant write proceed.

This file pins the corrected behaviour: the truth source for
"who hosts DNS for this zone" is ``cmdb_dns_records.integration_id``
(populated by ``_sync_dns_records`` which calls authoritative DNS
APIs). Both the admin picker and the employee picker route on it.
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.models import CmdbDnsRecord, CmdbDomain
from app.domains.cmdb.service import CmdbService
from app.integrations.models import Integration


def _integ(provider: str = "tencent", *, enabled: bool = True) -> Integration:
    slug = f"pytest-{provider}-{uuid.uuid4().hex[:6]}"
    return Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=slug,
        slug=slug,
        enabled=enabled,
        is_primary=False,
        capabilities=["dns_records"],
    )


@pytest.mark.asyncio
async def test_resolve_dns_integration_picks_record_count_winner(
    db_session: AsyncSession,
):
    """When multiple integrations have records for the same zone
    (transitional cutover, ICP-filing artefacts), the one with the
    most records wins."""
    big = _integ("tencent")
    small = _integ("alicloud_intl")
    db_session.add_all([big, small])
    await db_session.flush()

    # big has 3, small has 1 — same zone.
    for i in range(3):
        db_session.add(CmdbDnsRecord(
            integration_id=str(big.id), zone_name="hot.example",
            record_id=f"big-{i}", name=f"r{i}", record_type="A",
            value="1.1.1.1", ttl=60, status="enable",
        ))
    db_session.add(CmdbDnsRecord(
        integration_id=str(small.id), zone_name="hot.example",
        record_id="small-1", name="r0", record_type="A",
        value="2.2.2.2", ttl=60, status="enable",
    ))
    await db_session.commit()

    result = await CmdbService.resolve_dns_integration_for_zone(
        db_session, "hot.example",
    )
    assert result is not None
    integ_id, integ_name, provider = result
    assert integ_id == str(big.id)
    assert provider == "tencent"


@pytest.mark.asyncio
async def test_resolve_dns_integration_returns_none_for_no_records(
    db_session: AsyncSession,
):
    """A zone present only in ``cmdb_domains`` (registrar metadata)
    but with no synced records → no resolution."""
    integ = _integ("ucloud")
    db_session.add(integ)
    await db_session.flush()
    db_session.add(CmdbDomain(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        domain_name="orphan.example",
        status="active",
        allow_dns_request=True,
    ))
    await db_session.commit()

    result = await CmdbService.resolve_dns_integration_for_zone(
        db_session, "orphan.example",
    )
    # No records → no resolved DNS provider. The FE must show a
    # manual provider picker for this case.
    assert result is None


@pytest.mark.asyncio
async def test_list_dns_zones_returns_rich_entries(
    db_session: AsyncSession,
):
    """``GET /cmdb/dns-zones`` returns
    ``[{zone_name, integration_id, integration_name, provider,
    record_count}]`` so the FE picker no longer needs the buggy
    per-zone ``listDnsRecords?page_size=1`` round-trip."""
    integ = _integ("tencent")
    registrar = _integ("ucloud")
    db_session.add_all([integ, registrar])
    await db_session.flush()
    db_session.add_all([
        CmdbDnsRecord(
            integration_id=str(integ.id), zone_name="resolved.example",
            record_id=f"r{i}", name=f"name{i}", record_type="A",
            value="1.1.1.1", ttl=60, status="enable",
        )
        for i in range(2)
    ])
    # Registrar-only zone — present in cmdb_domains, absent from
    # cmdb_dns_records. integration_id on the DB row points at the
    # registrar (a real prod case: UCloud-registered tikv.io).
    db_session.add(CmdbDomain(
        id=uuid.uuid4(),
        integration_id=str(registrar.id),
        domain_name="registrar-only.example",
        status="active",
        allow_dns_request=True,
    ))
    await db_session.commit()

    zones = await CmdbService.list_dns_zones(db_session)
    by_zone = {z["zone_name"]: z for z in zones}

    # Zone WITH records → resolved provider.
    r = by_zone["resolved.example"]
    assert r["integration_id"] == str(integ.id)
    assert r["provider"] == "tencent"
    assert r["record_count"] == 2

    # Registrar-only zone → integration_id MUST be None so the FE
    # knows to ask the admin. Returning the registrar integration
    # here is the bug that produced "tikv.io → AliCloud (海外)".
    r2 = by_zone["registrar-only.example"]
    assert r2["integration_id"] is None
    assert r2["provider"] is None
    assert r2["record_count"] == 0


@pytest.mark.asyncio
async def test_selectable_for_dns_returns_resolved_dns_provider(
    client, admin_user, db_session,
):
    """``GET /cmdb/domains/selectable-for-dns`` returns the DNS
    provider (resolved from ``cmdb_dns_records``), not the
    registrar that ``cmdb_domains.integration_id`` historically
    carried."""
    # Registrar — UCloud holds tikv.io in cmdb_domains.
    registrar = _integ("ucloud")
    # Authoritative DNS — DNSPod (different tenant) holds the records.
    dns = _integ("tencent")
    db_session.add_all([registrar, dns])
    await db_session.flush()

    db_session.add(CmdbDomain(
        id=uuid.uuid4(),
        integration_id=str(registrar.id),
        domain_name="example.test",
        status="active",
        allow_dns_request=True,
    ))
    db_session.add(CmdbDnsRecord(
        integration_id=str(dns.id),
        zone_name="example.test",
        record_id="r1", name="www", record_type="A",
        value="1.1.1.1", ttl=60, status="enable",
    ))
    await db_session.commit()

    _, token = admin_user
    resp = await client.get(
        "/api/v1/cmdb/domains/selectable-for-dns",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    rows = resp.json()["data"]
    by_zone = {r["domain_name"]: r for r in rows}
    assert "example.test" in by_zone
    # Bug fix: must be the DNS provider's id (DNSPod), NOT the
    # registrar's id (UCloud).
    assert by_zone["example.test"]["integration_id"] == str(dns.id), (
        "selectable-for-dns must return the resolved DNS provider, "
        "not the registrar from cmdb_domains"
    )
    assert by_zone["example.test"]["provider"] == "tencent"


@pytest.mark.asyncio
async def test_selectable_for_dns_omits_resolution_for_unsynced_zones(
    client, admin_user, db_session,
):
    """A zone registered + ``allow_dns_request=True`` but with no
    synced records still appears, with ``integration_id=None`` so
    the FE filters it out (employees can't seed cold)."""
    registrar = _integ("ucloud")
    db_session.add(registrar)
    await db_session.flush()
    db_session.add(CmdbDomain(
        id=uuid.uuid4(),
        integration_id=str(registrar.id),
        domain_name="cold.test",
        status="active",
        allow_dns_request=True,
    ))
    await db_session.commit()

    _, token = admin_user
    resp = await client.get(
        "/api/v1/cmdb/domains/selectable-for-dns",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200
    rows = resp.json()["data"]
    by_zone = {r["domain_name"]: r for r in rows}
    assert by_zone["cold.test"]["integration_id"] is None


@pytest.mark.asyncio
async def test_resolve_dns_integration_excludes_soft_deleted_records(
    db_session: AsyncSession,
):
    """Soft-deleted records must not contribute to the count tie-break.

    A zone whose only "winner" is a soft-deleted historical record set
    should resolve to whatever other tenant has live records, not the
    ghost. Without this guard a sweep race could route writes to a
    tenant whose records have all been pruned.
    """
    ghost = _integ("alicloud_intl")
    live = _integ("tencent")
    db_session.add_all([ghost, live])
    await db_session.flush()
    # ghost: 5 soft-deleted records.
    from datetime import UTC, datetime
    now = datetime.now(UTC).replace(tzinfo=None)
    for i in range(5):
        db_session.add(CmdbDnsRecord(
            integration_id=str(ghost.id), zone_name="ghosted.test",
            record_id=f"g{i}", name=f"g{i}", record_type="A",
            value="1.1.1.1", ttl=60, status="enable",
            deleted_at=now,
        ))
    # live: 1 active record.
    db_session.add(CmdbDnsRecord(
        integration_id=str(live.id), zone_name="ghosted.test",
        record_id="l1", name="l1", record_type="A",
        value="2.2.2.2", ttl=60, status="enable",
    ))
    await db_session.commit()

    result = await CmdbService.resolve_dns_integration_for_zone(
        db_session, "ghosted.test",
    )
    assert result is not None
    assert result[0] == str(live.id), (
        "soft-deleted records must not win the count tie-break"
    )


@pytest.mark.asyncio
async def test_resolve_dns_integration_tie_break_is_deterministic(
    db_session: AsyncSession,
):
    """Equal record counts: resolver picks ONE deterministically and
    keeps picking the same one across calls.

    The guarantee admins rely on: a zone's displayed DNS provider
    doesn't flip between page refreshes. We don't promise *which* one
    wins in a tie (DB-collation-specific), only that it doesn't
    oscillate.
    """
    a = _integ("alicloud_intl")
    b = _integ("tencent")
    db_session.add_all([a, b])
    await db_session.flush()
    for i in range(3):
        db_session.add_all([
            CmdbDnsRecord(
                integration_id=str(a.id), zone_name="tied.test",
                record_id=f"a{i}", name=f"a{i}", record_type="A",
                value="1.1.1.1", ttl=60, status="enable",
            ),
            CmdbDnsRecord(
                integration_id=str(b.id), zone_name="tied.test",
                record_id=f"b{i}", name=f"b{i}", record_type="A",
                value="2.2.2.2", ttl=60, status="enable",
            ),
        ])
    await db_session.commit()

    runs = [
        await CmdbService.resolve_dns_integration_for_zone(
            db_session, "tied.test",
        )
        for _ in range(3)
    ]
    assert all(r is not None for r in runs)
    winners = {r[0] for r in runs}
    assert len(winners) == 1, (
        f"tie-break should be deterministic, got rotating winners: {winners}"
    )


@pytest.mark.asyncio
async def test_dns_zones_excludes_disabled_tenants_from_resolution(
    db_session: AsyncSession,
):
    """A stale historical record on a disabled tenant must not
    silently become the routing target for that zone — that's a
    documented attack-surface boundary we keep at every cloud-list
    endpoint via ``_active_cloud_integration_ids``."""
    disabled = _integ("alicloud_intl", enabled=False)
    enabled = _integ("tencent")
    db_session.add_all([disabled, enabled])
    await db_session.flush()
    db_session.add(CmdbDnsRecord(
        integration_id=str(disabled.id), zone_name="ambiguous.test",
        record_id="ghost", name="www", record_type="A",
        value="1.1.1.1", ttl=60, status="enable",
    ))
    db_session.add(CmdbDnsRecord(
        integration_id=str(enabled.id), zone_name="ambiguous.test",
        record_id="live", name="www2", record_type="A",
        value="2.2.2.2", ttl=60, status="enable",
    ))
    await db_session.commit()

    zones = await CmdbService.list_dns_zones(db_session)
    by_zone = {z["zone_name"]: z for z in zones}
    # Resolver must pick the enabled tenant despite the disabled one
    # having an equal record count (or even more).
    assert by_zone["ambiguous.test"]["integration_id"] == str(enabled.id)