"""2026-07-28/29 review P1 — the budget/threshold measure-conversion script
must be idempotent, and the rows PR #2079 already converted in prod must be
repairable.

``scripts/convert_cloud_budgets_to_list_measure.py`` re-denominates
``budget_usd_monthly`` and every ``alert_rules[*].threshold_usd`` by
``ratio = list / invoice``. Conversion is a MULTIPLICATION: without a marker
a second ``--apply`` (operator retry after a flaky SSM session, a second
operator running the cutover runbook, …) multiplies everything by ~2.7×
AGAIN on a 55%-discount account — thresholds drift to nonsense and the
false-trigger incident the script exists to prevent comes back inverted
(never-firing alerts).

Markers:

* each converted rule dict gains ``"measure": "list"`` (also the per-rule
  input of the billing evaluator's basis resolution);
* the budget's basis lives in ``cmdb_cloud_scopes.budget_measure``
  (migration 433). It used to be a ``platform_settings`` scope-id list read
  only by this script — invisible to the evaluator, the API and the UI,
  which is how an admin "correcting" the inflated budget silently re-created
  #2079.

The forward-only gap: the version of the script that RAN IN PROD wrote no
markers, so the rows it converted are "list numbers, zero markers" and a
re-run would multiply them again. Migration 433 backfills both markers from
the audit rows that run left behind; these tests pin that the backfilled
state makes ``--apply`` a proven no-op.
"""
from __future__ import annotations

import importlib.util
import uuid
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.domains.cmdb.models import (
    SCOPE_STATUS_ACTIVE,
    SCOPE_TYPE_GCP_PROJECT,
    CmdbCloudBillingDaily,
    CmdbCloudScope,
)
from app.integrations.models import Integration
from scripts.convert_cloud_budgets_to_list_measure import (
    _BUDGET_AUDIT_ACTION,
    _RULES_AUDIT_ACTION,
    convert_alert_rules,
    detect,
    run,
)

# ── Unit: the conversion function itself ─────────────────────────────


def test_convert_alert_rules_marks_and_converts_once() -> None:
    rules = [
        {"type": "daily", "threshold_usd": 250.0, "active": True},
        {"type": "monthly", "threshold_usd": 1000.0, "active": True},
        # Non-threshold entries pass through untouched.
        {"type": "daily", "threshold_usd": 0, "active": True},
    ]
    once, changes, already = convert_alert_rules(rules, 2.7)
    assert already == 0
    assert [c["new_threshold_usd"] for c in changes] == [675.0, 2700.0]
    assert once[0]["threshold_usd"] == 675.0
    assert once[0]["measure"] == "list"
    assert once[1]["threshold_usd"] == 2700.0
    assert once[1]["measure"] == "list"
    assert "measure" not in once[2]


def test_convert_alert_rules_second_pass_is_noop() -> None:
    """The P1 regression: converting an already-converted rule set must
    change NOTHING — no second ×ratio."""
    rules = [
        {"type": "daily", "threshold_usd": 250.0, "active": True},
        {"type": "monthly", "threshold_usd": 1000.0, "active": True},
    ]
    once, changes1, _ = convert_alert_rules(rules, 2.7)
    twice, changes2, already = convert_alert_rules(once, 2.7)
    assert len(changes1) == 2
    assert changes2 == []
    assert already == 2
    assert twice == once
    assert twice[0]["threshold_usd"] == 675.0  # NOT 1822.5 (= 675 × 2.7)


# ── Integration: the script's run() twice against a real scope ───────


async def _seed_scope_with_billing(
    db: AsyncSession,
    *,
    budget: float = 1000,
    rules: list[dict] | None = None,
) -> tuple[Integration, CmdbCloudScope]:
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="gcp",
        name="gcp-convert-test", slug=f"gcp-convert-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["billing"],
    )
    db.add(integ)
    # Commit the parent first — Integration lives in a different mapper
    # registry, so a single-flush insert does not topologically order it
    # before the scope's FK (same pattern as the sibling billing tests).
    await db.commit()
    scope = CmdbCloudScope(
        integration_id=str(integ.id), scope_type=SCOPE_TYPE_GCP_PROJECT,
        external_id=f"proj-{uuid.uuid4().hex[:8]}",
        display_name="convert-idempotency-twin",
        environment="dev",
        status=SCOPE_STATUS_ACTIVE,
        budget_usd_monthly=budget,
        alert_rules=rules if rules is not None else [
            {"type": "daily", "threshold_usd": 250.0, "active": True},
        ],
    )
    db.add(scope)
    await db.commit()

    # The script's reference window = the previous full calendar month
    # relative to now(UTC) — seed both measures there so ratio = 2.7.
    today = datetime.now(UTC).date()
    month_start = (today.replace(day=1) - timedelta(days=1)).replace(day=1)
    for i in range(3):
        db.add(CmdbCloudBillingDaily(
            id=str(uuid.uuid4()), integration_id=str(integ.id),
            scope_id=str(scope.id), linked_account_id=scope.external_id,
            date=month_start + timedelta(days=i),
            total_cost_usd=100.0, total_cost_usd_list=270.0,
            currency="USD", cost_by_service={}, raw_response={},
            fetched_at=datetime.now(UTC),
        ))
    await db.commit()
    return integ, scope


@pytest.mark.asyncio
async def test_run_apply_twice_leaves_values_unchanged(
    db_session: AsyncSession,
) -> None:
    _, scope = await _seed_scope_with_billing(db_session)

    @asynccontextmanager
    async def _factory():
        yield db_session

    await run(True, session_factory=_factory)
    await db_session.refresh(scope)
    budget_after_first = float(scope.budget_usd_monthly)
    rules_after_first = [dict(r) for r in scope.alert_rules]
    assert budget_after_first == 2700.0
    assert rules_after_first[0]["threshold_usd"] == 675.0
    assert rules_after_first[0]["measure"] == "list"
    # The budget's basis is now a first-class column, readable by the
    # evaluator / API / UI — not a script-private platform_settings list.
    assert scope.budget_measure == "list"

    # The regression: apply AGAIN — nothing may move.
    await run(True, session_factory=_factory)
    await db_session.refresh(scope)
    assert float(scope.budget_usd_monthly) == budget_after_first
    assert [dict(r) for r in scope.alert_rules] == rules_after_first
    assert scope.budget_measure == "list"


@pytest.mark.asyncio
async def test_run_dry_run_writes_nothing(db_session: AsyncSession) -> None:
    """Belt-and-braces: without --apply neither the values nor the
    markers may be persisted."""
    _, scope = await _seed_scope_with_billing(db_session)

    @asynccontextmanager
    async def _factory():
        yield db_session

    await run(False, session_factory=_factory)
    await db_session.refresh(scope)
    assert float(scope.budget_usd_monthly) == 1000.0
    assert scope.alert_rules[0]["threshold_usd"] == 250.0
    assert "measure" not in scope.alert_rules[0]
    assert scope.budget_measure is None


@pytest.mark.asyncio
async def test_apply_leaves_no_useless_platform_setting_row(
    db_session: AsyncSession,
) -> None:
    """P2: the old marker was a ``platform_settings`` id list written on
    EVERY ``--apply`` — including runs that converted nothing, which created
    a ``value=[]`` row and made "marker exists" a useless signal for "was
    this already run?". The per-scope column has no such row at all."""
    from app.integrations.models import PlatformSetting

    await _seed_scope_with_billing(db_session)

    @asynccontextmanager
    async def _factory():
        yield db_session

    await run(True, session_factory=_factory)
    stale = await db_session.get(
        PlatformSetting, "cloud.budget_measure_list_scope_ids",
    )
    assert stale is None, (
        "the script still maintains the legacy platform-settings marker — "
        "two sources of truth for one fact"
    )


# ── The already-converted-in-prod repair path (F1) ───────────────────


@pytest.mark.asyncio
async def test_unmarked_already_converted_scope_would_double_convert(
    db_session: AsyncSession,
) -> None:
    """The hazard itself, pinned so the repair has something to prove.

    A scope in the state PR #2079's prod run left behind — list-denominated
    budget and thresholds, NO markers — is indistinguishable from an
    unconverted scope, so ``--apply`` multiplies it a second time. This is
    why the repair lives in migration 433 (seeded from the audit rows) and
    not in the script.
    """
    _, scope = await _seed_scope_with_billing(
        db_session,
        budget=2700,  # already ×2.7
        rules=[{"type": "daily", "threshold_usd": 675.0, "active": True}],
    )

    @asynccontextmanager
    async def _factory():
        yield db_session

    await run(True, session_factory=_factory)
    await db_session.refresh(scope)
    assert float(scope.budget_usd_monthly) == 7290.0  # 2700 × 2.7
    assert scope.alert_rules[0]["threshold_usd"] == 1822.5  # 675 × 2.7


@pytest.mark.asyncio
async def test_markers_backfilled_from_audit_make_apply_a_noop(
    db_session: AsyncSession,
) -> None:
    """Same already-converted scope, but with migration 433's backfill
    applied first (here: exactly what the migration writes, derived from the
    audit rows the prod run left) → ``--apply`` must move NOTHING."""
    _, scope = await _seed_scope_with_billing(
        db_session,
        budget=2700,
        rules=[{"type": "daily", "threshold_usd": 675.0, "active": True}],
    )
    # What the prod run's audit trail records.
    audit = AuditService(db_session)
    await audit.log(
        trace_id=str(uuid.uuid4()), actor_type="system",
        action=_BUDGET_AUDIT_ACTION, status="ok",
        resource_type="cmdb_cloud_scope", resource_id=str(scope.id),
        detail={"old_budget_usd": 1000.0, "new_budget_usd": 2700.0,
                "ratio": 2.7},
    )
    await audit.log(
        trace_id=str(uuid.uuid4()), actor_type="system",
        action=_RULES_AUDIT_ACTION, status="ok",
        resource_type="cmdb_cloud_scope", resource_id=str(scope.id),
        detail={"changes": [{"type": "daily", "old_threshold_usd": 250.0,
                             "new_threshold_usd": 675.0}], "ratio": 2.7},
    )
    await db_session.commit()

    # ``--detect`` must call this state out BEFORE the repair.
    await detect(session_factory=_noop_factory(db_session))

    # Run migration 433's ACTUAL repair, not a hand-written equivalent.
    counts = await _run_migration_433_repair(db_session)
    assert counts == {"budgets": 1, "rule_sets": 1}, counts
    await db_session.commit()
    db_session.expire(scope)
    await db_session.refresh(scope)
    assert scope.budget_measure == "list"
    assert scope.alert_rules[0]["measure"] == "list"

    # Idempotent: a second run of the repair changes nothing.
    assert await _run_migration_433_repair(db_session) == {
        "budgets": 0, "rule_sets": 0,
    }

    @asynccontextmanager
    async def _factory():
        yield db_session

    await run(True, session_factory=_factory)
    await db_session.refresh(scope)
    assert float(scope.budget_usd_monthly) == 2700.0, (
        "re-running --apply after the migration-433 backfill converted a "
        "SECOND time — the backfill did not close the forward-only gap"
    )
    assert scope.alert_rules[0]["threshold_usd"] == 675.0
    assert scope.alert_rules[0]["measure"] == "list"


def _noop_factory(db_session: AsyncSession):
    @asynccontextmanager
    async def _factory():
        yield db_session
    return _factory


_MIGRATION_433 = (
    Path(__file__).resolve().parents[2]
    / "alembic" / "versions"
    / "433_cloud_budget_measure_backfill_from_audit.py"
)


async def _run_migration_433_repair(db: AsyncSession) -> dict[str, int]:
    """Execute migration 433's data-repair against the live test schema."""
    spec = importlib.util.spec_from_file_location("_m433", _MIGRATION_433)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    conn = await db.connection()
    return await conn.run_sync(module.repair_budget_measure_markers)


@pytest.mark.asyncio
async def test_migration_433_leaves_admin_retuned_rules_unstamped(
    db_session: AsyncSession,
) -> None:
    """Conservative matching: a rule whose threshold no longer equals the
    value the conversion audit recorded has been retuned by an admin SINCE,
    so its basis is whatever the admin was reading — net is the safe default
    and the evaluator's ambiguity warning covers the rest. Only rules that
    still carry the recorded post-conversion number get stamped."""
    _, scope = await _seed_scope_with_billing(
        db_session,
        budget=2700,
        rules=[
            # Still exactly what the conversion wrote → stamped.
            {"type": "daily", "threshold_usd": 675.0, "active": True},
            # Admin retuned this one afterwards → NOT stamped.
            {"type": "monthly", "threshold_usd": 4000.0, "active": True},
        ],
    )
    audit = AuditService(db_session)
    await audit.log(
        trace_id=str(uuid.uuid4()), actor_type="system",
        action=_RULES_AUDIT_ACTION, status="ok",
        resource_type="cmdb_cloud_scope", resource_id=str(scope.id),
        detail={"changes": [
            {"type": "daily", "old_threshold_usd": 250.0,
             "new_threshold_usd": 675.0},
            {"type": "monthly", "old_threshold_usd": 1000.0,
             "new_threshold_usd": 2700.0},
        ], "ratio": 2.7},
    )
    await db_session.commit()

    await _run_migration_433_repair(db_session)
    await db_session.commit()
    db_session.expire(scope)
    await db_session.refresh(scope)
    by_type = {r["type"]: r for r in scope.alert_rules}
    assert by_type["daily"].get("measure") == "list"
    assert "measure" not in by_type["monthly"]


@pytest.mark.asyncio
async def test_migration_433_is_a_noop_without_audit_rows(
    db_session: AsyncSession,
) -> None:
    """Fresh environments (and every test schema) have no conversion audit
    rows and no legacy platform setting — the repair must touch nothing."""
    _, scope = await _seed_scope_with_billing(db_session)
    assert await _run_migration_433_repair(db_session) == {
        "budgets": 0, "rule_sets": 0,
    }
    await db_session.refresh(scope)
    assert scope.budget_measure is None
    assert "measure" not in scope.alert_rules[0]


@pytest.mark.asyncio
async def test_migration_433_consumes_the_legacy_platform_setting(
    db_session: AsyncSession,
) -> None:
    """Scopes recorded only in the forward-only script's
    ``cloud.budget_measure_list_scope_ids`` list must land on the column, and
    the setting row must be deleted so there is one source of truth."""
    from app.integrations.models import PlatformSetting

    _, scope = await _seed_scope_with_billing(db_session, budget=2700)
    db_session.add(PlatformSetting(
        key="cloud.budget_measure_list_scope_ids", value=[str(scope.id)],
    ))
    await db_session.commit()

    assert await _run_migration_433_repair(db_session) == {
        "budgets": 1, "rule_sets": 0,
    }
    await db_session.commit()
    db_session.expire_all()
    await db_session.refresh(scope)
    assert scope.budget_measure == "list"
    assert await db_session.get(
        PlatformSetting, "cloud.budget_measure_list_scope_ids",
    ) is None