"""Regression coverage for the 2026-05-28 legacy-ID sweep-protection fix.

Prod incident: a single UCloud sync on 2026-04-24 silently soft-deleted
116 servers because the legacy-CMDB migration script wrote
``instance_id`` as ``legacy-{vendor}-{src_id}-{raw_iid}`` and the live
UCloud driver writes the raw ``uhost-...`` ID. The unique key
``(integration_id, instance_id)`` never collided → live sync created
shadow rows AND ``_safe_sweep`` marked every ``legacy-*`` row missing.

Fixes verified here:

1. **Sweep ignores rows the driver has never observed.** Before the
   fix, ``last_seen_sync_job_id IS NULL`` matched the sweep predicate;
   now it doesn't. Migration / manual rows that never participated in
   any sync are protected.
2. **``existing`` count narrows to ever-synced rows.** Otherwise the
   drop-ratio guard trips on every sync of a freshly-imported batch
   (those NULL rows inflate ``existing`` but the driver can never
   match them).
3. **IP search** on ``cmdb_servers`` covers ``public_ips`` /
   ``private_ips`` JSON arrays (the original prod use-case:
   "find 10.9.84.114").
4. **Recycle-bin list + restore** for soft-deleted rows across the 5
   admin-curated resource types.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

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

from app.core.exceptions import AppException
from app.domains.cmdb.models import (
    CmdbCdnDomain,
    CmdbDomain,
    CmdbServer,
)
from app.domains.cmdb.providers.exceptions import ProviderTransientError
from app.domains.cmdb.service import CmdbService
from app.domains.cmdb.sync import _safe_sweep
from app.integrations.models import Integration


async def _make_integration(db: AsyncSession, slug_prefix: str = "leg") -> Integration:
    iid = uuid.uuid4()
    integ = Integration(
        id=iid,
        kind="cloud",
        provider="ucloud",
        name=f"{slug_prefix}-test-{str(iid)[:6]}",
        slug=f"{slug_prefix}-{str(iid)[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(integ)
    await db.commit()
    return integ


async def _make_server(
    db: AsyncSession,
    integration: Integration,
    *,
    instance_id: str,
    last_seen_sync_job_id: str | None = None,
    public_ips: list[str] | None = None,
    private_ips: list[str] | None = None,
    deleted_at: datetime | None = None,
) -> CmdbServer:
    row = CmdbServer(
        id=str(uuid.uuid4()),
        integration_id=str(integration.id),
        instance_id=instance_id,
        name=instance_id,
        status="running",
        region="cn-bj2",
        zone=None,
        vcpu=2,
        memory_mb=4096,
        public_ips=public_ips or [],
        private_ips=private_ips or [],
        os_type=None,
        image_id=None,
        tags={},
        raw_data={},
        last_seen_sync_job_id=last_seen_sync_job_id,
        last_synced_at=datetime.now(UTC),
        deleted_at=deleted_at,
    )
    db.add(row)
    await db.commit()
    return row


# ── Sweep-protection coverage ───────────────────────────────────────


@pytest.mark.asyncio
async def test_sweep_skips_rows_never_observed_by_sync(
    db_session: AsyncSession,
):
    """The exact prod incident shape: 3 migration rows never touched by
    sync (``last_seen_sync_job_id IS NULL``) coexist with 2 live-sync
    rows. The current sync upserted both live rows. Pre-fix, the sweep
    would mark the 3 migration rows ``deleted_at=NOW()`` because they
    matched ``last_seen_sync_job_id IS NULL`` in the OR predicate.
    Post-fix, they're left alone.
    """
    integ = await _make_integration(db_session, "incident-shape")
    new_job_id = str(uuid.uuid4())

    # 3 migration rows — NEVER touched by any sync. These are the
    # 116-UCloud-server-incident analogs.
    for i in range(3):
        await _make_server(
            db_session, integ,
            instance_id=f"legacy-2-{i}-uhost-incident-{i}",
            last_seen_sync_job_id=None,
        )
    # 2 live-sync rows freshly stamped by the current job — keep alive.
    for i in range(2):
        await _make_server(
            db_session, integ,
            instance_id=f"uhost-live-{i}",
            last_seen_sync_job_id=new_job_id,
        )

    # Sweep with current upserted=2 (the two live rows). With the fix,
    # nothing should be deleted. Pre-fix would have swept the 3
    # migration rows.
    deleted = await _safe_sweep(
        db_session,
        model=CmdbServer,
        integration_id=str(integ.id),
        job_id=new_job_id,
        upserted=2,
        capability_label="servers",
        capability_key="servers",
    )
    await db_session.commit()
    assert deleted == 0, (
        f"sweep should not touch never-observed migration rows; "
        f"got {deleted} deletions (regression: legacy-ID sweep bug back)"
    )


@pytest.mark.asyncio
async def test_sweep_still_catches_genuinely_orphaned_rows(
    db_session: AsyncSession,
):
    """Counter-test: a row that WAS once observed (its
    last_seen_sync_job_id points to an older job) but isn't matched by
    the current sync IS swept. Otherwise the fix would also break the
    normal "instance terminated in cloud → soft-delete in Helix" flow.
    """
    integ = await _make_integration(db_session, "orphan")
    old_job_id = str(uuid.uuid4())
    new_job_id = str(uuid.uuid4())

    # One stale row tagged with an old sync job — genuinely missing from
    # the current cloud response. Should be swept.
    await _make_server(
        db_session, integ,
        instance_id="uhost-orphan",
        last_seen_sync_job_id=old_job_id,
    )
    # Four fresh rows tagged with the current job — keep alive.  With the
    # one stale row this is the exact default 5→4 (20%) boundary, so normal
    # orphan cleanup remains covered without weakening the drop guard.
    for index in range(4):
        await _make_server(
            db_session, integ,
            instance_id=f"uhost-fresh-{index}",
            last_seen_sync_job_id=new_job_id,
        )

    deleted = await _safe_sweep(
        db_session,
        model=CmdbServer,
        integration_id=str(integ.id),
        job_id=new_job_id,
        upserted=4,
        capability_label="servers",
        capability_key="servers",
    )
    await db_session.commit()
    assert deleted == 1, (
        "expected the 1 stale-job-id row to be swept; got "
        f"{deleted}. The fix must not over-protect: rows observed by "
        "an older sync but missing today are genuine orphans."
    )


@pytest.mark.asyncio
async def test_existing_count_narrows_to_ever_synced_rows(
    db_session: AsyncSession,
):
    """``existing`` (the baseline-fallback denominator) counts only rows
    with ``last_seen_sync_job_id IS NOT NULL``. Without this, a tenant
    with 100 freshly-imported migration rows + 5 live-sync rows would
    see ``existing=105``, ``upserted=5`` → 95% drop → guard trips on
    every sync forever (the migration rows can never close the gap
    because the driver doesn't know about them).

    Verifies by exercising the drop-ratio guard at the boundary: 100
    NULL-job rows + 5 alive in current sync. ``existing`` should be 0
    (no ever-synced rows yet), no prior sync, so baseline=0, no guard
    trip, sweep proceeds (and sweeps 0 because no never-observed rows
    are candidates anyway).
    """
    integ = await _make_integration(db_session, "exist-narrow")
    new_job_id = str(uuid.uuid4())

    # 10 migration rows never touched by any sync.
    for i in range(10):
        await _make_server(
            db_session, integ,
            instance_id=f"legacy-2-{i}-uhost-x{i}",
            last_seen_sync_job_id=None,
        )

    # No prior CmdbSyncJob, so baseline falls back to ``existing``
    # which must be 0 (no ever-synced rows). Pre-fix, ``existing``
    # would have been 10 → upserted=1 → 90% drop → guard trips → no
    # sweep AND, more importantly for the user-facing bug, the next
    # forensic question ("why did this sync abort?") would have led to
    # a force-sweep that wiped the 10 migration rows.
    deleted = await _safe_sweep(
        db_session,
        model=CmdbServer,
        integration_id=str(integ.id),
        job_id=new_job_id,
        upserted=1,
        capability_label="servers",
        capability_key="servers",
    )
    await db_session.commit()
    assert deleted == 0
    # Verify the 10 migration rows survived.
    from sqlalchemy import func, select  # noqa: PLC0415
    n_alive = (await db_session.execute(
        select(func.count(CmdbServer.id)).where(
            CmdbServer.integration_id == str(integ.id),
            CmdbServer.deleted_at.is_(None),
        ),
    )).scalar() or 0
    assert n_alive == 10, (
        f"expected all 10 migration rows alive; got {n_alive}"
    )


@pytest.mark.asyncio
async def test_sweep_guard_uses_unique_rows_not_raw_duplicate_upserts(
    db_session: AsyncSession,
):
    """A duplicate provider page must not inflate the destructive floor.

    The driver loop can report 100 successful upserts while repeatedly
    updating one unique row.  With five active rows in Helix, trusting that
    throughput counter would sweep the four rows omitted from the provider
    snapshot.  The guard must instead see the one row stamped by this job and
    fail closed (an 80% drop at the default 20% tolerance).
    """
    integ = await _make_integration(db_session, "duplicate-upserts")
    old_job_id = str(uuid.uuid4())
    new_job_id = str(uuid.uuid4())

    await _make_server(
        db_session,
        integ,
        instance_id="uhost-duplicated",
        last_seen_sync_job_id=new_job_id,
    )
    for index in range(4):
        await _make_server(
            db_session,
            integ,
            instance_id=f"uhost-omitted-{index}",
            last_seen_sync_job_id=old_job_id,
        )

    with pytest.raises(ProviderTransientError, match="persisted 1 unique"):
        await _safe_sweep(
            db_session,
            model=CmdbServer,
            integration_id=str(integ.id),
            job_id=new_job_id,
            upserted=100,
            capability_label="servers",
            capability_key="servers",
        )

    assert all(row.deleted_at is None for row in (
        await db_session.execute(
            select(CmdbServer).where(
                CmdbServer.integration_id == str(integ.id),
            )
        )
    ).scalars())


@pytest.mark.asyncio
async def test_sweep_observed_count_is_job_and_tenant_isolated(
    db_session: AsyncSession,
):
    """Other jobs and tenants cannot inflate the current snapshot."""
    target = await _make_integration(db_session, "target-isolation")
    sibling = await _make_integration(db_session, "sibling-isolation")
    current_job_id = str(uuid.uuid4())
    competing_job_id = str(uuid.uuid4())

    await _make_server(
        db_session, target,
        instance_id="target-current",
        last_seen_sync_job_id=current_job_id,
    )
    # Four rows were concurrently stamped by another job. Production
    # serializes these with cloud_sync:<integration>; this mutation proves
    # the SQL itself still fails closed if that invariant is ever broken.
    for index in range(4):
        await _make_server(
            db_session, target,
            instance_id=f"target-competing-{index}",
            last_seen_sync_job_id=competing_job_id,
        )
    # Same current job id on another tenant must not contribute either.
    for index in range(5):
        await _make_server(
            db_session, sibling,
            instance_id=f"sibling-current-{index}",
            last_seen_sync_job_id=current_job_id,
        )

    with pytest.raises(ProviderTransientError, match="persisted 1 unique"):
        await _safe_sweep(
            db_session,
            model=CmdbServer,
            integration_id=str(target.id),
            job_id=current_job_id,
            upserted=100,
            capability_label="servers",
            capability_key="servers",
        )


@pytest.mark.asyncio
async def test_sweep_does_not_count_soft_deleted_current_job_rows(
    db_session: AsyncSession,
):
    """A current-job stamp is insufficient unless the row is active."""
    integ = await _make_integration(db_session, "soft-delete-isolation")
    current_job_id = str(uuid.uuid4())
    old_job_id = str(uuid.uuid4())
    for index in range(4):
        await _make_server(
            db_session, integ,
            instance_id=f"active-old-{index}",
            last_seen_sync_job_id=old_job_id,
        )
    await _make_server(
        db_session, integ,
        instance_id="deleted-current",
        last_seen_sync_job_id=current_job_id,
        deleted_at=datetime.now(UTC),
    )

    with pytest.raises(ProviderTransientError, match="returned 0 rows"):
        await _safe_sweep(
            db_session,
            model=CmdbServer,
            integration_id=str(integ.id),
            job_id=current_job_id,
            upserted=1,
            capability_label="servers",
            capability_key="servers",
        )


# ── IP search coverage ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_list_servers_search_by_private_ip_finds_match(
    db_session: AsyncSession,
):
    """The user incident IP 10.9.84.114 must be findable via the
    standard server-search input. Pre-fix the search only LIKE-matched
    name + instance_id, so an admin pasting an IP got zero results."""
    integ = await _make_integration(db_session, "ip-search")
    sync_id = str(uuid.uuid4())
    await _make_server(
        db_session, integ,
        instance_id="uhost-dlp",
        last_seen_sync_job_id=sync_id,
        public_ips=["106.75.47.22"],
        private_ips=["10.9.84.114"],
    )
    await _make_server(
        db_session, integ,
        instance_id="uhost-other",
        last_seen_sync_job_id=sync_id,
        public_ips=["1.2.3.4"],
        private_ips=["10.0.0.1"],
    )
    rows, total = await CmdbService.list_servers(
        db_session,
        integration_id=str(integ.id),
        search="10.9.84.114",
    )
    assert total == 1
    assert rows[0].instance_id == "uhost-dlp"


@pytest.mark.asyncio
async def test_list_servers_search_by_public_ip_finds_match(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session, "ip-search-public")
    sync_id = str(uuid.uuid4())
    await _make_server(
        db_session, integ,
        instance_id="uhost-public-search",
        last_seen_sync_job_id=sync_id,
        public_ips=["106.75.47.22"],
        private_ips=["10.0.0.99"],
    )
    rows, total = await CmdbService.list_servers(
        db_session,
        integration_id=str(integ.id),
        search="106.75.47",  # partial — admin remembers prefix
    )
    assert total == 1
    assert rows[0].instance_id == "uhost-public-search"


# ── Recycle bin (list + restore) coverage ───────────────────────────


@pytest.mark.asyncio
async def test_list_deleted_servers_returns_only_soft_deleted(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session, "recyc-list")
    sync_id = str(uuid.uuid4())
    # Alive row — must NOT appear in recycle-bin list.
    await _make_server(
        db_session, integ,
        instance_id="uhost-alive", last_seen_sync_job_id=sync_id,
    )
    # Soft-deleted row — must appear.
    deleted_at = datetime.now(UTC)
    await _make_server(
        db_session, integ,
        instance_id="uhost-tombstoned", last_seen_sync_job_id=sync_id,
        deleted_at=deleted_at,
    )
    rows, total = await CmdbService.list_deleted_servers(
        db_session, integration_id=str(integ.id),
    )
    assert total == 1
    assert rows[0].instance_id == "uhost-tombstoned"


@pytest.mark.asyncio
async def test_restore_resource_clears_deleted_at_and_returns_forensics(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session, "recyc-restore")
    sync_id = str(uuid.uuid4())
    deleted_at = datetime.now(UTC)
    server = await _make_server(
        db_session, integ,
        instance_id="uhost-restorable", last_seen_sync_job_id=sync_id,
        deleted_at=deleted_at,
    )
    result = await CmdbService.restore_resource(
        db_session,
        model=CmdbServer,
        resource_id=str(server.id),
    )
    assert result is not None
    restored, forensics = result
    assert restored.deleted_at is None
    # Forensic snapshot captures the prior tombstone timestamp and
    # the last sync_job_id that touched the row — the two pieces of
    # info an investigator needs to reconcile "was this restore
    # reversing a real cloud-side decommission, or a sweep accident?".
    assert forensics["prior_deleted_at"] is not None
    assert forensics["prior_last_seen_sync_job_id"] == sync_id


@pytest.mark.asyncio
async def test_restore_resource_409_on_alive_row(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session, "recyc-409")
    sync_id = str(uuid.uuid4())
    server = await _make_server(
        db_session, integ,
        instance_id="uhost-alive-409", last_seen_sync_job_id=sync_id,
    )
    with pytest.raises(AppException) as exc_info:
        await CmdbService.restore_resource(
            db_session,
            model=CmdbServer,
            resource_id=str(server.id),
        )
    assert exc_info.value.status_code == 409
    assert exc_info.value.code == "CMDB_RESOURCE_NOT_DELETED"


@pytest.mark.asyncio
async def test_restore_resource_returns_none_for_missing_id(
    db_session: AsyncSession,
):
    result = await CmdbService.restore_resource(
        db_session,
        model=CmdbServer,
        resource_id=str(uuid.uuid4()),
    )
    assert result is None


# ── UCloud multi-project coverage ───────────────────────────────────


def test_ucloud_credentials_parse_project_ids_csv():
    """The plural ``project_ids`` field wins over the singular
    ``project_id``. Verifies the parsing layer in isolation — the
    actual per-project iteration is harder to mock cleanly without a
    full ``_call`` stub, but parse coverage is the highest-value
    regression for the prod multi-project blind spot."""
    from app.domains.cmdb.providers.base import ProviderCredentials  # noqa: PLC0415
    from app.domains.cmdb.providers.ucloud import UCloudProvider  # noqa: PLC0415

    creds = ProviderCredentials(
        provider_id="test-prov-1",
        provider_type="ucloud",
        account_alias="test-account",
        fields={
            "public_key": "pk",
            "private_key": "sk",
            "project_ids": "org-aaa, org-bbb,  ,org-ccc",
            "project_id": "org-legacy-fallback",  # should be IGNORED
        },
    )
    provider = UCloudProvider(creds)
    try:
        assert provider._project_ids == ["org-aaa", "org-bbb", "org-ccc"]
    finally:
        import asyncio  # noqa: PLC0415
        asyncio.run(provider.close())


def test_ucloud_credentials_fall_back_to_singular_project_id():
    from app.domains.cmdb.providers.base import ProviderCredentials  # noqa: PLC0415
    from app.domains.cmdb.providers.ucloud import UCloudProvider  # noqa: PLC0415

    creds = ProviderCredentials(
        provider_id="test-prov-2",
        provider_type="ucloud",
        account_alias="test-account",
        fields={
            "public_key": "pk",
            "private_key": "sk",
            "project_id": "org-only",
        },
    )
    provider = UCloudProvider(creds)
    try:
        assert provider._project_ids == ["org-only"]
    finally:
        import asyncio  # noqa: PLC0415
        asyncio.run(provider.close())


def test_ucloud_credentials_no_project_uses_none_sentinel():
    """Both fields empty → ``_project_ids = [None]``, which signals
    ``_call`` to omit the ``ProjectId`` parameter entirely (UCloud
    falls back to the AK's default project)."""
    from app.domains.cmdb.providers.base import ProviderCredentials  # noqa: PLC0415
    from app.domains.cmdb.providers.ucloud import UCloudProvider  # noqa: PLC0415

    creds = ProviderCredentials(
        provider_id="test-prov-3",
        provider_type="ucloud",
        account_alias="test-account",
        fields={"public_key": "pk", "private_key": "sk"},
    )
    provider = UCloudProvider(creds)
    try:
        assert provider._project_ids == [None]
    finally:
        import asyncio  # noqa: PLC0415
        asyncio.run(provider.close())


# ── Sibling-resource recycle-bin coverage ───────────────────────────


@pytest.mark.asyncio
async def test_list_deleted_domains_returns_only_soft_deleted(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session, "recyc-dom")
    # Alive
    db_session.add(CmdbDomain(
        id=str(uuid.uuid4()),
        integration_id=str(integ.id),
        domain_name="alive.example.com",
        registrar="legacy",
        status="active",
    ))
    # Tombstoned
    db_session.add(CmdbDomain(
        id=str(uuid.uuid4()),
        integration_id=str(integ.id),
        domain_name="tombstoned.example.com",
        registrar="legacy",
        status="active",
        deleted_at=datetime.now(UTC),
    ))
    await db_session.commit()
    rows, total = await CmdbService.list_deleted_domains(
        db_session, integration_id=str(integ.id),
    )
    assert total == 1
    assert rows[0].domain_name == "tombstoned.example.com"


@pytest.mark.asyncio
async def test_list_deleted_cdn_search_matches_cname(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session, "recyc-cdn")
    db_session.add(CmdbCdnDomain(
        id=str(uuid.uuid4()),
        integration_id=str(integ.id),
        domain_name="cdn-tomb.example.com",
        cname="abc123.ucdn.com",
        status="offline",
        source_type="domain",
        source_address="origin.example.com",
        is_https=True,
        deleted_at=datetime.now(UTC),
    ))
    await db_session.commit()
    rows, total = await CmdbService.list_deleted_cdn_domains(
        db_session, integration_id=str(integ.id), search="abc123",
    )
    assert total == 1
    assert rows[0].cname == "abc123.ucdn.com"