"""Driver contract for ``preflight_close_scope`` / ``clear_close_blockers``. These are the two methods that turned "Azure refuses to cancel a subscription that still contains resources — here is the ``az group delete`` command" into work Helix does itself. The contract they have to keep: * **Preflight never lies about an empty scope.** A failed read is ``checked=False``, never an empty ``blockers`` list presented as fact. The dialog's 「没有需要清理的资源」 copy is gated on ``supported and checked``, so a driver that collapses the two would make Helix promise an empty subscription it never looked into. * **Containers describe their contents.** A resource group's name is not enough information to consent to deleting it; ``children`` / ``child_count`` are what make the confirmation meaningful. This is the whole reason ``it-team-us`` could be submitted as 从未使用过. * **Locks are deleted before resource groups.** A ``CanNotDelete`` lock makes the group delete 409, so a cleanup that ignores locks passes on empty test subscriptions and fails on the real ones. * **Cross-subscription keys are refused wholesale.** A key that does not address the subscription being closed is a caller bug, and acting on it would delete resources in an unrelated subscription. * **Idempotency.** ``204``/``404`` (already gone) is ``completed``, not ``failed`` — the resume job calls this repeatedly while deletions are in flight. """ from __future__ import annotations import json as _json from unittest.mock import AsyncMock, patch import pytest from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers.azure_cloud import AzureCloudDriver SUB = "f761fca3-6e78-4661-ae52-b4176f0455b7" RG_KEY = f"/subscriptions/{SUB}/resourceGroups/rg-billing" LOCK_KEY = ( f"/subscriptions/{SUB}/resourceGroups/rg-billing/providers" "/Microsoft.Authorization/locks/keep-me" ) def _azure_driver() -> AzureCloudDriver: return AzureCloudDriver( integration_id="int-azure", config={ "tenant_id": "00000000-0000-0000-0000-000000000001", "client_id": "client-id", "client_secret": "client-secret", }, ) class _Resp: def __init__(self, status_code: int, payload: dict | None = None) -> None: self.status_code = status_code self._payload = payload self.text = "" if payload is None else _json.dumps(payload) self.headers: dict[str, str] = {} def json(self): if self._payload is None: raise ValueError("no body") return self._payload class _ArmStub: """Routes ARM GET/DELETE by URL fragment; records every delete.""" def __init__( self, *, groups: list[dict] | None = None, resources: list[dict] | None = None, locks: list[dict] | None = None, group_status: int = 202, lock_status: int = 200, groups_status: int = 200, resources_status: int = 200, locks_status: int = 200, ) -> None: self.groups = groups or [] self.resources = resources or [] self.locks = locks or [] self.group_status = group_status self.lock_status = lock_status self.groups_status = groups_status self.resources_status = resources_status self.locks_status = locks_status self.deletes: list[str] = [] async def __aenter__(self): return self async def __aexit__(self, *a): return False async def get(self, url, params=None, headers=None): low = url.lower() if "/locks" in low: return _Resp(self.locks_status, {"value": self.locks}) if low.endswith("/resourcegroups"): return _Resp(self.groups_status, {"value": self.groups}) if low.endswith("/resources"): return _Resp(self.resources_status, {"value": self.resources}) return _Resp(200, {"state": "Enabled"}) async def delete(self, url, params=None, headers=None): self.deletes.append(url) if "/locks/" in url.lower(): return _Resp(self.lock_status, None) return _Resp(self.group_status, None) def _ref() -> CloudScopeRef: return CloudScopeRef(scope_type="azure_subscription", external_id=SUB) @pytest.mark.asyncio async def test_preflight_lists_groups_with_their_contents() -> None: """The production shape: one resource group holding one storage account, in a subscription whose spend is $0.00. The blocker must name what is INSIDE it — that is the fact 「从未使用过」 got wrong.""" stub = _ArmStub( groups=[{ "id": RG_KEY, "name": "rg-billing", "location": "westus2", "properties": {"provisioningState": "Succeeded"}, }], resources=[{ "id": f"{RG_KEY}/providers/Microsoft.Storage/storageAccounts" "/billingexport123", "name": "billingexport123", "type": "Microsoft.Storage/storageAccounts", "resourceGroup": "rg-billing", }], ) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): pf = await _azure_driver().preflight_close_scope(_ref()) assert pf.supported is True assert pf.checked is True assert pf.cleanup_supported is True assert [b.kind for b in pf.blockers] == ["resource_group"] blocker = pf.blockers[0] assert blocker.key == RG_KEY assert blocker.child_count == 1 assert "billingexport123" in blocker.children[0] # The vendor's own type segment, not an invented translation. assert "storageAccounts" in blocker.children[0] @pytest.mark.asyncio async def test_preflight_groups_resources_by_parsing_the_arm_id() -> None: """ARM's list-by-subscription response is ``GenericResourceExpanded``, whose documented fields are id / name / type / location / tags — it does **not** promise ``resourceGroup``. The grouping therefore falls back to the resource-group segment of the ARM id, which is always there. Without this path every resource would land under the empty key and each blocker would read "contains 0 resources" for a resource group that is demonstrably not empty.""" stub = _ArmStub( groups=[{ "id": RG_KEY, "name": "rg-billing", "location": "westus2", "properties": {"provisioningState": "Succeeded"}, }], resources=[{ # No ``resourceGroup`` key at all — the shape to expect. "id": f"{RG_KEY}/providers/Microsoft.Storage/storageAccounts" "/billingexport123", "name": "billingexport123", "type": "Microsoft.Storage/storageAccounts", }], ) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): pf = await _azure_driver().preflight_close_scope(_ref()) [blocker] = pf.blockers assert blocker.child_count == 1 assert "billingexport123" in blocker.children[0] @pytest.mark.asyncio async def test_preflight_reports_locks_as_their_own_blockers() -> None: stub = _ArmStub( groups=[{ "id": RG_KEY, "name": "rg-billing", "location": "westus2", "properties": {"provisioningState": "Succeeded"}, }], locks=[{ "id": LOCK_KEY, "name": "keep-me", "properties": {"level": "CanNotDelete"}, }], ) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): pf = await _azure_driver().preflight_close_scope(_ref()) kinds = [b.kind for b in pf.blockers] # Locks first: they are the blocker's blocker, and cleanup deletes # in this order for that reason. assert kinds == ["lock", "resource_group"] assert pf.blockers[0].detail == "CanNotDelete" @pytest.mark.asyncio async def test_preflight_read_failure_is_not_an_empty_scope() -> None: """``checked=False``, not "nothing to clean". The dialog's empty-state copy is gated on ``checked``, so collapsing these two would have Helix promise an empty subscription it never read.""" stub = _ArmStub(groups_status=500) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): pf = await _azure_driver().preflight_close_scope(_ref()) assert pf.supported is True assert pf.checked is False assert pf.blockers == [] assert pf.error_code == "preflight_read_failed" @pytest.mark.asyncio async def test_preflight_partial_read_is_labelled_partial() -> None: """Groups read fine, resources did not. Showing the group list is useful; showing it as complete is not.""" stub = _ArmStub( groups=[{ "id": RG_KEY, "name": "rg-billing", "location": "westus2", "properties": {}, }], resources_status=503, ) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): pf = await _azure_driver().preflight_close_scope(_ref()) assert len(pf.blockers) == 1 assert pf.checked is False assert pf.error_code == "preflight_partial" @pytest.mark.asyncio async def test_cleanup_deletes_locks_before_resource_groups() -> None: stub = _ArmStub() with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): result = await _azure_driver().clear_close_blockers( _ref(), keys=[RG_KEY, LOCK_KEY], ) assert result.ok is True assert [("lock" if "/locks/" in u else "rg") for u in stub.deletes] \ == ["lock", "rg"] assert result.completed == [LOCK_KEY] # lock delete → 200 assert result.started == [RG_KEY] # RG delete → 202, async assert result.in_progress is True @pytest.mark.asyncio async def test_cleanup_treats_already_gone_as_completed() -> None: """ARM answers 204 for a resource group that no longer exists. The resume job calls this every 5 minutes while deletions are in flight, so "already gone" must be success, not failure.""" stub = _ArmStub(group_status=204) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): result = await _azure_driver().clear_close_blockers( _ref(), keys=[RG_KEY], ) assert result.ok is True assert result.completed == [RG_KEY] assert result.failed == [] assert result.in_progress is False @pytest.mark.asyncio async def test_cleanup_surfaces_a_409_as_a_named_failure() -> None: stub = _ArmStub(group_status=409) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): result = await _azure_driver().clear_close_blockers( _ref(), keys=[RG_KEY], ) assert result.ok is False assert result.started == [] and result.completed == [] assert result.failed and result.failed[0]["key"] == RG_KEY assert "409" in result.failed[0]["error"] @pytest.mark.asyncio async def test_cleanup_refuses_keys_from_another_subscription() -> None: """The blast-radius guard: one foreign key fails the whole call rather than being skipped, because a caller that can produce one can produce a list of them.""" stub = _ArmStub() foreign = ( "/subscriptions/00000000-0000-0000-0000-0000000000ff" "/resourceGroups/somebody-elses" ) with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \ patch.object(AzureCloudDriver, "_http", return_value=stub): result = await _azure_driver().clear_close_blockers( _ref(), keys=[RG_KEY, foreign], ) assert result.ok is False assert result.error_code == "key_scope_mismatch" assert stub.deletes == [] # nothing was deleted at all @pytest.mark.asyncio async def test_preflight_is_unsupported_for_non_subscription_scopes() -> None: pf = await _azure_driver().preflight_close_scope( CloudScopeRef(scope_type="azure_resource_group", external_id="rg-x"), ) assert pf.supported is False assert pf.blockers == [] # ── GCP: liens are the enumerable, clearable blocker ────────────── def _gcp_driver(): from app.integrations.drivers.gcp_cloud import GcpCloudDriver return GcpCloudDriver( integration_id="int-gcp", config={"service_account_json": "{}", "organization_id": "0000"}, ) @pytest.mark.asyncio async def test_gcp_preflight_lists_liens_by_origin() -> None: """``liens/p1234abcd`` is not a name anyone recognises; the origin (``google.com/billing``, a Terraform module path) is what tells the admin who attached it and therefore whether removing it is safe.""" from unittest.mock import MagicMock from app.integrations.drivers.gcp_cloud import GcpCloudDriver crm = MagicMock() crm.liens.return_value.list.return_value.execute.return_value = { "liens": [{ "name": "liens/p1234abcd", "origin": "google.com/billing", "reason": "Deleting this project will break billing export", }], } with patch.object(GcpCloudDriver, "_svc", return_value=crm): pf = await _gcp_driver().preflight_close_scope( CloudScopeRef(scope_type="gcp_project", external_id="proj-1"), ) assert pf.supported is True and pf.checked is True assert [b.kind for b in pf.blockers] == ["lien"] assert pf.blockers[0].key == "liens/p1234abcd" assert pf.blockers[0].name == "google.com/billing" @pytest.mark.asyncio async def test_gcp_cleanup_deletes_liens_and_is_idempotent_on_404() -> None: from unittest.mock import MagicMock from app.integrations.drivers import gcp_cloud from app.integrations.drivers.gcp_cloud import GcpCloudDriver class _R: def __init__(self, s): self.status = s gone = Exception("already deleted") gone.resp = _R(404) gone.content = b"{}" crm = MagicMock() crm.liens.return_value.delete.return_value.execute.side_effect = [ {}, gone, ] with patch.object(GcpCloudDriver, "_svc", return_value=crm), \ patch.object(gcp_cloud, "HttpError", Exception): result = await _gcp_driver().clear_close_blockers( CloudScopeRef(scope_type="gcp_project", external_id="proj-1"), keys=["liens/aaa", "liens/bbb"], ) assert result.ok is True assert result.completed == ["liens/aaa", "liens/bbb"] assert result.failed == [] # Nothing to wait for at the provider — unlike an Azure RG delete a # lien delete is synchronous, so the close can be retried at once. assert result.in_progress is False @pytest.mark.asyncio async def test_gcp_cleanup_refuses_keys_that_are_not_lien_names() -> None: """Lien names carry no project segment, so shape is the only guard available. Silently ignoring a non-lien key would report a blocker as cleared when nothing happened.""" result = await _gcp_driver().clear_close_blockers( CloudScopeRef(scope_type="gcp_project", external_id="proj-1"), keys=["projects/proj-1"], ) assert result.ok is False assert result.error_code == "key_shape_mismatch"