"""Tests for per-integration scheduled_jobs provisioning.

Covers the contract between IntegrationService CRUD and the
``scheduled_jobs`` rows driven by cloud integrations:

  * create → one scheduled_jobs row per supported capability + one
    billing row, all pointed at the new integration_id;
  * update (capabilities changed) → diff adds/removes rows;
  * delete → all rows for that integration_id are removed;
  * renaming → existing rows' display_name refreshes but cadence and
    enabled flag are preserved (admin-owned overrides).
"""
from __future__ import annotations

import uuid

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

from app.domains.cmdb.scheduled_jobs_provisioning import (
    BILLING_DAILY_HANDLER_KEY,
    BILLING_HANDLER_KEY,
    COST_FACTS_HANDLER_KEY,
    EFFICIENCY_HANDLER_KEY,
    GRANTS_HANDLER_KEY,
    HANDLER_KEY,
    ROLES_HANDLER_KEY,
    SCOPES_HANDLER_KEY,
    SERVER_METRICS_HANDLER_KEY,
    reconcile_all_cloud_integrations,
    remove_jobs_for_integration,
    sync_jobs_for_integration,
)
from app.integrations.models import Integration
from app.scheduled_jobs.models import ScheduledJob

_ALL_HANDLER_KEYS = [
    HANDLER_KEY, BILLING_HANDLER_KEY,
    SCOPES_HANDLER_KEY, ROLES_HANDLER_KEY, GRANTS_HANDLER_KEY,
    BILLING_DAILY_HANDLER_KEY, COST_FACTS_HANDLER_KEY,
    SERVER_METRICS_HANDLER_KEY, EFFICIENCY_HANDLER_KEY,
]


async def _all_jobs_for(db: AsyncSession, integration_id: str) -> list[ScheduledJob]:
    rows = (await db.execute(
        select(ScheduledJob).where(
            ScheduledJob.handler_key.in_(_ALL_HANDLER_KEYS),
        )
    )).scalars().all()
    return [r for r in rows if (r.config or {}).get("integration_id") == integration_id]


@pytest.mark.asyncio
async def test_sync_creates_one_row_per_capability_plus_billing(
    db_session: AsyncSession,
):
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="aws-test",
        slug=f"aws-test-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers", "iam"],
    )
    db_session.add(integ)
    await db_session.commit()

    summary = await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    rows = await _all_jobs_for(db_session, str(integ.id))
    # 2 caps + 1 monthly billing + 5 lifecycle
    # (scopes/roles/grants/billing_daily/cost_facts) + 2 efficiency
    # (server_metrics/efficiency — AWS declares the metrics capability)
    assert len(rows) == 10
    handler_keys = {r.handler_key for r in rows}
    assert HANDLER_KEY in handler_keys
    assert BILLING_HANDLER_KEY in handler_keys
    assert SCOPES_HANDLER_KEY in handler_keys
    assert ROLES_HANDLER_KEY in handler_keys
    assert GRANTS_HANDLER_KEY in handler_keys
    assert BILLING_DAILY_HANDLER_KEY in handler_keys
    assert COST_FACTS_HANDLER_KEY in handler_keys
    assert SERVER_METRICS_HANDLER_KEY in handler_keys
    assert EFFICIENCY_HANDLER_KEY in handler_keys
    assert summary["desired_count"] == 10


@pytest.mark.asyncio
async def test_sync_diffs_when_capabilities_change(db_session: AsyncSession):
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="diff-test",
        slug=f"diff-test-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()

    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()
    initial = await _all_jobs_for(db_session, str(integ.id))

    # Admin adds iam
    integ.capabilities = ["servers", "iam"]
    summary = await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    after_add = await _all_jobs_for(db_session, str(integ.id))
    assert len(after_add) == len(initial) + 1
    assert any("iam" in s for s in summary["added"])

    # Admin drops servers
    integ.capabilities = ["iam"]
    summary2 = await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    after_drop = await _all_jobs_for(db_session, str(integ.id))
    assert len(after_drop) == len(after_add) - 1
    assert any("servers" in s for s in summary2["removed"])


@pytest.mark.asyncio
async def test_sync_preserves_admin_cadence_override(db_session: AsyncSession):
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="override-test",
        slug=f"over-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()

    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    # Admin tweaks cadence + disables one row
    rows = await _all_jobs_for(db_session, str(integ.id))
    servers_row = next(r for r in rows if r.handler_key == HANDLER_KEY)
    servers_row.cron_expr = "15 */6 * * *"
    servers_row.enabled = False
    await db_session.commit()

    # Re-run provision (e.g. after a name refresh)
    integ.name = "override-test-renamed"
    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    rows_after = await _all_jobs_for(db_session, str(integ.id))
    refreshed = next(r for r in rows_after if r.handler_key == HANDLER_KEY)
    # Display name reflects rename
    assert "override-test-renamed" in refreshed.display_name
    # But admin overrides stuck
    assert refreshed.cron_expr == "15 */6 * * *"
    assert refreshed.enabled is False


@pytest.mark.asyncio
async def test_remove_drops_all_integration_rows(db_session: AsyncSession):
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="remove-test",
        slug=f"rm-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers", "iam"],
    )
    db_session.add(integ)
    await db_session.commit()
    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    count = await remove_jobs_for_integration(db_session, str(integ.id))
    await db_session.commit()
    # 2 caps + 1 monthly billing + 5 lifecycle
    # (scopes/roles/grants/billing_daily/cost_facts) + 2 efficiency
    assert count == 10

    leftover = await _all_jobs_for(db_session, str(integ.id))
    assert leftover == []


@pytest.mark.asyncio
async def test_reconcile_backfills_missing_lifecycle_rows(
    db_session: AsyncSession,
):
    """Regression guard for the PR #24 prod-drift incident: a new
    lifecycle handler was added in code but existing integrations in
    prod never got their per-integration row until someone touched
    the integration. Worker boot must reconcile automatically.

    We simulate the drift by (a) creating an integration with all
    lifecycle rows, (b) deleting the grants row by hand to mimic a
    pre-existing integration that missed the new handler, then
    verifying the reconcile call restores it.
    """
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="drift-test",
        slug=f"drift-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()
    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    # Simulate pre-PR-24 state: grants row doesn't exist yet.
    rows = await _all_jobs_for(db_session, str(integ.id))
    grants_row = next(r for r in rows if r.handler_key == GRANTS_HANDLER_KEY)
    await db_session.delete(grants_row)
    await db_session.commit()

    summary = await reconcile_all_cloud_integrations(db_session)

    assert summary["integrations_scanned"] >= 1
    assert summary["rows_added"] >= 1
    rows_after = await _all_jobs_for(db_session, str(integ.id))
    handler_keys_after = {r.handler_key for r in rows_after}
    assert GRANTS_HANDLER_KEY in handler_keys_after


@pytest.mark.asyncio
async def test_reconcile_is_idempotent(db_session: AsyncSession):
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="idem-test",
        slug=f"idem-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()
    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    first = await reconcile_all_cloud_integrations(db_session)
    second = await reconcile_all_cloud_integrations(db_session)
    # Second run must be a no-op for this integration.
    assert second["rows_added"] == 0
    assert second["rows_removed"] == 0
    # Both runs must have scanned at least this integration.
    assert first["integrations_scanned"] >= 1


@pytest.mark.asyncio
# Both params must be COLLECTION-STABLE literals: a value generated at collection
# time (e.g. uuid4()) gives every pytest-xdist worker a different test id and
# breaks `-n 4` with "Different tests were collected". The first param only has
# to be a syntactically valid UUID that is absent from the DB.
@pytest.mark.parametrize(
    "orphan_owner_id",
    ["00000000-0000-4000-8000-0000deadbeef", "legacy-invalid-id"],
)
async def test_sync_adopts_orphan_rows_after_connector_recreation(
    db_session: AsyncSession,
    orphan_owner_id: str,
):
    """A deleted connector's disabled cron is reused without losing overrides."""
    slug = f"recreated-{uuid.uuid4().hex[:6]}"
    current = Integration(
        id=uuid.uuid4(), kind="cloud", provider="ksyun",
        name="recreated connector", slug=slug, enabled=False,
        is_primary=False, capabilities=[],
    )
    orphan = ScheduledJob(
        id=uuid.uuid4(), name=f"cmdb.scopes.{slug}",
        display_name="orphan", description="legacy row",
        handler_key=SCOPES_HANDLER_KEY, trigger_type="cron",
        interval_seconds=None, cron_expr="7 2 * * *", timezone="Asia/Shanghai",
        enabled=False, config={"integration_id": orphan_owner_id},
    )
    db_session.add_all([current, orphan])
    await db_session.commit()

    summary = await sync_jobs_for_integration(db_session, current)
    await db_session.commit()

    await db_session.refresh(orphan)
    assert summary["adopted_orphan_names"] == [orphan.name]
    assert orphan.config["integration_id"] == str(current.id)
    assert orphan.enabled is False
    assert orphan.cron_expr == "7 2 * * *"


@pytest.mark.asyncio
async def test_reconcile_isolates_a_live_name_collision(
    db_session: AsyncSession,
):
    """One corrupt connector cannot leave the boot reconcile session aborted."""
    owner = Integration(
        id=uuid.uuid4(), kind="cloud", provider="ksyun",
        name="live owner", slug=f"owner-{uuid.uuid4().hex[:6]}",
        enabled=False, is_primary=False, capabilities=[],
    )
    target = Integration(
        id=uuid.uuid4(), kind="cloud", provider="ksyun",
        name="collision target", slug=f"target-{uuid.uuid4().hex[:6]}",
        enabled=False, is_primary=False, capabilities=[],
    )
    collision = ScheduledJob(
        id=uuid.uuid4(), name=f"cmdb.scopes.{target.slug}",
        display_name="foreign live row", description="collision",
        handler_key=SCOPES_HANDLER_KEY, trigger_type="cron",
        interval_seconds=None, cron_expr="9 2 * * *", timezone="Asia/Shanghai",
        enabled=False, config={"integration_id": str(owner.id)},
    )
    db_session.add_all([owner, target, collision])
    await db_session.commit()

    summary = await reconcile_all_cloud_integrations(db_session)

    assert str(target.id) in summary["failed_integration_ids"]
    # The outer transaction remains usable and successful integrations commit.
    assert (await db_session.scalar(select(Integration.id).where(
        Integration.id == owner.id,
    ))) == owner.id


@pytest.mark.asyncio
async def test_sync_skipped_for_non_cloud(db_session: AsyncSession):
    integ = Integration(
        id=uuid.uuid4(),
        kind="chat_approval",
        provider="feishu",
        name="not-cloud",
        slug=f"fs-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=True,
    )
    db_session.add(integ)
    await db_session.commit()

    summary = await sync_jobs_for_integration(db_session, integ)
    assert summary.get("skipped") == "not cloud integration"


# ── Billing job auto-provisioning for every live cloud provider ────


@pytest.mark.parametrize("provider_name", [
    "aws", "aliyun", "alicloud_intl", "azure", "gcp", "tencent", "ucloud",
])
@pytest.mark.asyncio
async def test_every_live_provider_gets_a_monthly_billing_row(
    db_session: AsyncSession, provider_name: str,
):
    """Regression guard for the prod-audit incident (2026-04-25):
    only AWS had a ``cmdb.billing.<slug>`` row because every other
    driver lacked a native ``get_monthly_billing``. Each live
    provider now ships a real implementation against the vendor's
    billing API; this parametrized test pins that contract.
    """
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider_name,
        name=f"{provider_name}-billing-test",
        slug=f"{provider_name}-bt-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()

    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    rows = await _all_jobs_for(db_session, str(integ.id))
    billing_rows = [r for r in rows if r.handler_key == BILLING_HANDLER_KEY]
    assert len(billing_rows) == 1, (
        f"{provider_name} did not get a cmdb.billing.<slug> row"
    )
    assert billing_rows[0].config["integration_id"] == str(integ.id)
    assert billing_rows[0].name == f"cmdb.billing.{integ.slug}"


@pytest.mark.asyncio
async def test_ksyun_does_not_get_monthly_billing_row(
    db_session: AsyncSession,
):
    """Ksyun is a deliberate read-only stub with no live API
    (historical data only — see provider docstring). Provisioning a
    billing row would create a job that 500s on every run; the
    correct behavior is to skip it. This test pins that contract so
    a future "provision billing for everyone" refactor doesn't
    regress."""
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="ksyun",
        name="ksyun-legacy",
        slug=f"ksyun-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=[],
    )
    db_session.add(integ)
    await db_session.commit()

    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    rows = await _all_jobs_for(db_session, str(integ.id))
    billing_rows = [r for r in rows if r.handler_key == BILLING_HANDLER_KEY]
    assert billing_rows == []


@pytest.mark.parametrize("provider_name", [
    "aws", "aliyun", "alicloud_intl", "azure", "gcp", "tencent", "ucloud",
])
def test_every_live_provider_class_has_get_monthly_billing(provider_name: str):
    """Static guard: every live cloud provider class must define
    ``get_monthly_billing`` — running the handler is an integration
    concern, but the *method existing* is a build-time invariant we
    can pin without spinning a session.
    """
    from app.domains.cmdb.providers.base import get_provider_class
    cls = get_provider_class(provider_name)
    assert callable(getattr(cls, "get_monthly_billing", None)), (
        f"{cls.__name__} is missing get_monthly_billing — every "
        f"live cloud provider must ship a native impl"
    )


def test_ksyun_provider_class_has_no_get_monthly_billing():
    """Mirror image: ksyun must NOT have it (it's a stub)."""
    from app.domains.cmdb.providers.base import get_provider_class
    cls = get_provider_class("ksyun")
    assert not callable(getattr(cls, "get_monthly_billing", None))


# ── cost_facts lifecycle row provisioning (regression guard) ───────


@pytest.mark.parametrize("provider_name", [
    "aws", "aliyun", "alicloud_intl", "azure", "gcp", "tencent", "ucloud",
])
@pytest.mark.asyncio
async def test_every_cloud_integration_gets_a_cost_facts_row(
    db_session: AsyncSession, provider_name: str,
):
    """Regression guard for the migration 139 seed bug: ``cmdb.cost_
    facts.<slug>`` rows weren't being seeded because the migration
    filtered on ``"billing" in capabilities``, a flag no driver
    declares. Post-fix, the row is provisioned for every cloud
    integration via ``scheduled_jobs_provisioning._build_lifecycle_
    rows``. Drivers that don't implement ``list_cost_facts`` have the
    handler return ``not_supported`` per scope rather than raising —
    so the cron is harmless on stub drivers.

    Pins the contract so a future "trim lifecycle rows by capability"
    refactor doesn't re-introduce the seed gap.
    """
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider_name,
        name=f"{provider_name}-cost-facts-test",
        slug=f"{provider_name}-cf-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()

    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    rows = await _all_jobs_for(db_session, str(integ.id))
    cost_facts_rows = [
        r for r in rows if r.handler_key == COST_FACTS_HANDLER_KEY
    ]
    assert len(cost_facts_rows) == 1, (
        f"{provider_name} did not get a cmdb.cost_facts.<slug> row"
    )
    assert cost_facts_rows[0].config["integration_id"] == str(integ.id)
    assert cost_facts_rows[0].name == f"cmdb.cost_facts.{integ.slug}"


@pytest.mark.asyncio
async def test_cost_facts_row_backfilled_by_reconcile(
    db_session: AsyncSession,
):
    """Mirror of ``test_reconcile_backfills_missing_lifecycle_rows``
    but specifically pinning the cost_facts handler. Pre-fix, an
    integration created before migration 139 ran would never get a
    cost_facts row even after a worker restart, because the seed
    capability filter rejected it. Post-fix, the worker boot
    reconcile path always provisions it.
    """
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="cost-facts-drift-test",
        slug=f"cfd-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.commit()
    await sync_jobs_for_integration(db_session, integ)
    await db_session.commit()

    # Simulate a pre-fix integration that never got the cost_facts row
    rows = await _all_jobs_for(db_session, str(integ.id))
    cost_facts_row = next(
        r for r in rows if r.handler_key == COST_FACTS_HANDLER_KEY
    )
    await db_session.delete(cost_facts_row)
    await db_session.commit()

    summary = await reconcile_all_cloud_integrations(db_session)
    assert summary["rows_added"] >= 1

    rows_after = await _all_jobs_for(db_session, str(integ.id))
    handler_keys_after = {r.handler_key for r in rows_after}
    assert COST_FACTS_HANDLER_KEY in handler_keys_after