"""Unit tests for the Azure cloud driver — lifecycle methods. All HTTP mocked via respx. Covers happy-path + error flows for: * test_connection (OAuth token + ARM subscription list + Graph probe) * list_scopes (pagination + state filter) * list_scope_roles (fallback + live merge, de-duplication) * grant_access (happy + 409 idempotent + PrincipalNotFound) * revoke_access (happy + 404 idempotent) * revoke_account (multi-subscription sweep + per-sub failure report) * get_last_login (P1 available + 403 fallback) * list_billing_daily (Cost Management Query response parse) """ from __future__ import annotations from datetime import date from typing import Any from unittest.mock import patch import httpx import pytest import respx from app.integrations.driver_base import ( CloudAccessContext, CloudAccountRevokeContext, CloudPrincipalNotFoundError, CloudScopeRef, ) from app.integrations.drivers.azure_cloud import ( _ARM_BASE, _AZURE_FALLBACK_ROLES, _GRAPH_BASE, _GRAPH_BETA_BASE, _LOGIN_BASE, AzureCloudDriver, ) _CFG = { "tenant_id": "11111111-1111-1111-1111-111111111111", "client_id": "22222222-2222-2222-2222-222222222222", "client_secret": "topsecret", } _SUB_ID = "33333333-3333-3333-3333-333333333333" _SUB_ID_2 = "44444444-4444-4444-4444-444444444444" _PRINCIPAL_ID = "principal-uid-1" _RD1_ID = ( "/subscriptions/x/providers/Microsoft.Authorization/" "roleDefinitions/rd1" ) def _driver() -> AzureCloudDriver: return AzureCloudDriver(integration_id="azure-test", config=_CFG) def _mock_token() -> None: respx.mock.post( f"{_LOGIN_BASE}/{_CFG['tenant_id']}/oauth2/v2.0/token" ).mock(return_value=httpx.Response(200, json={ "access_token": "tok", "token_type": "Bearer", "expires_in": 3599, })) # ── metadata ──────────────────────────────────────────────────── def test_metadata_registered() -> None: m = AzureCloudDriver.metadata() assert m.kind == "cloud" assert m.provider == "azure" caps = set(m.capabilities) assert {"servers", "dns_records", "object_storage", "load_balancers", "user_revoke"}.issubset(caps) def test_fallback_roles_match_azure_privileged_admin_tab() -> None: """The curated default-role list MUST mirror exactly Azure portal's "Privileged administrator roles" tab — no more, no less. These 5 GUIDs are stable across every tenant (verified against learn.microsoft.com/azure/role-based-access-control built-in role docs). Drift here means employees see the wrong primary-role dropdown. Pinning the GUIDs in a test catches: * Migration 060's old GUID typo for User Access Administrator (``36243c78-...`` was wrong; correct is ``18d7d88d-...``) * Future "let's add Reader back to defaults" drift without an explicit decision. """ expected_guid_to_name = { "8e3af657-a8ff-443c-a75c-2fe8c4bcb635": "Owner", "b24988ac-6180-42a0-ab88-20f7382dd24c": "Contributor", "76cc9ee4-d5d3-4a45-a930-26add3d73475": "Access Review Operator Service Role", "f58310d9-a9f6-439a-9e8d-f62e7b41a168": "Role Based Access Control Administrator", "18d7d88d-d35e-4fb5-a5c3-7773c20a72d9": "User Access Administrator", } actual = {} for entry in _AZURE_FALLBACK_ROLES: rk = entry["role_key"] assert rk.startswith( "/providers/Microsoft.Authorization/roleDefinitions/", ), f"role_key not in tenant-scoped form: {rk}" guid = rk.rsplit("/", 1)[-1] actual[guid] = entry["display_name"] assert actual == expected_guid_to_name, ( "_AZURE_FALLBACK_ROLES drifted from Azure portal's " "'Privileged administrator roles' tab. If this is intentional, " "update both this test AND migration 069." ) # Defense-in-depth: the wrong GUID from migration 060 must never # come back via copy-paste. assert all( "36243c78-bf99-498c-9df9-86d9f8d28608" not in entry["role_key"] for entry in _AZURE_FALLBACK_ROLES ), "migration 060's incorrect UAA GUID has resurfaced" # ── test_connection ───────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_test_connection_arm_and_graph_ok() -> None: _mock_token() respx.mock.get(f"{_ARM_BASE}/subscriptions").mock( return_value=httpx.Response(200, json={"value": [{"id": "x"}]}), ) respx.mock.get(f"{_GRAPH_BASE}/users").mock( return_value=httpx.Response(200, json={"value": [{"id": "a"}]}), ) r = await _driver().test_connection() assert r.ok is True assert "Graph OK" in r.message @pytest.mark.asyncio @respx.mock async def test_test_connection_graph_missing_is_failure() -> None: _mock_token() respx.mock.get(f"{_ARM_BASE}/subscriptions").mock( return_value=httpx.Response(200, json={"value": []}), ) respx.mock.get(f"{_GRAPH_BASE}/users").mock( return_value=httpx.Response(403, json={"error": "forbidden"}), ) r = await _driver().test_connection() assert r.ok is False assert "Graph" in r.message # ── list_scopes ───────────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_list_scopes_filters_and_maps_status() -> None: _mock_token() respx.mock.get( f"{_ARM_BASE}/subscriptions", ).mock(return_value=httpx.Response(200, json={ "value": [ { "subscriptionId": _SUB_ID, "displayName": "Production", "state": "Enabled", "tenantId": _CFG["tenant_id"], }, { "subscriptionId": _SUB_ID_2, "displayName": "Dev", "state": "Disabled", "tenantId": _CFG["tenant_id"], }, ], })) scopes = await _driver().list_scopes() assert len(scopes) == 2 assert scopes[0].scope_type == "azure_subscription" assert scopes[0].external_id == _SUB_ID assert scopes[0].status == "active" assert scopes[1].status == "disabled" # ── list_scope_roles ─────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_list_scope_roles_merges_fallback_and_live() -> None: _mock_token() respx.mock.get( f"{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleDefinitions", ).mock(return_value=httpx.Response(200, json={ "value": [ { "id": _RD1_ID, "name": "rd1", "properties": { "roleName": "Reader", "description": "Read all", "type": "BuiltInRole", }, }, { "id": ( "/subscriptions/x/providers/Microsoft.Authorization/" "roleDefinitions/rd2" ), "name": "rd2", "properties": { "roleName": "Custom App Role", "description": "", "type": "CustomRole", }, }, ], })) async def _noop(_k: str, fallback): return list(fallback) with patch( "app.integrations.drivers.azure_cloud.load_default_roles", _noop, ): ref = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) roles = await _driver().list_scope_roles(ref) # 5 curated privileged-admin fallbacks + 2 live ones (keys don't # overlap because the live mock uses synthetic ``rd1``/``rd2`` # role_keys not in the curated GUID set) = 7. assert len(roles) == 7 sources = {r.source for r in roles} assert sources == {"azure_predefined", "azure_builtin", "azure_custom"} custom = [r for r in roles if r.is_custom] assert any("Custom App Role" == r.display_name for r in custom) # Regression (2026-04-24): live role keys returned by ARM come as # ``/subscriptions//providers/...`` (~138 chars) — driver # MUST normalize to the tenant-scoped form so # (a) the catalog row fits ``cmdb_cloud_roles.role_key`` safely # (overflowed VARCHAR(128) in prod before the fix) # (b) the same role_key shape grants-sync persists, so # ``owner_role_keys`` match works. live_keys = [r.role_key for r in roles if r.source != "azure_predefined"] assert all(k.startswith("/providers/") for k in live_keys), live_keys assert all(not k.startswith("/subscriptions/") for k in live_keys) # ── grant_access ─────────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_grant_access_creates_assignment() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock(return_value=httpx.Response( 200, json={"value": [{"id": _PRINCIPAL_ID, "mail": "alice@example.com"}]}, )) put = respx.mock.put( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments/.+" ), ).mock(return_value=httpx.Response(201, json={ "id": "ra1", "properties": {"principalId": _PRINCIPAL_ID}, })) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) result = await _driver().grant_access(ctx) assert result.ok assert result.external_ref.startswith(f"/subscriptions/{_SUB_ID}::") assert put.called @pytest.mark.asyncio @respx.mock async def test_grant_access_409_is_idempotent_and_finds_existing() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock(return_value=httpx.Response( 200, json={"value": [{"id": _PRINCIPAL_ID}]}, )) respx.mock.put( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments/.+" ), ).mock(return_value=httpx.Response(409, json={"error": { "code": "RoleAssignmentExists", }})) respx.mock.get( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": [ {"name": "existing-ra-guid", "properties": { "principalId": _PRINCIPAL_ID, "roleDefinitionId": _RD1_ID, }}, ]})) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name=None, scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) result = await _driver().grant_access(ctx) assert result.ok assert result.already_exists is True assert "existing-ra-guid" in result.external_ref @pytest.mark.asyncio @respx.mock async def test_grant_access_409_unresolvable_existing_fails_not_phantom() -> None: """#1649: Azure returns 409 RoleAssignmentExists but the reverse lookup can't locate the existing assignment GUID (read-replica lag / the assignment was scoped elsewhere). The driver must return ``ok=False`` — recording our locally-generated UUID as external_ref would persist a PHANTOM ref that a later revoke can never resolve, silently orphaning the binding. Pre-fix this landed ok=True.""" _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock(return_value=httpx.Response( 200, json={"value": [{"id": _PRINCIPAL_ID}]}, )) respx.mock.put( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments/.+" ), ).mock(return_value=httpx.Response(409, json={"error": { "code": "RoleAssignmentExists", }})) # The reverse lookup finds no matching assignment for this principal. respx.mock.get( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": []})) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name=None, scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) result = await _driver().grant_access(ctx) assert result.ok is False assert result.external_ref is None assert "phantom" in (result.error_message or "").lower() @pytest.mark.asyncio @respx.mock async def test_grant_access_400_mentioning_role_exists_is_not_idempotent() -> None: """#1649: a 400 (or 403) whose body merely CONTAINS the string 'RoleAssignmentExists' must NOT be misread as idempotent success — only a genuine 409 + that ARM error code is idempotent. Pre-fix the gate keyed on the substring alone, so a 400 error body that happened to mention the code flipped a real failure into a false ok=True and the reverse-lookup path (which never runs on a 400) was skipped.""" _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock(return_value=httpx.Response( 200, json={"value": [{"id": _PRINCIPAL_ID}]}, )) listed = respx.mock.get( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": [ {"name": "should-not-be-consulted", "properties": { "principalId": _PRINCIPAL_ID, "roleDefinitionId": _RD1_ID, }}, ]})) respx.mock.put( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments/.+" ), ).mock(return_value=httpx.Response(400, json={"error": { "code": "InvalidRequest", "message": "some upstream text mentioning RoleAssignmentExists", }})) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name=None, scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) result = await _driver().grant_access(ctx) assert result.ok is False assert result.already_exists is not True # The idempotent reverse-lookup must NOT have run on a 400. assert not listed.called @pytest.mark.asyncio @respx.mock async def test_grant_access_unknown_principal_reports_not_found() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock(return_value=httpx.Response( 200, json={"value": []}, )) ctx = CloudAccessContext( principal_email="ghost@example.com", principal_display_name=None, scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) result = await _driver().grant_access(ctx) assert result.ok is False assert "has no user" in result.error_message @pytest.mark.asyncio async def test_odata_filter_injection_is_rejected() -> None: d = _driver() with pytest.raises(CloudPrincipalNotFoundError): d._sanitize_email_for_odata("alice' or 1 eq 1--") # ── revoke_access ────────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_revoke_access_404_is_idempotent() -> None: _mock_token() respx.mock.delete( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments/.+" ), ).mock(return_value=httpx.Response(404, json={"error": "not found"})) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, existing_external_ref=f"/subscriptions/{_SUB_ID}::ra1", ) result = await _driver().revoke_access(ctx) assert result.ok assert result.already_absent # ── revoke_account (sweep) ───────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_revoke_account_sweeps_every_subscription() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock( return_value=httpx.Response(200, json={"value": [{"id": _PRINCIPAL_ID}]}), ) respx.mock.get(f"{_ARM_BASE}/subscriptions").mock( return_value=httpx.Response(200, json={"value": [ {"subscriptionId": _SUB_ID, "displayName": "Prod", "state": "Enabled"}, {"subscriptionId": _SUB_ID_2, "displayName": "Dev", "state": "Enabled"}, ]}), ) # Sub 1 has two assignments to delete. respx.mock.get( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": [ {"name": "ra-a", "properties": {"principalId": _PRINCIPAL_ID}}, {"name": "ra-b", "properties": {"principalId": _PRINCIPAL_ID}}, ]})) # Sub 2 has none. respx.mock.get( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID_2}/providers/" "Microsoft.Authorization/roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": []})) delete = respx.mock.delete( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.Authorization/roleAssignments/ra-.+" ), ).mock(return_value=httpx.Response(204)) result = await _driver().revoke_account( CloudAccountRevokeContext(provider="azure", account_id="alice@example.com"), ) assert result.ok assert result.raw_response["subscriptions_scanned"] == 2 assert _SUB_ID in result.raw_response["subscriptions_modified"] assert _SUB_ID_2 not in result.raw_response["subscriptions_modified"] assert result.raw_response["subscriptions_failed"] == [] assert delete.call_count == 2 @pytest.mark.asyncio @respx.mock async def test_revoke_account_missing_principal_is_noop_success() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BASE}/users").mock( return_value=httpx.Response(200, json={"value": []}), ) result = await _driver().revoke_account( CloudAccountRevokeContext(provider="azure", account_id="ghost@example.com"), ) assert result.ok assert result.raw_response["subscriptions_scanned"] == 0 assert result.raw_response["principal_id"] is None @pytest.mark.asyncio async def test_revoke_account_requires_email() -> None: result = await _driver().revoke_account( CloudAccountRevokeContext(provider="azure", account_id="not-an-email"), ) assert result.ok is False # ── get_last_login ───────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_get_last_login_parses_graph_signins() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BETA_BASE}/auditLogs/signIns").mock( return_value=httpx.Response(200, json={"value": [ {"createdDateTime": "2026-04-01T09:15:30Z"}, ]}), ) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name=None, scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) dt = await _driver().get_last_login(ctx) assert dt is not None assert dt.year == 2026 assert dt.month == 4 assert dt.day == 1 @pytest.mark.asyncio @respx.mock async def test_get_last_login_403_returns_none() -> None: _mock_token() respx.mock.get(f"{_GRAPH_BETA_BASE}/auditLogs/signIns").mock( return_value=httpx.Response(403, json={"error": "no P1"}), ) ctx = CloudAccessContext( principal_email="alice@example.com", principal_display_name=None, scope=CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID), role_key=_RD1_ID, ) dt = await _driver().get_last_login(ctx) assert dt is None # ── billing daily ──────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_list_billing_daily_parses_rows() -> None: _mock_token() respx.mock.post( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ), ).mock(return_value=httpx.Response(200, json={ "properties": { "columns": [ {"name": "Cost"}, {"name": "UsageDate"}, {"name": "ServiceName"}, {"name": "Currency"}, ], "rows": [ [12.5, 20260101, "Storage", "USD"], [7.25, 20260101, "Compute", "USD"], [3.0, 20260102, "Storage", "USD"], ], }, })) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) rows = await _driver().list_billing_daily( date(2026, 1, 1), date(2026, 1, 2), scope=scope, ) assert len(rows) == 2 by_date = {r.date: r for r in rows} assert by_date[date(2026, 1, 1)].total_cost_usd == pytest.approx(19.75) assert by_date[date(2026, 1, 2)].total_cost_usd == pytest.approx(3.0) assert by_date[date(2026, 1, 1)].cost_by_service["Storage"] == pytest.approx(12.5) @pytest.mark.asyncio @respx.mock async def test_list_billing_daily_follows_next_link() -> None: """Cost Management Query returns ``properties.nextLink`` when the response spills. Driver must POST the original body to the nextLink URL until the cursor is empty, merging rows across pages.""" _mock_token() first_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) next_link = ( f"{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query?api-version=2023-11-01&%24skiptoken=abc" ) call_count = {"n": 0} def _route(request: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: return httpx.Response(200, json={ "properties": { "nextLink": next_link, "columns": [ {"name": "Cost"}, {"name": "UsageDate"}, {"name": "ServiceName"}, {"name": "Currency"}, ], "rows": [[12.5, 20260101, "Storage", "USD"]], }, }) # Second page — no more nextLink, different rows. return httpx.Response(200, json={ "properties": { "columns": [ {"name": "Cost"}, {"name": "UsageDate"}, {"name": "ServiceName"}, {"name": "Currency"}, ], "rows": [[7.25, 20260101, "Compute", "USD"]], }, }) respx.mock.post(url__regex=first_url_regex).mock(side_effect=_route) respx.mock.post(next_link).mock(side_effect=_route) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) rows = await _driver().list_billing_daily( date(2026, 1, 1), date(2026, 1, 1), scope=scope, ) assert call_count["n"] == 2 assert len(rows) == 1 row = rows[0] assert row.total_cost_usd == pytest.approx(19.75) # Both pages' services must land — that's the silent-truncation # failure mode the fix addresses. assert row.cost_by_service["Storage"] == pytest.approx(12.5) assert row.cost_by_service["Compute"] == pytest.approx(7.25) @pytest.mark.asyncio @respx.mock async def test_list_billing_daily_raises_on_non_200() -> None: """Pre-fix the driver silently returned ``[]`` on 401/403/429/5xx from Cost Management. ``last_seen_billing_daily_at`` advanced and spike-detection ran on stale 7-day averages while the admin saw nothing in scheduled_job_runs.errors. ``list_cost_facts`` already raised on the same condition; daily was the holdout (audited 2026-05-10). Post-fix both paths surface real failures so the lifecycle service records the per-scope error.""" _mock_token() respx.mock.post( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ), ).mock(return_value=httpx.Response(429, text="rate limited")) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(RuntimeError, match="Azure cost query"): await _driver().list_billing_daily( date(2026, 1, 1), date(2026, 1, 2), scope=scope, ) # ── list_scope_grants (reverse discovery) ───────────────────── @pytest.mark.asyncio @respx.mock async def test_list_scope_grants_resolves_users_via_graph_batch() -> None: """Happy path: two user assignments + one service-principal are returned by ARM, only the users make it into the emitted grants, and both principal ids are resolved in a single Graph batch.""" _mock_token() sub_path = f"/subscriptions/{_SUB_ID}" respx.mock.get( url__regex=( f"^{_ARM_BASE}{sub_path}/providers/Microsoft.Authorization/" "roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": [ { "id": f"{sub_path}/providers/.../roleAssignments/ra1", "name": "ra1", "properties": { "principalId": "uid-alice", "principalType": "User", "roleDefinitionId": ( "/subscriptions/x/providers/Microsoft.Authorization/" "roleDefinitions/reader-guid" ), "scope": sub_path, "createdOn": "2026-03-15T09:00:00Z", }, }, { "id": f"{sub_path}/providers/.../roleAssignments/ra2", "name": "ra2", "properties": { "principalId": "uid-bob", "principalType": "User", # Real Azure Owner built-in role GUID — must match the # driver's ``owner_role_keys`` so auto-owner assignment # works on real tenants. "roleDefinitionId": ( "/providers/Microsoft.Authorization/" "roleDefinitions/" "8e3af657-a8ff-443c-a75c-2fe8c4bcb635" ), "scope": sub_path, }, }, { "id": f"{sub_path}/providers/.../roleAssignments/ra3", "name": "ra3", "properties": { "principalId": "uid-sp", "principalType": "ServicePrincipal", # must be skipped "roleDefinitionId": ( "/providers/Microsoft.Authorization/" "roleDefinitions/reader-guid" ), "scope": sub_path, }, }, ]})) respx.mock.post( f"{_GRAPH_BASE}/directoryObjects/getByIds", ).mock(return_value=httpx.Response(200, json={ "value": [ {"id": "uid-alice", "mail": "Alice@example.com"}, { "id": "uid-bob", "userPrincipalName": "bob@example.com", "mail": None, }, ], })) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) grants = await _driver().list_scope_grants(scope) assert len(grants) == 2 by_email = {g.principal_email: g for g in grants} assert set(by_email) == {"alice@example.com", "bob@example.com"} alice = by_email["alice@example.com"] # Subscription-prefixed role definition id must be normalized to # the tenant-scoped form so it matches the roles catalog + the # owner_role_keys set. assert alice.role_key.startswith("/providers/Microsoft.Authorization/") assert alice.external_ref == f"{sub_path}::ra1" bob = by_email["bob@example.com"] assert bob.role_key in AzureCloudDriver.owner_role_keys # `granted_at` parsed where provided. assert alice.granted_at is not None assert alice.granted_at.year == 2026 @pytest.mark.asyncio @respx.mock async def test_list_scope_grants_graph_403_raises_provider_auth_error() -> None: """Cycle 2 audit fix: a Graph 403 (or any non-200) means we can't resolve principal_email for SOME principals — returning ``[]`` silently looks like "no users assigned", which the grants-sync sweep would interpret as drift and mass-revoke. Raise loudly so the sweep aborts with a partial-failure status instead of writing a corrupting empty snapshot. """ from app.domains.cmdb.providers.exceptions import ProviderAuthError _mock_token() sub_path = f"/subscriptions/{_SUB_ID}" respx.mock.get( url__regex=( f"^{_ARM_BASE}{sub_path}/providers/Microsoft.Authorization/" "roleAssignments" ), ).mock(return_value=httpx.Response(200, json={"value": [ { "id": f"{sub_path}/.../roleAssignments/ra1", "name": "ra1", "properties": { "principalId": "uid-alice", "principalType": "User", "roleDefinitionId": "/providers/.../owner-guid", "scope": sub_path, }, }, ]})) respx.mock.post( f"{_GRAPH_BASE}/directoryObjects/getByIds", ).mock(return_value=httpx.Response(403, json={"error": "forbidden"})) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(ProviderAuthError) as exc_info: await _driver().list_scope_grants(scope) assert "Directory.Read.All" in str(exc_info.value) or "403" in str(exc_info.value) # ─── Microsoft Foundry auto-provisioning (apikey autogrant) ───────── @pytest.mark.asyncio @respx.mock async def test_provision_foundry_account_sends_kind_aiservices_body() -> None: """Lock the ARM PUT body to ``kind=AIServices`` + ``allowProjectManagement=True`` + ``identity=SystemAssigned``. A future refactor that flips ``kind`` back to ``OpenAI`` (the pre-Foundry shape) or drops ``allowProjectManagement`` would silently break Foundry-project provisioning — the project PUT returns 400 ``ProjectsNotEnabled`` against an account that lacks that flag.""" _mock_token() captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = request.read() captured["url"] = str(request.url) return httpx.Response(200, json={ "id": ( f"/subscriptions/{_SUB_ID}/resourceGroups/myrg/providers/" f"Microsoft.CognitiveServices/accounts/myacc" ), "name": "myacc", "kind": "AIServices", "identity": {"type": "SystemAssigned"}, "location": "eastus2", "sku": {"name": "S0"}, "properties": { "provisioningState": "Succeeded", "allowProjectManagement": True, "customSubDomainName": "myacc", "endpoint": "https://myacc.cognitiveservices.azure.com/", }, }) respx.mock.put( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/resourceGroups/myrg/" "providers/Microsoft.CognitiveServices/accounts/myacc" ), ).mock(side_effect=handler) result = await _driver().provision_foundry_account( subscription_id=_SUB_ID, resource_group="myrg", account_name="myacc", location="eastus2", custom_subdomain="myacc", tags={"helix-assignment-id": "test"}, ) import json as _json body = _json.loads(captured["body"].decode("utf-8")) assert body["kind"] == "AIServices", ( "Foundry resources MUST be kind=AIServices, not the legacy " "kind=OpenAI shape — Microsoft is consolidating Azure OpenAI " "Service under the Foundry surface." ) assert body["properties"]["allowProjectManagement"] is True, ( "Foundry-project sub-resources require allowProjectManagement " "on the parent account; without it, the project PUT returns " "400 ProjectsNotEnabled." ) assert body["identity"] == {"type": "SystemAssigned"} assert body["sku"] == {"name": "S0"} assert body["location"] == "eastus2" assert body["properties"]["customSubDomainName"] == "myacc" assert body["properties"]["publicNetworkAccess"] == "Enabled" assert body["properties"]["disableLocalAuth"] is False assert body["tags"]["helix-assignment-id"] == "test" # api-version pinned so a silent SDK churn doesn't break us. assert "api-version=2025-06-01" in captured["url"] # Returned shape passes through. assert result["kind"] == "AIServices" @pytest.mark.asyncio @respx.mock async def test_provision_foundry_project_sends_displayname_body() -> None: """Lock the project ARM PUT body shape: ``displayName`` + ``description`` in properties, ``identity=SystemAssigned``, api-version 2025-06-01. The project URL must include the parent account name (``/accounts/{account}/projects/{project}``).""" _mock_token() captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = request.read() captured["url"] = str(request.url) return httpx.Response(200, json={ "id": ( f"/subscriptions/{_SUB_ID}/resourceGroups/myrg/providers/" "Microsoft.CognitiveServices/accounts/myacc/projects/myproj" ), "name": "myproj", "type": "Microsoft.CognitiveServices/accounts/projects", "location": "eastus2", "identity": {"type": "SystemAssigned"}, "properties": { "provisioningState": "Succeeded", "displayName": "AK-X · Test Project", "description": "Auto-provisioned", }, }) respx.mock.put( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/resourceGroups/myrg/" "providers/Microsoft.CognitiveServices/accounts/myacc/" "projects/myproj" ), ).mock(side_effect=handler) result = await _driver().provision_foundry_project( subscription_id=_SUB_ID, resource_group="myrg", account_name="myacc", project_name="myproj", location="eastus2", display_name="AK-X · Test Project", description="Auto-provisioned", tags={"helix-request-number": "AK-X"}, ) import json as _json body = _json.loads(captured["body"].decode("utf-8")) assert body["properties"]["displayName"] == "AK-X · Test Project" assert body["properties"]["description"] == "Auto-provisioned" assert body["identity"] == {"type": "SystemAssigned"} assert body["location"] == "eastus2" assert body["tags"]["helix-request-number"] == "AK-X" # Project URL nests under the account. assert "/accounts/myacc/projects/myproj" in captured["url"] assert "api-version=2025-06-01" in captured["url"] assert result["name"] == "myproj" @pytest.mark.asyncio @respx.mock async def test_list_foundry_account_keys_returns_key1_key2() -> None: """Validates the listKeys POST shape + response parsing. Keys live on the parent account, not the project — the project URL embeds the project name and the key authenticates against the account- level endpoint.""" _mock_token() respx.mock.post( url__regex=( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/resourceGroups/myrg/" "providers/Microsoft.CognitiveServices/accounts/myacc/listKeys" ), ).mock(return_value=httpx.Response(200, json={ "key1": "abcdef1234567890" * 2, "key2": "fedcba0987654321" * 2, })) keys = await _driver().list_foundry_account_keys( subscription_id=_SUB_ID, resource_group="myrg", account_name="myacc", ) assert keys["key1"] == "abcdef1234567890" * 2 assert keys["key2"] == "fedcba0987654321" * 2 # ── list_cost_facts: 429 throttle handling ──────────────────── @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_retries_on_429_honouring_retry_after( monkeypatch: pytest.MonkeyPatch, ) -> None: """Regression for the 2026-05-11 prod failures: the ``cmdb.cost_facts_sync`` cron 429'd against ``Microsoft.CostManagement/query`` on 3 of 4 subscriptions in a single run because the driver raised immediately on 429 — no Retry-After honouring, no exponential backoff. Post-fix the driver retries up to 5 times honouring ``Retry-After`` (canonical HTTP header) and the per-scope / clienttype / tenant ``x-ms-ratelimit-microsoft.costmanagement-*-retry-after`` headers, so a single 429 → wait-then-retry round resolves the call cleanly instead of polluting ``scheduled_job_runs.errors`` with a partial failure.""" _mock_token() # Skip the actual sleep — we just need to assert the call sequence. sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) responses = iter([ httpx.Response( 429, headers={"Retry-After": "7"}, json={"error": {"code": "429", "message": "Too many requests"}}, ), httpx.Response(200, json={"properties": { "columns": [ {"name": "PreTaxCost"}, {"name": "CostUSD"}, {"name": "UsageDate"}, {"name": "Currency"}, {"name": "ResourceId"}, {"name": "ServiceName"}, {"name": "MeterCategory"}, {"name": "MeterSubCategory"}, {"name": "ResourceLocation"}, ], "rows": [[ 10.0, 10.0, 20260510, "USD", ( f"/subscriptions/{_SUB_ID}/resourceGroups/rg1/providers/" "Microsoft.Compute/virtualMachines/vm-a" ), "Virtual Machines", "Compute", "VM A-Series", "eastus", ]], }}), ]) respx.mock.post(url__regex=query_url_regex).mock( side_effect=lambda _req: next(responses), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert sleeps == [7], ( "driver must honour Retry-After=7 from the throttle response " f"before retrying; got sleeps={sleeps!r}" ) assert len(facts) == 1 assert facts[0].amount_usd == pytest.approx(10.0) @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_prefers_largest_costmanagement_retry_after( monkeypatch: pytest.MonkeyPatch, ) -> None: """When multiple throttle headers are present, the driver waits the LONGEST cooldown — backing off less than the server told us guarantees another 429 on the retry.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) responses = iter([ httpx.Response(429, headers={ "Retry-After": "5", "x-ms-ratelimit-microsoft.costmanagement-entity-retry-after": "12", "x-ms-ratelimit-microsoft.costmanagement-tenant-retry-after": "30", }, text="rate limited"), httpx.Response(200, json={"properties": { "columns": [{"name": "UsageDate"}, {"name": "PreTaxCost"}], "rows": [], }}), ]) respx.mock.post(url__regex=query_url_regex).mock( side_effect=lambda _req: next(responses), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert sleeps == [30], ( "driver must wait the LARGEST of Retry-After / entity / " f"tenant headers (=30s); got sleeps={sleeps!r}" ) @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_gives_up_after_5_throttle_retries( monkeypatch: pytest.MonkeyPatch, ) -> None: """Retry budget caps at 5 — a permanently-throttled scope must surface as a real failure rather than wedge the cron run forever. The cron handler then records the per-scope error in ``scheduled_job_runs.errors``, keeping the partial-failure semantics intact for the other subs in the same run.""" _mock_token() async def _no_sleep(_seconds: float) -> None: return None monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock( return_value=httpx.Response( 429, headers={"Retry-After": "1"}, text="permanently throttled", ), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(RuntimeError, match="status=429"): await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) # ── list_cost_facts: transport-error retry (httpx.ConnectError etc) ── @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_retries_on_transport_connect_error( monkeypatch: pytest.MonkeyPatch, ) -> None: """Regression for the 2026-05-21 + 2026-05-22 prod incident: the ``cmdb.cost_facts_sync`` cron failed two consecutive nights on the same subscription (PingCAP Azure OpenAI, 7bc8ab17-...) with ``httpx.ConnectError: All connection attempts failed`` — a TCP-level flake the pre-PR retry loop did not catch (it only handled HTTP 429 / 5xx, never reached on transport exception). Post-fix ``_post_with_retry`` catches the umbrella transport exceptions tuple ``_TRANSPORT_RETRY_EXCEPTIONS`` (ConnectError / ReadTimeout / RemoteProtocolError / etc.) and retries with exponential backoff ``min(2 ** attempt, 60)`` — independent of the 429/5xx Retry-After path.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) # First call raises ConnectError (the prod failure mode), second # call returns a clean 200 — driver must absorb the transport # error, sleep, and retry. call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.ConnectError("All connection attempts failed") return httpx.Response(200, json={"properties": { "columns": [ {"name": "PreTaxCost"}, {"name": "CostUSD"}, {"name": "UsageDate"}, {"name": "Currency"}, {"name": "ResourceId"}, {"name": "ServiceName"}, {"name": "MeterCategory"}, {"name": "MeterSubCategory"}, {"name": "ResourceLocation"}, ], "rows": [[ 7.5, 7.5, 20260510, "USD", ( f"/subscriptions/{_SUB_ID}/resourceGroups/rg1/providers/" "Microsoft.Compute/virtualMachines/vm-x" ), "Virtual Machines", "Compute", "VM A-Series", "eastus", ]], }}) respx.mock.post(url__regex=query_url_regex).mock(side_effect=_side_effect) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) # First retry uses exp backoff: min(2 ** 0, 60) = 1 second. assert sleeps == [1], ( "transport-error retry must use exponential backoff (1s on " f"attempt 0), not honour any header; got sleeps={sleeps!r}" ) assert len(facts) == 1 assert facts[0].amount_usd == pytest.approx(7.5) assert call_count["n"] == 2, "must retry exactly once after ConnectError" @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_retries_on_read_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: """Mid-stream socket loss (httpx.ReadTimeout) — same retry class as ConnectError but a different exception type. Catching the umbrella tuple covers both.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.ReadTimeout("timed out reading response") return httpx.Response(200, json={"properties": { "columns": [{"name": "PreTaxCost"}, {"name": "UsageDate"}], "rows": [], }}) respx.mock.post(url__regex=query_url_regex).mock(side_effect=_side_effect) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert sleeps == [1] assert facts == [] assert call_count["n"] == 2 @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_gives_up_after_5_transport_errors( monkeypatch: pytest.MonkeyPatch, ) -> None: """5-attempt budget caps transport retry too — a permanently transport-broken endpoint surfaces as a per-scope error so the partial-failure notifier fires (1 of N failed) instead of wedging the cron forever. Verifies the ORIGINAL httpx exception class survives — the failure-notifier's ``format_exception_for_log`` needs the real ``ConnectError`` type to land in the dedup key, not a generic RuntimeError wrapper.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock( side_effect=httpx.ConnectError("permanent TCP fail"), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(httpx.ConnectError, match="permanent TCP fail"): await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) # 4 sleeps between 5 attempts: 1, 2, 4, 8 seconds. assert sleeps == [1, 2, 4, 8], ( "transport-error backoff schedule must be 2^attempt; " f"got sleeps={sleeps!r}" ) # ── list_billing_daily / list_billing_by_dimension: transport retry ── @pytest.mark.asyncio @respx.mock async def test_list_billing_daily_retries_on_transport_error( monkeypatch: pytest.MonkeyPatch, ) -> None: """Pre-PR ``list_billing_daily`` had NO retry at all — a single transport flake dropped the day's billing-daily ingest. Post-PR it shares ``_post_with_retry`` with the cost path.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.ConnectError("DNS resolution failed") return httpx.Response(200, json={"properties": { "columns": [ {"name": "PreTaxCost"}, {"name": "UsageDate"}, {"name": "ServiceName"}, {"name": "Currency"}, ], "rows": [[3.25, 20260510, "Virtual Machines", "USD"]], }}) respx.mock.post(url__regex=url_regex).mock(side_effect=_side_effect) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) rows = await _driver().list_billing_daily( date(2026, 5, 10), date(2026, 5, 10), scope=scope, ) assert sleeps == [1] assert call_count["n"] == 2 assert len(rows) == 1 assert rows[0].total_cost_usd == pytest.approx(3.25) @pytest.mark.asyncio @respx.mock async def test_list_billing_by_dimension_retries_on_transport_error( monkeypatch: pytest.MonkeyPatch, ) -> None: """Same retry contract for ``list_billing_by_dimension`` (admin / dashboard surface, not on the cron path but same backend).""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.RemoteProtocolError("server disconnected mid-stream") return httpx.Response(200, json={"properties": { "columns": [ {"name": "PreTaxCost"}, {"name": "UsageDate"}, {"name": "ServiceName"}, {"name": "Currency"}, ], "rows": [[1.10, 20260510, "Storage", "USD"]], }}) respx.mock.post(url__regex=url_regex).mock(side_effect=_side_effect) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) rows = await _driver().list_billing_by_dimension( date(2026, 5, 10), date(2026, 5, 10), group_by=["service"], scope=scope, ) assert sleeps == [1] assert call_count["n"] == 2 assert len(rows) == 1 # ── _token: transport-error retry (cascades into every driver call) ─ @pytest.mark.asyncio @respx.mock async def test_token_retries_on_transport_error( monkeypatch: pytest.MonkeyPatch, ) -> None: """The token POST to login.microsoftonline.com is the prerequisite for EVERY other call in the driver. Pre-PR a transport flake here cascaded to every scope in the same cron tick (38-of-38 fail rather than 1-of-38). Post-PR ``_post_with_retry`` covers the AAD endpoint too.""" sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) token_url = f"{_LOGIN_BASE}/{_CFG['tenant_id']}/oauth2/v2.0/token" call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.ConnectError("TLS handshake failed") return httpx.Response(200, json={ "access_token": "tok-after-retry", "token_type": "Bearer", "expires_in": 3599, }) respx.mock.post(token_url).mock(side_effect=_side_effect) # Token cache is per-driver-instance, so a fresh driver forces a # real POST instead of returning a cached token. driver = _driver() token = await driver._token(_ARM_BASE) assert token == "tok-after-retry" assert sleeps == [1] assert call_count["n"] == 2 # Code-review #3: confirm the cache landed correctly after retry — # a regression that broke cache writes post-retry would silently # re-POST AAD on every subsequent call (cascading load on the AAD # endpoint that already throttled us). cached = driver._token_cache.get(_ARM_BASE) assert cached is not None, "token cache must be populated after retry" assert cached[0] == "tok-after-retry" # ── PoolTimeout (resource exhaustion in our own pool) ──────────── @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_retries_on_pool_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: """``httpx.PoolTimeout`` fires when the local connection pool can't lease a socket within the configured timeout — usually upstream is slow and earlier requests are still in flight. Treating it as transient gives the pool time to drain. Code-review #4 explicitly asked for this case to be exercised.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.PoolTimeout("pool exhausted") return httpx.Response(200, json={"properties": { "columns": [{"name": "PreTaxCost"}, {"name": "UsageDate"}], "rows": [], }}) respx.mock.post(url__regex=query_url_regex).mock(side_effect=_side_effect) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert sleeps == [1] assert facts == [] assert call_count["n"] == 2 # ── Mixed retry: transport-then-429-then-success ───────────────── @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_handles_transport_then_429_then_success( monkeypatch: pytest.MonkeyPatch, ) -> None: """The two retry paths (transport-error + HTTP-status) share one ``attempt`` counter inside ``_post_with_retry``. Code-review #4 flagged the interaction was untested — verify a single call that hits ConnectError, then 429, then 200 still resolves within the 5-attempt budget AND uses the right backoff for each (exp for transport, Retry-After for 429).""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) call_count = {"n": 0} def _side_effect(_req: httpx.Request) -> httpx.Response: call_count["n"] += 1 if call_count["n"] == 1: raise httpx.ConnectError("first attempt — TCP fail") if call_count["n"] == 2: return httpx.Response( 429, headers={"Retry-After": "3"}, text="throttled", ) return httpx.Response(200, json={"properties": { "columns": [{"name": "PreTaxCost"}, {"name": "UsageDate"}], "rows": [], }}) respx.mock.post(url__regex=query_url_regex).mock(side_effect=_side_effect) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) # attempt 0: transport, exp = min(2**0, 60) = 1 # attempt 1: 429, Retry-After=3 (honoured directly, within ceiling) # attempt 2: 200, no sleep assert sleeps == [1, 3], ( "must use exp backoff for transport then honour Retry-After " f"for 429; got sleeps={sleeps!r}" ) assert facts == [] assert call_count["n"] == 3 # ── 429 with Retry-After exceeding the honour ceiling ─────────────── @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_skips_retry_when_retry_after_exceeds_ceiling( monkeypatch: pytest.MonkeyPatch, ) -> None: """Code-review #1 fix: if the server-stated cooldown exceeds our honour ceiling (90s), retrying inside the ceiling is guaranteed to 429 again — load shedder hasn't moved its window. Better to surface the throttle immediately so the failure-notifier fires once with the real "Retry-After=600s" context. Pre-fix: cron would burn all 5 attempts with 90s sleeps, then raise — total ~7.5 min worker stall, 5 guaranteed-throttled requests on a tenant already crying for mercy. Post-fix: fail fast on attempt 1, the cron retries tomorrow.""" _mock_token() sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock( return_value=httpx.Response( 429, headers={"Retry-After": "600"}, text="tenant throttled", ), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(RuntimeError, match="status=429"): await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) # Zero sleeps — we surfaced immediately on attempt 1 without ever # honouring the 600s wait. assert sleeps == [], ( "must NOT sleep when Retry-After exceeds the 90s honour ceiling; " f"got sleeps={sleeps!r}" ) # ── _post_with_retry: defensive — buggy parse_retry_after callback ── @pytest.mark.asyncio async def test_post_with_retry_falls_back_when_parse_retry_after_raises( monkeypatch: pytest.MonkeyPatch, ) -> None: """A buggy ``parse_retry_after`` callback must not break the retry loop. Defensive guard so a regression in the header parser only degrades to exponential backoff, never breaks the retry budget.""" from app.integrations.drivers.azure_cloud import _post_with_retry sleeps: list[int] = [] async def _no_sleep(seconds: float) -> None: sleeps.append(int(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) def _broken_parser(_headers: httpx.Headers) -> int | None: raise ValueError("parser blew up") class _StubClient: def __init__(self) -> None: self.calls = 0 async def post(self, *_args: Any, **_kwargs: Any) -> httpx.Response: self.calls += 1 if self.calls == 1: # 429 triggers HTTP-status retry; broken parser must # not propagate — should fall back to exponential. return httpx.Response(429, text="throttled") return httpx.Response(200, json={"ok": True}) client = _StubClient() resp = await _post_with_retry( client, # type: ignore[arg-type] "https://example.com/query", json={"k": "v"}, endpoint_label="defensive_test", parse_retry_after=_broken_parser, ) assert resp.status_code == 200 assert client.calls == 2 # Fell back to exp: min(2 ** 0, 60) = 1 assert sleeps == [1] # ── list_cost_facts: 折扣前(原价)gross via Retail Prices meter-join ── # # The Cost Management ``query`` API exposes only POST-discount metrics # (Cost / PreTaxCost / CostUSD). To surface a pre-discount list price for # the AI dashboard's Azure Foundry bucket the driver augments the SAME # query with a ``MeterId`` grouping + ``UsageQuantity`` aggregation, then # joins each meter's usage to the public Azure Retail Prices catalogue: # gross_usd = Σ(qty × PAYG unit price). Forward-only — historical # aggregated rows carry no per-meter quantity and stay NULL (the dashboard # already handles gross=NULL via gross_coverage_pct). Mirrors the AWS # gross tests in test_aws_cloud_bq_billing.py. _RETAIL_RE = r"^https://prices\.azure\.com/api/retail/prices.*" _M1 = "0000aaaa-1111-2222-3333-444444444444" _M2 = "0000bbbb-1111-2222-3333-555555555555" _RID = ( f"/subscriptions/{_SUB_ID}/resourceGroups/rg1/providers/" "Microsoft.CognitiveServices/accounts/foundry-a" ) def _gross_query_response(rows: list[list[Any]]) -> httpx.Response: """200 response shaped like the AUGMENTED CM query (MeterId grouping + UsageQuantity aggregation present).""" return httpx.Response(200, json={"properties": { "columns": [ {"name": "PreTaxCost"}, {"name": "CostUSD"}, {"name": "UsageQuantity"}, {"name": "UsageDate"}, {"name": "Currency"}, {"name": "ResourceId"}, {"name": "ServiceName"}, {"name": "MeterCategory"}, {"name": "MeterSubCategory"}, {"name": "ResourceLocation"}, {"name": "MeterId"}, ], "rows": rows, }}) def _retail_response(prices: dict[str, float]) -> httpx.Response: return httpx.Response(200, json={ "Items": [ { "meterId": mid, "retailPrice": price, "tierMinimumUnits": 0.0, "priceType": "Consumption", "unitOfMeasure": "1K", "currencyCode": "USD", } for mid, price in prices.items() ], "NextPageLink": None, }) @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_emits_gross_via_retail_join() -> None: """Two meter-split rows for one fact grain collapse to a single fact; gross_usd = Σ(qty × retail price) across both meters, net unchanged.""" _mock_token() query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock( return_value=_gross_query_response([ # M1: qty 100, M2: qty 10 — same grain → one collapsed fact. [40.0, 40.0, 100.0, 20260510, "USD", _RID, "Azure OpenAI", "Cognitive Services", "S0 Tokens", "eastus", _M1], [20.0, 20.0, 10.0, 20260510, "USD", _RID, "Azure OpenAI", "Cognitive Services", "S0 Tokens", "eastus", _M2], ]), ) respx.mock.get(url__regex=_RETAIL_RE).mock( return_value=_retail_response({_M1: 0.5, _M2: 8.0}), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert len(facts) == 1, "meter-split rows must collapse to the fact grain" f = facts[0] # Net is the discounted sum, unchanged by the augmentation. assert f.amount_usd == pytest.approx(60.0) assert f.amount_native == pytest.approx(60.0) # Gross = 100*0.5 + 10*8.0 = 130 (USD-billed → native == usd). assert f.amount_usd_gross == pytest.approx(130.0) assert f.amount_native_gross == pytest.approx(130.0) @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_gross_null_when_meter_price_absent() -> None: """Marketplace / support / unlisted meters have no Consumption price. The fact's gross stays NULL (unknown) rather than fabricated.""" _mock_token() query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock( return_value=_gross_query_response([ [40.0, 40.0, 100.0, 20260510, "USD", _RID, "Azure OpenAI", "Cognitive Services", "S0 Tokens", "eastus", _M1], ]), ) # Catalogue returns nothing for this meter. respx.mock.get(url__regex=_RETAIL_RE).mock( return_value=_retail_response({}), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, ) assert len(facts) == 1 assert facts[0].amount_usd == pytest.approx(40.0) assert facts[0].amount_usd_gross is None assert facts[0].amount_native_gross is None @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_gross_null_when_below_net_guard() -> None: """Honesty guard: list price must be ≥ contracted. A gross that comes out BELOW net (qty/unit mismatch on an exotic meter) is untrustworthy → dropped to NULL rather than surfaced too-low.""" _mock_token() query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock( return_value=_gross_query_response([ # qty 1 * price 5 = 5 gross, but net is 40 → implausible. [40.0, 40.0, 1.0, 20260510, "USD", _RID, "Azure OpenAI", "Cognitive Services", "S0 Tokens", "eastus", _M1], ]), ) respx.mock.get(url__regex=_RETAIL_RE).mock( return_value=_retail_response({_M1: 5.0}), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, ) assert len(facts) == 1 assert facts[0].amount_usd == pytest.approx(40.0) assert facts[0].amount_usd_gross is None @pytest.mark.asyncio @respx.mock async def test_list_cost_facts_falls_back_to_net_only_on_gross_400() -> None: """If the augmented grain (MeterId + UsageQuantity) is rejected (400), the driver re-issues the BASE query so net ingestion never breaks; gross stays NULL for the run.""" _mock_token() bodies: list[dict[str, Any]] = [] def handler(request: httpx.Request) -> httpx.Response: import json as _json body = _json.loads(request.content.decode("utf-8")) bodies.append(body) grouping_names = { g["name"] for g in body["dataset"]["grouping"] } if "MeterId" in grouping_names: return httpx.Response(400, json={"error": { "code": "BadRequest", "message": "Aggregation 'UsageQuantity' is not supported", }}) # Base query (no MeterId) → succeed with base columns. return httpx.Response(200, json={"properties": { "columns": [ {"name": "PreTaxCost"}, {"name": "CostUSD"}, {"name": "UsageDate"}, {"name": "Currency"}, {"name": "ResourceId"}, {"name": "ServiceName"}, {"name": "MeterCategory"}, {"name": "MeterSubCategory"}, {"name": "ResourceLocation"}, ], "rows": [[ 40.0, 40.0, 20260510, "USD", _RID, "Azure OpenAI", "Cognitive Services", "S0 Tokens", "eastus", ]], }}) query_url_regex = ( f"^{_ARM_BASE}/subscriptions/{_SUB_ID}/providers/" "Microsoft.CostManagement/query.+" ) respx.mock.post(url__regex=query_url_regex).mock(side_effect=handler) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) facts = await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, ) # First body augmented, second body base — fallback fired. assert len(bodies) == 2 assert "MeterId" in {g["name"] for g in bodies[0]["dataset"]["grouping"]} assert "MeterId" not in {g["name"] for g in bodies[1]["dataset"]["grouping"]} # Net intact, gross NULL. assert len(facts) == 1 assert facts[0].amount_usd == pytest.approx(40.0) assert facts[0].amount_usd_gross is None assert facts[0].amount_native_gross is None # ── Cost Management client-side pacing (prod 2026-09-04 throttle) ── # # Context for all five tests below: on 2026-09-04/05 the monthly # ``cmdb.cost_facts_monthclose`` fan-out logged 83 consecutive # "429 … sleeping 26-58s" lines — EVERY subscription 429'd on its # first Cost Management query, for 77 minutes straight. One sub # (7f070020) then exhausted its retry budget and failed the whole run. def _pacing_driver() -> AzureCloudDriver: """Driver with the opt-in Cost Management pacer enabled. Pacing ships OFF (see ``AzureCloudDriver.__init__``): the post-deploy verification run on 2026-09-05 hit the job's 1800s timeout, and while the limiter was partly hot from my own earlier run, it proved the "pacing is wall-clock-neutral" model can be wrong on a cron that cannot afford to find that out nightly. These tests therefore exercise the flag explicitly; the default is pinned by ``test_pacing_is_off_by_default``. """ return AzureCloudDriver( integration_id="azure-test", config={**_CFG, "cost_pacing_enabled": "true"}, ) def _cost_query_regex(sub: str) -> str: return ( f"^{_ARM_BASE}/subscriptions/{sub}/providers/" "Microsoft.CostManagement/query.+" ) def _empty_cost_page() -> httpx.Response: return httpx.Response(200, json={"properties": { "columns": [{"name": "UsageDate"}, {"name": "PreTaxCost"}], "rows": [], }}) def _record_sleeps(monkeypatch: pytest.MonkeyPatch) -> list[float]: """Patch the driver's ``asyncio.sleep`` and record every duration. Durations are floats: the pacing gate sleeps *until a deadline*, so a paced wait comes back a hair under the armed value (real time elapses between arming and waiting even though the patched sleep itself is instant). Assertions round. """ slept: list[float] = [] async def _no_sleep(seconds: float) -> None: slept.append(float(seconds)) monkeypatch.setattr( "app.integrations.drivers.azure_cloud.asyncio.sleep", _no_sleep, ) return slept @pytest.mark.asyncio @respx.mock async def test_headerless_429_reuses_last_server_stated_cooldown( monkeypatch: pytest.MonkeyPatch, ) -> None: """Regression for the retry-budget burn that failed sub 7f070020. Azure does not always attach a Retry-After header to a Cost Management 429. The old fallback was ``min(2 ** attempt, 60)`` starting at 1s, so that subscription's five attempts slept 56s → **2s** → 57s → **8s**: two of them could not possibly clear a ~60s throttle window, and it ran out of attempts and failed the run. Post-fix a header-less 429 reuses the last cooldown the server DID state, so every attempt is spent on a request that has a chance. """ _mock_token() slept = _record_sleeps(monkeypatch) responses = iter([ # Server states 45s … httpx.Response(429, headers={"Retry-After": "45"}, text="throttled"), # … then throttles again with NO header at all (the prod shape). httpx.Response(429, text="throttled, no header"), _empty_cost_page(), ]) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( side_effect=lambda _req: next(responses), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert [round(s) for s in slept] == [45, 45], ( "a header-less 429 must reuse the last server-stated cooldown " "(45s), not restart the exponential at 2s — that is exactly how " f"sub 7f070020 burned its retry budget; got {slept!r}" ) @pytest.mark.asyncio @respx.mock async def test_cooldown_learned_on_one_scope_paces_the_next_scope( monkeypatch: pytest.MonkeyPatch, ) -> None: """The learned cooldown must cross scope boundaries. ``cost_service`` reuses ONE driver instance for every subscription in a run. Pre-fix nothing carried "the server just told us to wait 45s" from subscription N to N+1, so N+1 fired immediately into a limiter already known to be closed and collected its own 429 — which is why the prod trace shows one 429 per subscription rather than one per run. Post-fix subscription 2 waits out the learned spacing BEFORE its first request, and gets a 200 on that single request. """ _mock_token() slept = _record_sleeps(monkeypatch) sub1 = iter([ httpx.Response(429, headers={"Retry-After": "45"}, text="throttled"), _empty_cost_page(), ]) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( side_effect=lambda _req: next(sub1), ) sub2_route = respx.mock.post( url__regex=_cost_query_regex(_SUB_ID_2), ).mock(return_value=_empty_cost_page()) driver = _pacing_driver() # ONE instance, as the cron does await driver.list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=CloudScopeRef( scope_type="azure_subscription", external_id=_SUB_ID, ), cny_per_usd=7.2, ) slept_on_sub1 = list(slept) slept.clear() await driver.list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=CloudScopeRef( scope_type="azure_subscription", external_id=_SUB_ID_2, ), cny_per_usd=7.2, ) assert [round(s) for s in slept_on_sub1] == [45] assert [round(s) for s in slept] == [45], ( "subscription 2 must wait out the cooldown subscription 1 " f"learned before issuing its first query; got {slept!r}" ) assert sub2_route.call_count == 1, ( "having waited, subscription 2 should need exactly one request " f"— no 429 round-trip; got {sub2_route.call_count}" ) @pytest.mark.asyncio @respx.mock async def test_pacing_decays_after_a_clean_success( monkeypatch: pytest.MonkeyPatch, ) -> None: """Pacing must probe back toward full speed (AIMD), not pin the rest of the run at the worst cooldown the tenant ever stated. Without the multiplicative decrease, one hostile minute would pace all 36 subscriptions at 45s each and make the run SLOWER than the 429-per-scope behaviour it replaces. """ _mock_token() slept = _record_sleeps(monkeypatch) sub1 = iter([ httpx.Response(429, headers={"Retry-After": "40"}, text="throttled"), _empty_cost_page(), ]) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( side_effect=lambda _req: next(sub1), ) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID_2)).mock( return_value=_empty_cost_page(), ) driver = _pacing_driver() for sub in (_SUB_ID, _SUB_ID_2, _SUB_ID_2, _SUB_ID_2): await driver.list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=CloudScopeRef( scope_type="azure_subscription", external_id=sub, ), cny_per_usd=7.2, ) # 40s retry backoff on sub1, then the paced waits: 40 (no decay — # that call needed a retry), then 20, then 10 as clean successes # halve the spacing. assert [round(s) for s in slept] == [40, 40, 20, 10], ( "spacing must halve after every retry-free success; got " f"{slept!r}" ) @pytest.mark.asyncio @respx.mock async def test_throttle_sleep_budget_stops_before_the_per_scope_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: """A hard-throttled scope must give up on its own terms. ``cost_service`` wraps each scope in a 300s timeout. Sleeping out four 57s cooldowns plus a paced wait plus the query overruns that, so the coroutine gets cancelled mid-sleep and the run records the useless ``status='timeout'``. The cumulative budget stops first, so the operator gets "429, throttled" instead. """ _mock_token() slept = _record_sleeps(monkeypatch) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( return_value=httpx.Response( 429, headers={"Retry-After": "57"}, text="hard throttled", ), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(RuntimeError, match="status=429"): await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) assert sum(slept) <= 200.0, ( "cumulative throttle sleep must stay inside the 200s budget so " "the scope raises a truthful 429 rather than being killed by " f"cost_service's 300s per-scope timeout; slept {sum(slept)}s" ) assert [round(s) for s in slept] == [57, 57, 57] @pytest.mark.asyncio @respx.mock async def test_429_error_names_the_throttle_headers( monkeypatch: pytest.MonkeyPatch, ) -> None: """The error that reaches ``scheduled_job_runs.errors`` must say which limiter fired. Azure's 429 body is content-free ("Too many requests. Please retry.") — the prod row on 2026-09-04 carried nothing else, so answering "did the server even tell us how long to wait?" needed a dig through container logs. The headers now travel with the error. """ _mock_token() _record_sleeps(monkeypatch) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( return_value=httpx.Response( 429, headers={ "x-ms-ratelimit-microsoft.costmanagement-clienttype-" "retry-after": "57", }, json={"error": { "code": "429", "message": "Too many requests. Please retry.", }}, ), ) scope = CloudScopeRef(scope_type="azure_subscription", external_id=_SUB_ID) with pytest.raises(RuntimeError) as excinfo: await _driver().list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=scope, cny_per_usd=7.2, ) msg = str(excinfo.value) assert "throttle_headers=clienttype=57" in msg, msg @pytest.mark.asyncio @respx.mock async def test_non_cost_endpoints_keep_the_plain_exponential_backoff( monkeypatch: pytest.MonkeyPatch, ) -> None: """Pacing is scoped to Cost Management. The AAD token endpoint has its own, far more generous limiter and passes no pacer, so a 429 there must still ramp 1s → 2s → … rather than inherit a Cost Management cooldown. """ slept = _record_sleeps(monkeypatch) responses = iter([ httpx.Response(429, text="aad throttled"), httpx.Response(429, text="aad throttled"), httpx.Response(200, json={ "access_token": "tok", "token_type": "Bearer", "expires_in": 3599, }), ]) respx.mock.post( f"{_LOGIN_BASE}/{_CFG['tenant_id']}/oauth2/v2.0/token" ).mock(side_effect=lambda _req: next(responses)) token = await _driver()._token(_ARM_BASE) assert token == "tok" assert [round(s) for s in slept] == [1, 2], ( "token endpoint must keep the un-paced exponential ramp; got " f"{slept!r}" ) @pytest.mark.asyncio @respx.mock async def test_pacing_reverts_to_reactive_once_the_run_budget_is_spent( monkeypatch: pytest.MonkeyPatch, ) -> None: """Pacing must not be able to push a cron past its own timeout. ``cmdb.cost_facts.azure-primary`` inherits the executor's 1800s default and finishes in 587-775s in prod. Pacing is *expected* to be wall-clock-neutral — it replaces 429 round-trips that were sleeping the same cooldown anyway — but that expectation is a model of Azure's limiter, and the model being wrong must not convert a healthy run into a timed-out one. Past ``_CM_PACE_RUN_BUDGET_SEC`` the driver reverts to the pre-pacing reactive behaviour. """ from app.integrations.drivers import azure_cloud as _az _mock_token() slept = _record_sleeps(monkeypatch) sub1 = iter([ httpx.Response(429, headers={"Retry-After": "60"}, text="throttled"), _empty_cost_page(), ]) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( side_effect=lambda _req: next(sub1), ) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID_2)).mock( return_value=_empty_cost_page(), ) driver = _pacing_driver() # Rebuild the pacer with a two-waits-worth budget. Passed # explicitly, NOT by patching ``_CM_PACE_RUN_BUDGET_SEC``: it is a # default argument, so Python bound it at def time and a # monkeypatch of the module constant would silently do nothing (it # did — this test read green against a 600s budget until the # expectation caught the extra wait). driver._cm_pacer = _az._AzureThrottlePacer( pacing_enabled=True, min_backoff_s=_az._CM_MIN_THROTTLE_BACKOFF_SEC, pace_run_budget_s=70.0, ) for sub in (_SUB_ID, _SUB_ID_2, _SUB_ID_2, _SUB_ID_2): await driver.list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=CloudScopeRef( scope_type="azure_subscription", external_id=sub, ), cny_per_usd=7.2, ) # 60s retry backoff, then paced 60 and 30 (=90s > the 70s budget, # which is only consulted BEFORE arming) — the fourth call finds the # budget spent and issues its request with no wait at all. assert [round(s) for s in slept] == [60, 60, 30], ( "once the run's pacing budget is spent the driver must stop " f"adding proactive waits; got {slept!r}" ) assert driver._cm_pacer._paced_total_s >= 70.0 @pytest.mark.asyncio @respx.mock async def test_pacing_is_off_by_default_but_headerless_fallback_still_works( monkeypatch: pytest.MonkeyPatch, ) -> None: """The default driver must add ZERO proactive waits. Pacing ships opt-in (``cost_pacing_enabled``) because the post-deploy verification run on 2026-09-05 hit the job's 1800s timeout — 27 of 36 subscriptions, against a 587-775s baseline — with the pacing budget never exhausted, i.e. pacing was spending headroom without preventing the 429s under that limiter. What must still work by default is the fix the reported incident actually needed: a header-less 429 reuses the last server-stated cooldown rather than restarting the exponential at 1s. That path runs through ``observe_throttle``, which learns regardless of the pacing flag — so this test pins BOTH halves: no proactive wait between scopes, and the learned fallback still applied. """ _mock_token() slept = _record_sleeps(monkeypatch) sub1 = iter([ httpx.Response(429, headers={"Retry-After": "45"}, text="throttled"), httpx.Response(429, text="throttled, no header"), _empty_cost_page(), ]) respx.mock.post(url__regex=_cost_query_regex(_SUB_ID)).mock( side_effect=lambda _req: next(sub1), ) sub2_route = respx.mock.post( url__regex=_cost_query_regex(_SUB_ID_2), ).mock(return_value=_empty_cost_page()) driver = _driver() # default config — pacing OFF for sub in (_SUB_ID, _SUB_ID_2): await driver.list_cost_facts( date(2026, 5, 10), date(2026, 5, 10), scope=CloudScopeRef( scope_type="azure_subscription", external_id=sub, ), cny_per_usd=7.2, ) assert [round(s) for s in slept] == [45, 45], ( "default driver must sleep ONLY its own retry backoff — 45s " "server-stated, then 45s learned for the header-less 429 — and " f"add no proactive wait before subscription 2; got {slept!r}" ) assert sub2_route.call_count == 1 assert driver._cm_pacer._pacing_enabled is False