"""Decommission that finishes itself: blocker cleanup + resume.

The behaviour under test is the answer to a real 2026-09-07 report: an
admin clicked 下线 on the Azure subscription ``it-team-us``, Azure
answered ``400 ResourcesExistingOnSubscription``, and Helix rendered
that as a red modal containing an ``az group delete`` one-liner — i.e.
it asked the person who requested the decommission to go perform the
decommission. ``decommission_reconcile``'s module docstring already
had the rule ("Reporting work back to the person who asked you to do it
is not diligence"); it only covered the half AFTER the provider accepts.

Contract locked in here:

* With ``cleanup_blockers=True`` and a confirmed key list, a
  ``resources_existing`` refusal starts a cleanup instead of raising:
  the driver is told to delete the confirmed keys, the scope records an
  in-flight decommission, and the caller gets
  ``error_code='cleanup_in_progress'`` (a 200, not a 502).
* **Consent is the boundary.** A live blocker the admin never saw stops
  everything with 409 ``SCOPE_CLOSE_BLOCKERS_CHANGED``, writes nothing,
  and deletes nothing. This is the property that makes automatic
  deletion safe.
* ``status`` stays ``active`` for the whole cleanup window — the cloud
  account really is alive until the provider accepts.
* Without the opt-in, behaviour is bit-for-bit the old one (502).
* The resume job carries it to the end, clears the pending columns, and
  a stalled/unconfirmed cleanup escalates to the clicker instead of
  retrying forever or going quiet.
"""
from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.audit.models import AuditLog
from app.core.exceptions import AppException
from app.domains.cmdb import decommission_cleanup as cleanup_mod
from app.domains.cmdb import lifecycle_service as lifecycle_service_mod
from app.domains.cmdb.decommission_cleanup import resume_pending_decommissions
from app.domains.cmdb.lifecycle_service import (
    CLEANUP_STATE_CLEANING,
    CLEANUP_STATE_NEEDS_RECONFIRM,
    CLEANUP_STATE_STALLED,
    MAX_CLEANUP_ATTEMPTS,
    CloudLifecycleService,
    list_idle_scope_candidates,
    plan_blocker_cleanup,
)
from app.domains.cmdb.models import (
    SCOPE_STATUS_ACTIVE,
    SCOPE_STATUS_CLOSED,
    CmdbCloudScope,
)
from app.integrations.driver_base import (
    ClearBlockersResult,
    CloseBlocker,
    ClosePreflight,
    CloseScopeResult,
    CloudScopeRef,
)
from app.integrations.models import Integration
from tests.test_cmdb.test_scope_decommission import (
    _make_integration,
    _make_person,
    _make_scope,
)

RG_A = "/subscriptions/sub-1/resourceGroups/rg-a"
RG_B = "/subscriptions/sub-1/resourceGroups/rg-b"


def _blocker(key: str, **kw: Any) -> CloseBlocker:
    return CloseBlocker(
        key=key, kind="resource_group",
        name=key.rsplit("/", 1)[-1], **kw,
    )


@dataclass
class StubCleanupDriver:
    """Driver stub covering the three methods the cleanup path uses."""

    blockers: list[CloseBlocker] = field(default_factory=list)
    close_results: list[CloseScopeResult] = field(default_factory=list)
    clear_result: ClearBlockersResult | None = None
    preflight_supported: bool = True
    preflight_checked: bool = True
    cleared_keys: list[list[str]] = field(default_factory=list)
    close_calls: int = 0

    async def close_scope_on_provider(
        self, scope: CloudScopeRef, *, iam_account_ids: list[str] | None = None,
    ) -> CloseScopeResult:
        self.close_calls += 1
        if len(self.close_results) > 1:
            return self.close_results.pop(0)
        return self.close_results[0]

    async def preflight_close_scope(
        self, scope: CloudScopeRef,
    ) -> ClosePreflight:
        return ClosePreflight(
            supported=self.preflight_supported,
            checked=self.preflight_checked,
            cleanup_supported=True,
            blockers=list(self.blockers),
        )

    async def clear_close_blockers(
        self, scope: CloudScopeRef, *, keys: list[str],
    ) -> ClearBlockersResult:
        self.cleared_keys.append(list(keys))
        if self.clear_result is not None:
            return self.clear_result
        return ClearBlockersResult(
            ok=True, started=list(keys), in_progress=True,
        )


def _patch(monkeypatch, integration_id: str, driver: StubCleanupDriver) -> None:
    async def _fake_load(session, scope):
        integ = await session.get(Integration, uuid.UUID(integration_id))
        return integ, driver

    for mod in (lifecycle_service_mod, cleanup_mod):
        monkeypatch.setattr(mod, "_load_driver_for_scope", _fake_load)

    async def _noop_audit_failure(**_kwargs):
        return None

    monkeypatch.setattr(
        lifecycle_service_mod, "_audit_decommission_failure",
        _noop_audit_failure,
    )


async def _reread(db_engine, scope_id: uuid.UUID) -> CmdbCloudScope:
    """Re-read a scope in a FRESH session.

    The resume job commits in its own session; this test session's
    REPEATABLE READ snapshot predates that commit, and a plain
    ``select`` here would also hand back the stale object already in
    its identity map. Rolling back to end the snapshot expires those
    objects and the next attribute touch tries to refresh mid-assert
    (``MissingGreenlet``). A separate session sidesteps both.
    """
    factory = async_sessionmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    async with factory() as session:
        row = await session.get(CmdbCloudScope, scope_id)
        assert row is not None
        return row


def _refused() -> CloseScopeResult:
    return CloseScopeResult(
        ok=False, error_code="resources_existing",
        error_message="Azure refuses to cancel a subscription that …",
        detail={"arm_code": "ResourcesExistingOnSubscription"},
    )


# ── plan_blocker_cleanup: the consent split ──────────────────────


def test_plan_matches_keys_case_insensitively() -> None:
    """ARM returns ``resourceGroups`` and ``resourcegroups`` from
    different endpoints. A case-sensitive compare would classify a
    confirmed group as unconfirmed and stall every Azure decommission
    on a spurious re-confirmation."""
    plan = plan_blocker_cleanup(
        [_blocker("/subscriptions/sub-1/RESOURCEGROUPS/rg-a")],
        [RG_A],
    )
    assert plan.deletable == ["/subscriptions/sub-1/RESOURCEGROUPS/rg-a"]
    assert plan.unseen == [] and plan.unremovable == []


def test_plan_separates_unconfirmed_and_unremovable() -> None:
    plan = plan_blocker_cleanup(
        [
            _blocker(RG_A),
            _blocker(RG_B),                                    # never confirmed
            _blocker("lock-1", removable=False, blocked_reason="no permission"),
        ],
        [RG_A, "lock-1"],
    )
    assert plan.deletable == [RG_A]
    assert [b.key for b in plan.unseen] == [RG_B]
    assert [b.key for b in plan.unremovable] == ["lock-1"]


# ── click path ───────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_refusal_with_confirmed_keys_starts_cleanup_instead_of_erroring(
    db_session: AsyncSession, admin_user, monkeypatch,
):
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration,
        scope_type="azure_subscription", external_id="sub-1",
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(RG_A, child_count=1, children=["storageAccounts x"])],
        close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)

    service = CloudLifecycleService(db_session)
    got_scope, disabled, close_result = await service.decommission_scope(
        str(scope.id), reason="从未使用过", actor_user_id=admin.id,
        trace_id="trace-cleanup-1",
        cleanup_blockers=True, confirmed_blocker_keys=[RG_A],
    )
    await db_session.commit()

    # Accepted, not failed, and NOT closed.
    assert close_result.error_code == "cleanup_in_progress"
    assert disabled == 0
    assert got_scope.status == SCOPE_STATUS_ACTIVE
    # Only the confirmed key was handed to the driver.
    assert driver.cleared_keys == [[RG_A]]
    # In-flight bookkeeping survived (this path must NOT raise, or the
    # rollback would take these columns with it).
    fresh = await db_session.get(CmdbCloudScope, scope.id)
    assert fresh.decommission_pending_since is not None
    assert fresh.decommission_pending_by == str(admin.id)
    assert fresh.decommission_pending_reason == "从未使用过"
    assert fresh.decommission_cleanup_state == CLEANUP_STATE_CLEANING
    assert fresh.decommission_cleanup_keys == [RG_A]
    assert fresh.decommission_cleanup_attempts == 1
    actions = [
        a for (a,) in (await db_session.execute(
            select(AuditLog.action).where(
                AuditLog.resource_id == str(scope.id),
            )
        )).all()
    ]
    assert "cloud.scope.decommission_cleanup_started" in actions


@pytest.mark.asyncio
async def test_a_blocker_nobody_confirmed_refuses_and_deletes_nothing(
    db_session: AsyncSession, admin_user, monkeypatch,
):
    """The safety property. Something appeared after the dialog was
    rendered; Helix must not sweep it up on the strength of a
    confirmation that never covered it."""
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration,
        scope_type="azure_subscription", external_id="sub-1",
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(RG_A), _blocker(RG_B)],
        close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)

    service = CloudLifecycleService(db_session)
    with pytest.raises(AppException) as exc:
        await service.decommission_scope(
            str(scope.id), reason="cleanup", actor_user_id=admin.id,
            trace_id="trace-cleanup-2",
            cleanup_blockers=True, confirmed_blocker_keys=[RG_A],
        )
    assert exc.value.code == "SCOPE_CLOSE_BLOCKERS_CHANGED"
    assert exc.value.status_code == 409
    assert driver.cleared_keys == []            # nothing deleted
    # No rollback needed to prove this: the service raised before it
    # wrote anything, so the in-session row is still untouched.
    assert scope.decommission_pending_since is None


@pytest.mark.asyncio
async def test_cleanup_that_cannot_start_tells_the_admin_now(
    db_session: AsyncSession, admin_user, monkeypatch,
):
    """The admin is still at the keyboard, so a provider refusal on the
    DELETE itself belongs in their face — pending would mean going
    quiet on work that never started."""
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration,
        scope_type="azure_subscription", external_id="sub-1",
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(RG_A)],
        close_results=[_refused()],
        clear_result=ClearBlockersResult(
            ok=False, failed=[{"key": RG_A, "error": "403 AuthorizationFailed"}],
            error_message="1 blocker(s) could not be deleted",
        ),
    )
    _patch(monkeypatch, str(integration.id), driver)

    service = CloudLifecycleService(db_session)
    with pytest.raises(AppException) as exc:
        await service.decommission_scope(
            str(scope.id), reason="cleanup", actor_user_id=admin.id,
            trace_id="trace-cleanup-3",
            cleanup_blockers=True, confirmed_blocker_keys=[RG_A],
        )
    assert exc.value.code == "SCOPE_CLOSE_CLEANUP_FAILED"


@pytest.mark.asyncio
async def test_without_the_opt_in_the_old_502_behaviour_is_unchanged(
    db_session: AsyncSession, admin_user, monkeypatch,
):
    """Automatic deletion must never be something a caller gets by
    omission — every pre-existing client keeps the old contract."""
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration,
        scope_type="azure_subscription", external_id="sub-1",
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(RG_A)], close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)

    service = CloudLifecycleService(db_session)
    with pytest.raises(AppException) as exc:
        await service.decommission_scope(
            str(scope.id), reason="cleanup", actor_user_id=admin.id,
            trace_id="trace-cleanup-4",
        )
    assert exc.value.code == "SCOPE_CLOSE_PROVIDER_ERROR"
    assert driver.cleared_keys == []


# ── resume job ───────────────────────────────────────────────────


async def _make_pending(
    db: AsyncSession, integration: Integration, actor_id: str,
    *, keys: list[str], attempts: int = 1,
    state: str = CLEANUP_STATE_CLEANING,
) -> CmdbCloudScope:
    scope = await _make_scope(
        db, integration, scope_type="azure_subscription",
        external_id="sub-1",
    )
    scope.decommission_pending_since = datetime.now(UTC) - timedelta(minutes=10)
    scope.decommission_pending_by = actor_id
    scope.decommission_pending_reason = "从未使用过"
    scope.decommission_cleanup_keys = keys
    scope.decommission_cleanup_state = state
    scope.decommission_cleanup_attempts = attempts
    scope.decommission_cleanup_trace_id = "trace-resume"
    await db.commit()
    return scope


@pytest.mark.asyncio
async def test_resume_closes_the_scope_once_the_blockers_are_gone(
    db_session: AsyncSession, admin_user, monkeypatch, client, db_engine,
):
    """The completion path: cleanup finished at the provider, so the
    close is retried and Helix's bookkeeping runs — with the ORIGINAL
    clicker as the actor, since it is their action being finished."""
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_pending(
        db_session, integration, str(admin.id), keys=[RG_A],
    )
    driver = StubCleanupDriver(
        blockers=[],                                   # provider is empty now
        close_results=[CloseScopeResult(
            ok=True, provider_state="cancelled",
            detail={"grace_period_days": 90},
        )],
    )
    _patch(monkeypatch, str(integration.id), driver)
    sent: list[Any] = []
    monkeypatch.setattr(
        cleanup_mod, "_notify_actor",
        lambda *a, **kw: sent.append(kw) or _true(),
    )

    summary = await resume_pending_decommissions()

    assert summary["closed"] == 1
    fresh = await _reread(db_engine, scope.id)
    assert fresh.status == SCOPE_STATUS_CLOSED
    assert fresh.decommissioned_by == str(admin.id)
    assert fresh.decommission_reason == "从未使用过"
    # Pending bookkeeping is cleared as one unit.
    assert fresh.decommission_pending_since is None
    assert fresh.decommission_cleanup_state is None
    assert fresh.decommission_cleanup_keys is None


async def _true() -> bool:
    return True


@pytest.mark.asyncio
async def test_resume_stops_for_an_unconfirmed_blocker(
    db_session: AsyncSession, admin_user, monkeypatch, client, db_engine,
):
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_pending(
        db_session, integration, str(admin.id), keys=[RG_A],
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(RG_B)],       # appeared after the confirmation
        close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)
    monkeypatch.setattr(
        cleanup_mod, "_notify_actor", lambda *a, **kw: _true(),
    )

    summary = await resume_pending_decommissions()

    assert summary["needs_reconfirm"] == 1
    assert driver.cleared_keys == []     # nothing unconfirmed was deleted
    fresh = await _reread(db_engine, scope.id)
    assert fresh.decommission_cleanup_state == CLEANUP_STATE_NEEDS_RECONFIRM
    assert fresh.status == SCOPE_STATUS_ACTIVE


@pytest.mark.asyncio
async def test_resume_escalates_once_when_the_budget_is_spent(
    db_session: AsyncSession, admin_user, monkeypatch, client, db_engine,
):
    """Six hours of the provider still refusing is a person's problem —
    but exactly one message, and the scope leaves the retry loop."""
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_pending(
        db_session, integration, str(admin.id), keys=[RG_A],
        attempts=MAX_CLEANUP_ATTEMPTS,
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(RG_A)], close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)
    notices: list[Any] = []
    monkeypatch.setattr(
        cleanup_mod, "_notify_actor",
        lambda *a, **kw: notices.append(kw.get("title_suffix")) or _true(),
    )

    summary = await resume_pending_decommissions()
    assert summary["stalled"] == 1
    assert len(notices) == 1

    fresh = await _reread(db_engine, scope.id)
    assert fresh.decommission_cleanup_state == CLEANUP_STATE_STALLED

    # Second tick must not re-notify: ``stalled`` is out of the queue.
    summary2 = await resume_pending_decommissions()
    assert summary2["checked"] == 0
    assert len(notices) == 1


@pytest.mark.asyncio
async def test_resume_never_advances_on_an_unreadable_preflight(
    db_session: AsyncSession, admin_user, monkeypatch, client,
):
    """"Could not look" must not become "nothing left to delete" — the
    close would then be retried as if the scope were empty."""
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    await _make_pending(
        db_session, integration, str(admin.id), keys=[RG_A],
    )
    driver = StubCleanupDriver(
        blockers=[], preflight_checked=False,
        close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)

    summary = await resume_pending_decommissions()

    assert summary["unreadable"] == 1
    assert summary["still_cleaning"] == 1
    assert driver.close_calls == 0


# ── cross-surface: the idle sweep must not nag a dying scope ──────


@pytest.mark.asyncio
async def test_idle_sweep_skips_a_scope_whose_decommission_is_in_flight(
    db_session: AsyncSession, admin_user,
):
    admin, _ = admin_user
    integration = await _make_integration(db_session, provider="azure")
    owner = await _make_person(
        db_session, email="owner-pending@test.com", name="Owner",
    )
    scope = await _make_scope(
        db_session, integration, scope_type="azure_subscription",
        external_id="sub-idle", owner_person_id=str(owner.id),
    )
    before = await list_idle_scope_candidates(db_session)
    scope.decommission_pending_since = datetime.now(UTC)
    scope.decommission_cleanup_state = CLEANUP_STATE_CLEANING
    await db_session.commit()
    after = await list_idle_scope_candidates(db_session)

    ids_before = {c.scope_id for c in before}
    ids_after = {c.scope_id for c in after}
    # Whether this scope qualifies as idle depends on billing rows the
    # helper needs; the invariant under test is one-directional — a
    # pending decommission can only REMOVE it from the candidate set.
    assert str(scope.id) not in ids_after
    assert ids_after <= ids_before


# ── Router: preflight endpoint ────────────────────────────────────


@pytest.mark.asyncio
async def test_preflight_endpoint_requires_the_decommission_role(
    client, db_session: AsyncSession, employee_user,
):
    """Same gate as the button it serves. The blocker list is a partial
    inventory of a cloud account's contents, and it describes an
    irreversible admin action — not an employee read."""
    _, emp_token = employee_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration, scope_type="azure_subscription",
        external_id="sub-rbac",
    )
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/decommission-preflight",
        headers={"Authorization": f"Bearer {emp_token}"},
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_preflight_endpoint_returns_blockers_and_pending_state(
    client, db_session: AsyncSession, admin_user, monkeypatch,
):
    _, admin_token = admin_user
    integration = await _make_integration(db_session, provider="azure")
    scope = await _make_scope(
        db_session, integration, scope_type="azure_subscription",
        external_id="sub-1",
    )
    driver = StubCleanupDriver(
        blockers=[_blocker(
            RG_A, child_count=1, children=["storageAccounts billingexport123"],
        )],
        close_results=[_refused()],
    )
    _patch(monkeypatch, str(integration.id), driver)

    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/decommission-preflight",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["supported"] is True and data["checked"] is True
    assert data["cleanup_supported"] is True
    assert data["pending"] is None
    [blocker] = data["blockers"]
    assert blocker["key"] == RG_A
    assert blocker["child_count"] == 1
    assert "billingexport123" in blocker["children"][0]
    assert blocker["removable"] is True


@pytest.mark.asyncio
async def test_a_driver_without_preflight_is_reported_as_unknown_not_empty(
    client, db_session: AsyncSession, admin_user, monkeypatch,
):
    """Every driver that has NOT implemented preflight (AWS, Aliyun,
    Tencent, UCloud today) must come back ``supported=false``. The
    dialog renders that as "cannot list contents", never as "nothing to
    clean" — otherwise adding the feature for two clouds would have
    quietly made the other four claim empty accounts."""
    _, admin_token = admin_user
    integration = await _make_integration(db_session, provider="aws")
    scope = await _make_scope(db_session, integration)

    class _NoPreflightDriver:
        async def preflight_close_scope(self, scope_ref):
            raise NotImplementedError("no preflight here")

    async def _fake_load(session, scope_row):
        integ = await session.get(Integration, uuid.UUID(str(integration.id)))
        return integ, _NoPreflightDriver()

    monkeypatch.setattr(
        lifecycle_service_mod, "_load_driver_for_scope", _fake_load,
    )
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/decommission-preflight",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["supported"] is False
    assert data["blockers"] == []