"""Microsoft Foundry (Azure OpenAI) auto-grant: on approved-status callback, the helper provisions a fresh Resource Group + Foundry resource (Cognitive Services account of ``kind=AIServices`` with ``allowProjectManagement=True``) + Foundry project under the standing PingCAP Azure OpenAI subscription, mirrors the RG into ``cmdb_cloud_scopes`` as a child scope, writes a priority-10 cost- allocation rule, and stores ``KEY1`` on the assignment. We don't hit Azure ARM — the four driver methods (``ensure_resource_group`` / ``provision_foundry_account`` / ``provision_foundry_project`` / ``list_foundry_account_keys``) are patched to return canned payloads so the test stays hermetic. """ from __future__ import annotations import uuid from decimal import Decimal from unittest.mock import AsyncMock import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.apikey.azure_openai_autogrant import ( _build_resource_names, _is_azure_openai_provider, _slugify_for_azure, maybe_auto_grant_azure_openai, ) from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider from app.domains.cmdb.models import ( SCOPE_TYPE_AZURE_RESOURCE_GROUP, SCOPE_TYPE_AZURE_SUBSCRIPTION, CmdbCloudScope, CmdbCostAllocationRule, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration, PlatformSetting # Subscription GUID the autogrant defaults to (PingCAP Azure OpenAI). _SUB_ID = "7bc8ab17-59fb-44c1-92ca-61fd2b814735" async def _make_azure_integ( db_session: AsyncSession, slug: str = "azure-openai", ) -> Integration: integ = Integration( id=uuid.uuid4(), kind="cloud", provider="azure", name="Azure OpenAI (PingCAP)", slug=slug, enabled=True, is_primary=False, capabilities=["servers"], config_public={"tenant_id": "tenant-fake"}, secrets_encrypted="ciphertext-placeholder", secrets_nonce="nonce-placeholder", ) db_session.add(integ) await db_session.commit() return integ async def _make_subscription_scope( db_session: AsyncSession, integ: Integration, ) -> CmdbCloudScope: scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION, external_id=_SUB_ID, display_name="PingCAP Azure OpenAI", environment="dev", ) db_session.add(scope) await db_session.commit() return scope async def _make_person(db_session: AsyncSession) -> IdentityPerson: person = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name="Shuai Feng", work_email="shuai.feng@pingcap.com", employment_status="active", ) db_session.add(person) await db_session.commit() return person async def _make_aoai_assignment( db_session: AsyncSession, person_id: uuid.UUID, *, provider_name: str = "Azure OpenAI", ) -> ApiKeyAssignment: provider = ApiKeyProvider( id=uuid.uuid4(), name=provider_name, description="Azure OpenAI access", is_active=True, sort_order=0, ) db_session.add(provider) await db_session.commit() assignment = ApiKeyAssignment( id=uuid.uuid4(), provider_id=provider.id, assigned_to=person_id, status="approved", request_number=f"AK-20260510-{uuid.uuid4().hex[:4].upper()}", project_name="LLM eval", reason="Azure OpenAI model benchmarking", monthly_budget_usd=Decimal("250.00"), ) db_session.add(assignment) await db_session.commit() return assignment def _patch_driver_methods( monkeypatch, *, captured: dict, key1: str = "fake-key1-32hex", endpoint: str | None = None, ) -> None: """Patch the four ARM-touching driver methods with deterministic stubs.""" from app.integrations.drivers.azure_cloud import AzureCloudDriver async def fake_ensure_rg(self, *, subscription_id, name, location, tags): captured["rg_subscription"] = subscription_id captured["rg_name"] = name captured["rg_location"] = location captured["rg_tags"] = dict(tags or {}) return { "id": ( f"/subscriptions/{subscription_id}/resourceGroups/{name}" ), "name": name, "location": location, "properties": {"provisioningState": "Succeeded"}, } async def fake_provision_foundry_account( self, *, subscription_id, resource_group, account_name, location, custom_subdomain, tags, sku_name="S0", poll_seconds=300, ): captured["account_subscription"] = subscription_id captured["account_rg"] = resource_group captured["account_name"] = account_name captured["account_location"] = location captured["account_subdomain"] = custom_subdomain captured["account_tags"] = dict(tags or {}) ep = ( endpoint or f"https://{account_name}.cognitiveservices.azure.com/" ) return { "id": ( f"/subscriptions/{subscription_id}/resourceGroups/" f"{resource_group}/providers/" f"Microsoft.CognitiveServices/accounts/{account_name}" ), "name": account_name, "kind": "AIServices", "identity": {"type": "SystemAssigned"}, "location": location, "sku": {"name": sku_name}, "properties": { "provisioningState": "Succeeded", "allowProjectManagement": True, "customSubDomainName": custom_subdomain, "endpoint": ep, }, } async def fake_provision_foundry_project( self, *, subscription_id, resource_group, account_name, project_name, location, display_name, description="", tags=None, poll_seconds=180, ): captured["project_subscription"] = subscription_id captured["project_rg"] = resource_group captured["project_account"] = account_name captured["project_name"] = project_name captured["project_location"] = location captured["project_display_name"] = display_name captured["project_description"] = description captured["project_tags"] = dict(tags or {}) return { "id": ( f"/subscriptions/{subscription_id}/resourceGroups/" f"{resource_group}/providers/" f"Microsoft.CognitiveServices/accounts/{account_name}" f"/projects/{project_name}" ), "name": project_name, "type": "Microsoft.CognitiveServices/accounts/projects", "location": location, "identity": {"type": "SystemAssigned"}, "properties": { "provisioningState": "Succeeded", "displayName": display_name, "description": description, }, } async def fake_list_keys(self, *, subscription_id, resource_group, account_name): captured["keys_subscription"] = subscription_id captured["keys_rg"] = resource_group captured["keys_account"] = account_name return {"key1": key1, "key2": "fake-key2-32hex"} monkeypatch.setattr( AzureCloudDriver, "ensure_resource_group", fake_ensure_rg, ) monkeypatch.setattr( AzureCloudDriver, "provision_foundry_account", fake_provision_foundry_account, ) monkeypatch.setattr( AzureCloudDriver, "provision_foundry_project", fake_provision_foundry_project, ) monkeypatch.setattr( AzureCloudDriver, "list_foundry_account_keys", fake_list_keys, ) # Skip real credential decryption — the driver methods don't read # credentials in these stubs. monkeypatch.setattr( "app.domains.apikey.azure_openai_autogrant.decrypt_fields", lambda cipher, nonce, *, integration_id=None: { "tenant_id": "tenant-fake", "client_id": "client-fake", "client_secret": "secret-fake", }, ) # ─── Pure-function tests ──────────────────────────────────────────── def test_is_azure_openai_provider_matches_common_variants() -> None: """Admins rename the provider to multiple display names — accept every common token.""" class _P: def __init__(self, name: str) -> None: self.name = name assert _is_azure_openai_provider(_P("Azure OpenAI")) assert _is_azure_openai_provider(_P("Azure-OpenAI Service")) assert _is_azure_openai_provider(_P("AOAI")) assert _is_azure_openai_provider(_P("AzureOpenAI Service")) assert _is_azure_openai_provider(_P("aoai (eastus)")) # Whitespace-quirk regression — prod incident AK-20260509-EF8A had # provider name "Azure OpenAI" (double space) and the matcher # silently short-circuited, skipping RG auto-grant entirely. The # normalised matcher must accept double space, NBSP, full-width # space, and tabs. assert _is_azure_openai_provider(_P("Azure OpenAI")) # 2 spaces assert _is_azure_openai_provider(_P("Azure OpenAI")) # NBSP assert _is_azure_openai_provider(_P("Azure　OpenAI")) # full-width assert _is_azure_openai_provider(_P("Azure\tOpenAI")) # tab assert _is_azure_openai_provider(_P(" Azure OpenAI ")) # padding assert not _is_azure_openai_provider(_P("AWS Bedrock")) assert not _is_azure_openai_provider(_P("Aliyun Bailian")) assert not _is_azure_openai_provider(_P("OpenAI")) # no "azure" prefix assert not _is_azure_openai_provider(None) def test_slugify_for_azure_drops_cjk_and_punctuation() -> None: assert _slugify_for_azure("AI Support") == "ai-support" assert _slugify_for_azure(" AI Support ") == "ai-support" assert _slugify_for_azure("AI/Support+v2") == "ai-support-v2" assert _slugify_for_azure("项目名称") == "" # CJK-only → empty assert _slugify_for_azure("--foo--") == "foo" assert _slugify_for_azure("A!@#$%B") == "a-b" def test_build_resource_names_within_azure_limits() -> None: aid = uuid.UUID("12345678-1234-5678-1234-567812345678") rg, account, project = _build_resource_names( "AI Support", "AK-20260510-A4B2", assignment_id=aid, ) assert rg.startswith("helix-aoai-20260510-a4b2-ai-support") assert len(rg) <= 90 # Azure RG limit assert account.startswith("aoai-20260510-a4b2-ai-support-") assert len(account) <= 64 # Cognitive Services account limit # Foundry project name is the readable shape; no aid_tail. assert project == "proj-20260510-a4b2-ai-support" assert len(project) <= 64 # Foundry project name limit # All-CJK project falls through to suffix-only form. rg2, acc2, proj2 = _build_resource_names( "项目名称", "AK-20260510-FFFF", assignment_id=aid, ) assert rg2 == "helix-aoai-20260510-ffff" assert acc2.startswith("aoai-20260510-ffff-") assert proj2 == "proj-20260510-ffff" def test_build_resource_names_deterministic_per_assignment() -> None: """Concurrent Feishu-callback retries for the same assignment must compute the SAME account name so Azure's idempotent ARM PUT returns the existing resource instead of minting a duplicate billable account. The 8-hex tail is derived from ``assignment.id``, NOT random. Foundry project names also stay deterministic across retries (they derive purely from request_number + project slug — no random tail needed because they're scoped to the parent account). """ aid = uuid.UUID("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") _, acc_a, proj_a = _build_resource_names( "AI Support", "AK-20260510-A4B2", assignment_id=aid, ) _, acc_b, proj_b = _build_resource_names( "AI Support", "AK-20260510-A4B2", assignment_id=aid, ) assert acc_a == acc_b, "same assignment must yield same account name" assert proj_a == proj_b, "same assignment must yield same project name" # Different assignments yield different account names (collision # defence on the globally-unique subdomain). Foundry project names # don't need to differ because each lives under its own account. aid2 = uuid.UUID("ffffffff-ffff-ffff-ffff-ffffffffffff") _, acc_c, proj_c = _build_resource_names( "AI Support", "AK-20260510-A4B2", assignment_id=aid2, ) assert acc_a != acc_c assert proj_a == proj_c # same request → same project name OK # Tail matches the first 8 hex of the UUID (assignment-id-derived). assert acc_a.endswith(aid.hex[:8]) assert acc_c.endswith(aid2.hex[:8]) def test_foundry_project_endpoint_format() -> None: """The project's inference URL is constructed client-side as ``https://{accountSubdomain}.services.ai.azure.com/api/projects/{name}`` — bypasses the project's ``properties.endpoints`` dict because the schema has shifted across api-versions and not every key in that dict is the inference endpoint.""" from app.domains.apikey.azure_openai_autogrant import ( _foundry_project_endpoint, ) assert _foundry_project_endpoint("aoai-20260510-a4b2-foo-1234abcd", "proj-20260510-a4b2-foo") == ( "https://aoai-20260510-a4b2-foo-1234abcd.services.ai.azure.com" "/api/projects/proj-20260510-a4b2-foo" ) # ─── Async DB-backed integration tests ────────────────────────────── @pytest.mark.asyncio async def test_autogrant_succeeds_for_azure_openai( db_session: AsyncSession, monkeypatch, ): """Happy path: end-to-end provisioning writes RG scope, allocation rule, key bundle, and binds cloud_scope_id on the assignment.""" integ = await _make_azure_integ(db_session) sub_scope = await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) notify_spy = AsyncMock() from app.notifications.service import NotificationService monkeypatch.setattr(NotificationService, "dispatch", notify_spy) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result["status"] == "granted" assert assignment.status == "active" assert assignment.key_value is not None # Decrypt + validate the credential bundle. Both SDK styles must # ship working values: the Foundry SDK constructor args # (FOUNDRY_PROJECT_ENDPOINT + FOUNDRY_API_KEY — names match # Microsoft's docs convention since azure-ai-projects has no env- # var auto-discovery) and the openai client auto-discovery vars # (AZURE_OPENAI_API_KEY + AZURE_OPENAI_ENDPOINT — these the openai # SDK reads on its own). from app.domains.apikey.service import _decrypt_key_value decrypted = _decrypt_key_value( assignment.key_value, assignment_id=assignment.id, ) assert "FOUNDRY_API_KEY=fake-key1-32hex" in decrypted assert "FOUNDRY_PROJECT_ENDPOINT=" in decrypted assert "FOUNDRY_PROJECT_NAME=" in decrypted assert "FOUNDRY_ACCOUNT_NAME=" in decrypted assert "AZURE_OPENAI_API_KEY=fake-key1-32hex" in decrypted assert "AZURE_OPENAI_ENDPOINT=" in decrypted assert f"AZURE_SUBSCRIPTION_ID={_SUB_ID}" in decrypted # Empty AZURE_OPENAI_DEPLOYMENT_NAME would clobber a pre-existing # env var on ``source `` — every line containing that # var name must START with ``#`` so the shell skips it. Asserts # both the comment-presence and the line-start contract; a future # single-character edit that turns the leading ``# `` into a real # ``AZURE_OPENAI_DEPLOYMENT_NAME=`` (with the trailing inline # ``# uncomment + fill...`` comment intact) would not change the # presence assertion above but WOULD violate the line-start one # below. assert "# AZURE_OPENAI_DEPLOYMENT_NAME=" in decrypted assert "\nAZURE_OPENAI_DEPLOYMENT_NAME=\n" not in decrypted deployment_lines = [ line for line in decrypted.splitlines() if "AZURE_OPENAI_DEPLOYMENT_NAME" in line ] assert deployment_lines, "deployment-name placeholder line missing" for line in deployment_lines: assert line.lstrip().startswith("#"), ( f"AZURE_OPENAI_DEPLOYMENT_NAME line must be commented out " f"so source doesn't clobber a pre-existing value; " f"got: {line!r}" ) # The autogrant routed every ARM call (RG + account + project + # listKeys) at the configured sub. assert captured["rg_subscription"] == _SUB_ID assert captured["account_subscription"] == _SUB_ID assert captured["project_subscription"] == _SUB_ID assert captured["keys_subscription"] == _SUB_ID # Tags carry the chargeback dimensions on RG, AIServices account, # AND Foundry project (so finance can filter by any dimension). assert captured["rg_tags"]["helix-assignment-id"] == str(assignment.id) assert captured["rg_tags"]["helix-owner-email"] == person.work_email # Round-31 audit B7: ``Owner`` tag aligned with Bedrock convention # (= applicant email). Previously set to ``project_name`` which # broke cross-cloud Cost Management ``Owner`` reports — see # ROADMAP.md Round-31 section D. Project name is preserved on # ``helix-project-name`` for AOAI-specific drilldown. assert captured["rg_tags"]["Owner"] == person.work_email assert captured["account_tags"]["helix-project-name"] == "LLM eval" assert captured["project_tags"]["helix-project-name"] == "LLM eval" # Foundry project lives under the account we just provisioned. assert captured["project_account"] == captured["account_name"] assert captured["project_rg"] == captured["rg_name"] assert captured["project_display_name"].startswith( assignment.request_number, ) # cloud_scope_id is bound to the new RG scope. assert assignment.cloud_scope_id is not None rg_scope = (await db_session.execute( select(CmdbCloudScope).where( CmdbCloudScope.id == uuid.UUID(assignment.cloud_scope_id), ) )).scalar_one() assert rg_scope.scope_type == SCOPE_TYPE_AZURE_RESOURCE_GROUP assert rg_scope.parent_scope_id == str(sub_scope.id) # ``is_ai_dedicated`` removed in PR #253 / migration 152 — AI cost # is now identified by service-name classification, not by a # scope-level boolean. assert rg_scope.owner_person_id == str(person.id) assert rg_scope.budget_usd_monthly == assignment.monthly_budget_usd assert rg_scope.external_id == captured["rg_name"] # Allocation rule keyed on the RG name → applicant. rule = (await db_session.execute( select(CmdbCostAllocationRule).where( CmdbCostAllocationRule.match_resource_group == captured["rg_name"], ) )).scalar_one() assert rule.priority == 10 assert rule.match_provider == "azure" assert rule.match_integration_ids == [str(integ.id)] assert rule.assign_person_id == str(person.id) assert rule.enabled is True # Notification dispatched once with the right template + recipient. assert notify_spy.await_count == 1 env = notify_spy.await_args.args[0] assert env.recipient.id == person.id assert env.template_key == "apikey.ready" # Detail block records the granted=True flag + tombstone cleared, # and carries the full Foundry hierarchy so the admin UI / MCP # tool surface every level (RG → account → project) without # parsing the raw_data blob. The legacy ``account_name`` / # ``endpoint`` top-level keys from the kind=OpenAI era are NOT # aliased — the audit confirmed no consumer reads them. detail = assignment.detail or {} info = detail.get("azure_openai", {}) assert info["granted"] is True assert info["in_flight"] is False assert info["resource_group"] == captured["rg_name"] assert info["foundry_account_name"] == captured["account_name"] assert info["foundry_project_name"] == captured["project_name"] assert info["foundry_account_arm_id"].endswith( f"/accounts/{captured['account_name']}", ) assert info["foundry_project_arm_id"].endswith( f"/projects/{captured['project_name']}", ) assert info["foundry_project_endpoint"].startswith("https://") assert info["foundry_project_endpoint"].endswith( f"/api/projects/{captured['project_name']}", ) assert info["allocation_rule_id"] == rule.id # Legacy aliases must NOT be present — guards against accidental # reintroduction by a future merge. assert "account_name" not in info assert "endpoint" not in info assert "account_arm_id" not in info @pytest.mark.asyncio async def test_autogrant_reuses_allocation_rule_when_name_collides( db_session: AsyncSession, monkeypatch, ): """Defensive idempotency: an admin re-running the autogrant after manually clearing ``detail.azure_openai.granted`` must NOT insert a duplicate ``CmdbCostAllocationRule`` row. The rule name embeds the request_number + applicant email, so the second invocation finds the existing row by name and reuses it. Mirror of bedrock_autogrant.py:467-506 (added in PR #220). AOAI was missing the guard until the 2026-05-12 cost-attribution audit flagged it as a duplicate-rule risk on retried autogrants. """ integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) notify_spy = AsyncMock() from app.notifications.service import NotificationService monkeypatch.setattr(NotificationService, "dispatch", notify_spy) # First run — happy path mints the rule. result1 = await maybe_auto_grant_azure_openai(db_session, assignment) assert result1["status"] == "granted" rules_after_first = (await db_session.execute( select(CmdbCostAllocationRule), )).scalars().all() assert len(rules_after_first) == 1 first_rule_id = rules_after_first[0].id # Simulate admin clearing ``granted`` (e.g. via /admin/api-keys to # force a re-run after diagnosing a partial cloud state). detail = dict(assignment.detail or {}) detail.setdefault("azure_openai", {})["granted"] = False detail["azure_openai"]["in_flight"] = False assignment.detail = detail await db_session.commit() # Second run — must reuse the existing rule, not duplicate it. result2 = await maybe_auto_grant_azure_openai(db_session, assignment) assert result2["status"] == "granted" rules_after_second = (await db_session.execute( select(CmdbCostAllocationRule), )).scalars().all() assert len(rules_after_second) == 1 assert rules_after_second[0].id == first_rule_id @pytest.mark.asyncio async def test_autogrant_noop_for_non_azure_openai( db_session: AsyncSession, ): """Provider mismatch returns ``not-azure-openai`` without touching the assignment.""" person = await _make_person(db_session) other = ApiKeyProvider( id=uuid.uuid4(), name="OpenAI", description="", is_active=True, sort_order=0, ) db_session.add(other) await db_session.commit() assignment = ApiKeyAssignment( id=uuid.uuid4(), provider_id=other.id, assigned_to=person.id, status="approved", request_number="AK-X", project_name="x", reason="x", monthly_budget_usd=Decimal("10.00"), ) db_session.add(assignment) await db_session.commit() result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "not-azure-openai"} assert assignment.key_value is None assert assignment.cloud_scope_id is None @pytest.mark.asyncio async def test_autogrant_skipped_when_no_subscription_scope( db_session: AsyncSession, ): """No ``azure_subscription`` scope row matching the configured sub means the autogrant can't resolve the integration → ``no-subscription`` surfaces with a clear admin-facing error in detail.""" person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) # Don't seed any cloud integration / subscription scope. result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result["status"] == "no-subscription" detail = assignment.detail or {} info = detail.get("azure_openai", {}) assert info.get("granted") is False assert "default_subscription_id" in (info.get("error") or "") @pytest.mark.asyncio async def test_autogrant_idempotent_when_already_granted( db_session: AsyncSession, ): """Re-invoke after a successful run is a no-op (the granted flag short-circuits before any ARM call).""" person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) assignment.detail = {"azure_openai": {"granted": True}} await db_session.commit() result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "already-granted"} @pytest.mark.asyncio async def test_autogrant_refuses_retry_when_in_flight( db_session: AsyncSession, ): """Leak-safety: a previous callback **process-crashed** mid-call (after the in-flight tombstone was committed but before either the happy-path commit OR any of the per-error-branch in_flight=False clears). The next callback short-circuits with manual-recovery status — admins decide whether the prior attempt left orphan Azure resources. Note: a non-crash failure (RG / account / project / listKeys raised a normal exception) clears in_flight before returning — see ``test_autogrant_rg_failure_persists_diagnostic`` and ``test_autogrant_project_failure_persists_diagnostic_with_account_arm_id``. Those paths must NOT wedge the assignment. The genuinely-stuck case is the one this test exercises: in_flight=True with no accompanying ``error`` field, which is the signature of a mid-call SIGKILL / OOM where the autogrant never reached its error-write block. """ person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) assignment.detail = { "azure_openai": { "in_flight": True, "resource_group_attempted": "helix-aoai-foo", }, } await db_session.commit() result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "in-flight-needs-manual-recovery"} @pytest.mark.asyncio async def test_resolve_target_subscription_picks_enabled_when_multiple_match( db_session: AsyncSession, ): """If two ``azure_subscription`` scope rows share the same ``external_id`` across different integrations (multi-tenant SP, or a stale row left behind on a disabled integration), the autogrant must NOT raise ``MultipleResultsFound`` — it picks the most-recently synced row whose owning integration is enabled.""" from datetime import UTC, datetime, timedelta from app.domains.apikey.azure_openai_autogrant import ( _resolve_target_subscription, ) # Two integrations: one disabled (stale, older sync), one enabled # (fresh sync). Both have a scope row pointing at the same sub GUID. disabled_integ = Integration( id=uuid.uuid4(), kind="cloud", provider="azure", name="Azure (stale tenant)", slug="azure-stale", enabled=False, is_primary=False, capabilities=[], config_public={"tenant_id": "stale"}, secrets_encrypted="x", secrets_nonce="y", ) enabled_integ = Integration( id=uuid.uuid4(), kind="cloud", provider="azure", name="Azure (PingCAP)", slug="azure-pingcap", enabled=True, is_primary=False, capabilities=[], config_public={"tenant_id": "pingcap"}, secrets_encrypted="x", secrets_nonce="y", ) db_session.add_all([disabled_integ, enabled_integ]) await db_session.commit() older = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(disabled_integ.id), scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION, external_id=_SUB_ID, display_name="stale", environment="dev", last_synced_at=datetime.now(UTC) - timedelta(days=30), ) newer = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(enabled_integ.id), scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION, external_id=_SUB_ID, display_name="fresh", environment="dev", last_synced_at=datetime.now(UTC), ) db_session.add_all([older, newer]) await db_session.commit() resolved = await _resolve_target_subscription(db_session) assert resolved is not None sub_id, scope, integration = resolved assert sub_id == _SUB_ID # Picked the fresh, enabled-integration row. assert str(integration.id) == str(enabled_integ.id) assert scope.id == newer.id @pytest.mark.asyncio async def test_resolve_target_subscription_lowercases_guid( db_session: AsyncSession, ): """Azure GUIDs are case-insensitive; a hand-set platform_setting with uppercase chars must still match the lowercased external_id stored by the cloud-scope sync.""" from app.domains.apikey.azure_openai_autogrant import ( _resolve_target_subscription, ) integ = await _make_azure_integ(db_session, slug="azure-case") scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION, external_id=_SUB_ID.lower(), # synced lowercase per Azure convention display_name="x", environment="dev", ) db_session.add(scope) # Override platform setting to uppercase. db_session.add(PlatformSetting( key="cloud.azure_openai.default_subscription_id", value=_SUB_ID.upper(), )) await db_session.commit() resolved = await _resolve_target_subscription(db_session) assert resolved is not None sub_id, _, _ = resolved assert sub_id == _SUB_ID.lower() # always lowered for Azure consistency @pytest.mark.asyncio async def test_autogrant_rg_failure_persists_diagnostic( db_session: AsyncSession, monkeypatch, ): """When ``ensure_resource_group`` raises (e.g. SP missing the ``Contributor`` role on the sub), the autogrant durably commits ``detail.azure_openai.error`` alongside the in-flight tombstone so admins see WHY it failed on /admin/api-keys, even if the outer Feishu-callback transaction subsequently rolls back.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) from app.integrations.drivers.azure_cloud import AzureCloudDriver async def fake_ensure_rg_403(self, **kwargs): raise RuntimeError("ensure_resource_group 403: AuthorizationFailed") monkeypatch.setattr( AzureCloudDriver, "ensure_resource_group", fake_ensure_rg_403, ) monkeypatch.setattr( "app.domains.apikey.azure_openai_autogrant.decrypt_fields", lambda c, n, *, integration_id=None: { "tenant_id": "x", "client_id": "y", "client_secret": "z", }, ) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "rg-failed"} # Re-fetch from DB (not the in-memory copy) to confirm durability. await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert "ensure_resource_group" in (info.get("error") or "") assert "AuthorizationFailed" in (info.get("error") or "") # ``in_flight`` is CLEARED so the next Feishu callback can retry # once the SP role assignment is fixed — sticky tombstones would # permanently wedge the assignment otherwise. The diagnostic in # ``error`` is what tells admins the prior attempt failed; the # in-flight short-circuit is reserved for the genuinely-stuck case # of a mid-call process crash that didn't write ``error``. assert info.get("in_flight") is False @pytest.mark.asyncio async def test_autogrant_decrypt_failure_returns_decrypt_failed_status( db_session: AsyncSession, monkeypatch, ): """A credential-decryption failure (mismatched AAD, key rotation incomplete) returns ``decrypt-failed`` with a clear diagnostic — NOT a misleading ``rg-failed``.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) def boom(*args, **kwargs): raise ValueError("AEAD authentication failed") monkeypatch.setattr( "app.domains.apikey.azure_openai_autogrant.decrypt_fields", boom, ) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "decrypt-failed"} await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert "credential decryption failed" in (info.get("error") or "") @pytest.mark.asyncio async def test_autogrant_project_failure_persists_diagnostic_with_account_arm_id( db_session: AsyncSession, monkeypatch, ): """When ``provision_foundry_project`` raises (e.g. the parent AIServices account was created with ``allowProjectManagement=False`` or the SP lacks Cognitive Services Contributor on the project sub-resource), the autogrant durably commits ``project-failed`` — capturing ``foundry_account_arm_id`` so the admin can manually finish provisioning the project from Azure portal without re-creating the account.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) # Override the project stub to raise. from app.integrations.drivers.azure_cloud import AzureCloudDriver async def fake_project_403(self, **kwargs): raise RuntimeError( "create_foundry_project 400: ProjectsNotEnabled — " "set allowProjectManagement=true on the parent account" ) monkeypatch.setattr( AzureCloudDriver, "provision_foundry_project", fake_project_403, ) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "project-failed"} await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert "provision_foundry_project" in (info.get("error") or "") # Account ARM id is preserved so the admin can keep the account # and only re-attempt the project step. assert info.get("foundry_account_arm_id", "").endswith( f"/accounts/{captured['account_name']}", ) assert info.get("resource_group_arm_id", "").endswith( f"/resourceGroups/{captured['rg_name']}", ) # ``in_flight`` is CLEARED on the error branch so a transient # ProjectsNotEnabled / 503 doesn't permanently wedge the assignment # — the deterministic naming contract (account + project names # derived from assignment.id + request_number) makes ARM PUT # idempotent, so the next callback safely re-runs with the same # names and picks up where this one left off. The diagnostic stays # in ``error`` so admins know the prior attempt failed. assert info.get("in_flight") is False assert "provision_foundry_project" in (info.get("error") or "") # No allocation rule should be persisted on partial failure. rule_count = (await db_session.execute( select(CmdbCostAllocationRule), )).scalars().all() assert rule_count == [] # No cloud_scope_id binding either. assert assignment.cloud_scope_id is None @pytest.mark.asyncio async def test_autogrant_account_failure_clears_in_flight( db_session: AsyncSession, monkeypatch, ): """`account-failed` must clear `in_flight=False` so a transient ARM 503 in busy regions can be retried. Mirror of the `rg-failed` and `project-failed` regression tests — covers the 2nd of 5 ARM-touching error branches that the round-1 review identified as needing the sticky-tombstone fix.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) from app.integrations.drivers.azure_cloud import AzureCloudDriver async def fake_account_503(self, **kwargs): raise RuntimeError( "create_foundry_account 503: Service Unavailable — " "Cognitive Services regional pressure", ) monkeypatch.setattr( AzureCloudDriver, "provision_foundry_account", fake_account_503, ) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "account-failed"} await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert info.get("in_flight") is False, ( "account-failed must clear in_flight so the next callback can " "retry — the deterministic-naming contract makes ARM PUT " "idempotent on retry." ) assert "provision_foundry_account" in (info.get("error") or "") assert info.get("resource_group_arm_id", "").endswith( f"/resourceGroups/{captured['rg_name']}", ) @pytest.mark.asyncio async def test_autogrant_list_keys_failure_clears_in_flight( db_session: AsyncSession, monkeypatch, ): """`list-keys-failed` must clear `in_flight=False`. listKeys is a read-only POST against an account that's already created, so the next retry is safe.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) from app.integrations.drivers.azure_cloud import AzureCloudDriver async def fake_list_keys_409(self, **kwargs): raise RuntimeError( "list_foundry_account_keys 409: Conflict — " "account still in Creating state", ) monkeypatch.setattr( AzureCloudDriver, "list_foundry_account_keys", fake_list_keys_409, ) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "list-keys-failed"} await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert info.get("in_flight") is False assert "list_keys" in (info.get("error") or "") # Foundry account + project ARM ids are captured so an admin # diagnosing the failure has the full context. assert info.get("foundry_account_arm_id", "").endswith( f"/accounts/{captured['account_name']}", ) assert info.get("foundry_project_arm_id", "").endswith( f"/projects/{captured['project_name']}", ) @pytest.mark.asyncio async def test_autogrant_list_keys_empty_clears_in_flight( db_session: AsyncSession, monkeypatch, ): """`list-keys-empty` (driver returned empty key1 — corrupt response shape) must clear `in_flight=False`. The retry is safe because listKeys is read-only.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured, key1="") result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result == {"status": "list-keys-empty"} await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert info.get("in_flight") is False assert "empty key1" in (info.get("error") or "") @pytest.mark.asyncio async def test_autogrant_no_subscription_persists_diagnostic( db_session: AsyncSession, ): """`no-subscription` must DURABLY commit the diagnostic so admins see the failure reason on /admin/api-keys even if the outer Feishu-callback transaction subsequently rolls back. Symmetric with every other failure branch in the module.""" person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) # Don't seed any cloud integration / subscription scope. result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result["status"] == "no-subscription" # Re-fetch from DB (not the in-memory copy) to confirm durability. await db_session.refresh(assignment) info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is False assert "default_subscription_id" in (info.get("error") or "") @pytest.mark.asyncio async def test_autogrant_handles_concurrent_callback_scope_conflict( db_session: AsyncSession, monkeypatch, ): """Race-defence: a concurrent Feishu callback already created the cmdb_cloud_scopes row (UNIQUE on (integration_id, external_id)). The autogrant must catch the IntegrityError, look up the existing scope, reuse its id on this assignment — NOT wedge with in_flight=True like the round-2 audit identified. Without this guard, the second callback's ``db.flush()`` raises, propagates to the outer log-and-swallow, and leaves the assignment permanently stuck on ``in-flight-needs-manual-recovery``.""" integ = await _make_azure_integ(db_session) sub_scope = await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) assignment = await _make_aoai_assignment(db_session, person.id) captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) notify_spy = AsyncMock() from app.notifications.service import NotificationService monkeypatch.setattr(NotificationService, "dispatch", notify_spy) # Simulate a concurrent callback that already inserted the scope # row before this autogrant reaches its ``db.flush()``. aid = assignment.id rg_name, _, _ = ( await __import__( "app.domains.apikey.azure_openai_autogrant", fromlist=["_build_resource_names"], )._build_resource_names( assignment.project_name or "", assignment.request_number, assignment_id=aid, ) if False else _build_resource_names( assignment.project_name or "", assignment.request_number, assignment_id=aid, ) ) pre_existing_scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type=SCOPE_TYPE_AZURE_RESOURCE_GROUP, external_id=rg_name, display_name="pre-existing from concurrent callback", parent_scope_id=str(sub_scope.id), environment="dev", ) db_session.add(pre_existing_scope) await db_session.commit() result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result["status"] == "granted" await db_session.refresh(assignment) # Assignment was bound to the pre-existing scope, NOT a new one. assert assignment.cloud_scope_id == str(pre_existing_scope.id) # No duplicate scope rows for this RG. duplicates = (await db_session.execute( select(CmdbCloudScope).where( CmdbCloudScope.integration_id == str(integ.id), CmdbCloudScope.external_id == rg_name, ) )).scalars().all() assert len(duplicates) == 1 # Tombstone cleared, granted=True. info = (assignment.detail or {}).get("azure_openai") or {} assert info.get("granted") is True assert info.get("in_flight") is False @pytest.mark.asyncio async def test_autogrant_notification_strips_newlines_in_project_name( db_session: AsyncSession, monkeypatch, ): """``project_name`` is admin-controlled at request time and accepts free text. A value with embedded newlines or carriage returns must NOT survive into the multi-line notification body — that would visually break rendering and (on a markdown-aware downstream renderer) inject formatting / links / control characters. Cheap defence-in-depth — the autogrant strips CR/LF from the fields it splices into the body.""" integ = await _make_azure_integ(db_session) await _make_subscription_scope(db_session, integ) person = await _make_person(db_session) # IdentityPerson display name with an embedded newline + an evil # project_name that tries to inject a fake "— Helix IT Platform" # signature line above the real one. person.display_name = "Shuai\r\nFake-Admin" assignment = await _make_aoai_assignment(db_session, person.id) assignment.project_name = ( "LLM eval\n— Helix IT Platform\nFake injected line" ) await db_session.commit() captured: dict = {} _patch_driver_methods(monkeypatch, captured=captured) notify_spy = AsyncMock() from app.notifications.service import NotificationService monkeypatch.setattr(NotificationService, "dispatch", notify_spy) result = await maybe_auto_grant_azure_openai(db_session, assignment) assert result["status"] == "granted" assert notify_spy.await_count == 1 env = notify_spy.await_args.args[0] body = env.body_text # Critical contract — NO embedded CR/LF survives sanitisation. # If the strip helper regresses, an attacker-controlled # ``project_name`` (admin-controlled at request time, free text) # could split the multi-line body and inject fake signature / # markdown / link content into a Feishu rich-card renderer. assert "Shuai\r\n" not in body assert "Shuai\nFake-Admin" not in body assert "LLM eval\n— Helix IT Platform" not in body, ( "newline-injected pseudo-signature must be sanitised" ) assert "Shuai Fake-Admin" in body # newline → space # The injected pseudo-signature got flattened into the project # line in BOTH the zh-CN and en-US sections of the body — so the # bare phrase "— Helix IT Platform" now appears 3× (twice as # part of the sanitised project line, once as the genuine en-US # signature). What matters for the security posture is that the # genuine signature (which terminates the body) still ends the # body verbatim — i.e. nothing was prepended above the real # closer. Lock that contract: assert body.rstrip().endswith("— Helix IT Platform") # And the genuine zh-CN signature is unchanged (count ==1 because # the injected text uses Latin "Platform", not Chinese "平台"). assert body.count("— Helix IT 平台") == 1 def test_service_callback_wires_to_azure_openai_autogrant() -> None: """Static guard: ``ApiKeyService.handle_feishu_callback`` MUST invoke ``maybe_auto_grant_azure_openai`` in its approved-status branch alongside the existing Bedrock + Bailian autogrants. A rename or accidental removal of the call site would silently regress the Azure OpenAI auto-grant flow back to a no-op without any of the per-module tests catching it (they exercise the autogrant directly, not the wiring). We assert via source-inspection rather than driving the full callback handler because the latter requires seeding an ``ApprovalRequest`` + Feishu instance fixture chain whose surface area is unrelated to the wiring claim being tested. """ import inspect from app.domains.apikey.service import ApiKeyService src = inspect.getsource(ApiKeyService.handle_feishu_callback) # Both the import and the call site must be present in the # callback handler's body. assert "maybe_auto_grant_azure_openai" in src, ( "handle_feishu_callback no longer references " "maybe_auto_grant_azure_openai — the Azure OpenAI auto-grant " "wiring has regressed." ) assert "azure_openai_autogrant" in src, ( "handle_feishu_callback no longer imports the Azure OpenAI " "autogrant module." ) # The REGISTRATION ORDER must put Azure OpenAI after bedrock + bailian # so a future re-order doesn't accidentally route Azure OpenAI through a # different branch — the Feishu callback's only invariant on ordering is # that all three live under the same ``approved`` gate, but the # dispatch-list position is a useful canary. # # 2026-07-30: this used to compare ``src.index(...)`` of the bare # function names, which finds the FIRST textual occurrence — i.e. the # lazy ``from ... import`` block, not the call site. Alphabetising that # import block (ruff I001, now a CI gate) therefore turned this test red # while the actual registration order was untouched. Anchor on the # ``matchers.append`` lines so the assertion measures the thing its own # docstring claims to measure, and stays immune to import formatting. def _register_pos(name: str) -> int: pos = src.find(f"maybe_auto_grant_{name}))") if pos < 0: # multi-line append form: ("", maybe_auto_grant_), pos = src.find(f"maybe_auto_grant_{name}),") assert pos >= 0, f"no matchers.append registration found for {name}" return pos azure_pos = _register_pos("azure_openai") bedrock_pos = _register_pos("bedrock") bailian_pos = _register_pos("bailian") assert azure_pos > bedrock_pos assert azure_pos > bailian_pos