"""Tests for the 068 role-curation + grant-group + scope-request flows. These pin down behavior that's *not* covered by per-driver respx suites: * ``CmdbCloudRole.is_promoted`` is admin-controlled. The roles_sync handler upserts on insert but skips ``is_promoted`` on update — flips persist across syncs. * ``submit_access_request`` with ``extra_role_keys`` writes N+1 grants sharing one ``request_group_id``. * Approving / rejecting any grant in a group acts on the whole group atomically. * Sync of a brand-new scope writes the platform-default ``budget_usd_monthly`` + ``alert_rules``; an admin's later edit to those fields is preserved across subsequent syncs. * ``CloudScopeRequestService.submit`` validates name pattern + budget bounds and rejects providers whose driver doesn't override ``create_scope``. """ from __future__ import annotations import uuid from contextlib import asynccontextmanager from datetime import UTC, date, datetime from decimal import Decimal from unittest.mock import AsyncMock, patch import pytest from sqlalchemy import func as sa_func from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException, ConflictError from app.domains.cmdb.lifecycle_service import ( CloudLifecycleService, _build_default_alert_rules, evaluate_scope_alert_rules, sync_roles_for_integration, sync_roles_for_scope, sync_scopes_for_integration, ) from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING, GRANT_STATUS_REVOKED, SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudRole, CmdbCloudScope, ) from app.domains.cmdb.scope_request_service import ( _validate_budget, _validate_name, ) from app.identity.person_models import IdentityPerson from app.integrations.driver_base import ( CloudRoleInfo, CloudScopeInfo, CloudScopeRef, ) from app.integrations.models import Integration from app.notifications.service import resolve_card_text as _resolve_card_text def _t(value, lang: str = "zh") -> str: """What the READER actually sees. Card chrome is a ``LocalizedText`` resolved per recipient, so asserting on the raw field would only prove the dataclass exists. Resolving first also makes the assertion stronger: it pins the copy in one language rather than whichever one the producer happened to write. """ return str(_resolve_card_text(value, lang)) # ── Service-boundary patching ──────────────────────────────────── def _patch_session_factory(db_session: AsyncSession): """Pipe every ``async with async_session_factory()`` inside the service through this test's session — same trick as ``test_grants_sync.py``. The service binds the name at import, so the patch target is the service's namespace. """ @asynccontextmanager async def _factory(): yield db_session return patch( "app.domains.cmdb.lifecycle_service.async_session_factory", new=_factory, ) def _patch_driver(driver): """Replace ``_load_cloud_driver`` with a coroutine that returns a pre-built fake. Tests don't go through the live integration row, so the credential / driver-class lookup would otherwise fail.""" async def _loader(_session, _integration): return driver return patch( "app.domains.cmdb.lifecycle_service._load_cloud_driver", side_effect=_loader, ) # ── Shared fixtures ─────────────────────────────────────────────── async def _make_person( db: AsyncSession, *, email: str, name: str = "T", ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=email, display_name=name, employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.commit() return p async def _grant_role( db: AsyncSession, person: IdentityPerson, role_name: str, ) -> None: """Give ``person`` a stored ``auth_user_roles`` row. Needed by the alert-card tests: a NET-basis card renders its absolute figures only for the invoice-measure circle (finance / cloud_admin / global_admin) — everyone else gets the amount-free variant (2026-07-29 confidentiality sweep). """ from sqlalchemy import text from app.auth.models import AuthUserRole rid = (await db.execute( text("SELECT id FROM auth_roles WHERE name = :n"), {"n": role_name}, )).scalar_one() db.add(AuthUserRole( user_id=person.id, role_id=uuid.UUID(rid) if isinstance(rid, str) else rid, )) await db.commit() async def _make_integration( db: AsyncSession, *, provider: str = "azure", ) -> Integration: row = Integration( id=uuid.uuid4(), kind="cloud", provider=provider, name=f"{provider}-curation-test", slug=f"{provider}-curation-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], config_public=( {"organization_id": "478164292948"} if provider == "gcp" else None ), ) db.add(row) await db.commit() return row async def _make_scope( db: AsyncSession, *, integration: Integration, owner: IdentityPerson | None = None, ) -> CmdbCloudScope: scope = CmdbCloudScope( integration_id=str(integration.id), scope_type=( "gcp_project" if integration.provider == "gcp" else "aws_account" ), external_id=f"acct-{uuid.uuid4().hex[:6]}", display_name="curation-test-scope", owner_person_id=str(owner.id) if owner else None, status=SCOPE_STATUS_ACTIVE, budget_usd_monthly=Decimal("500.00"), raw_data=( { "organization_id": "478164292948", "tenant_boundary_verified": True, } if integration.provider == "gcp" else {} ), ) db.add(scope) await db.commit() return scope # ── Tests ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_default_alert_rules_translates_pct_to_usd(): """``_build_default_alert_rules`` materialises the template's ``threshold_pct`` into absolute ``threshold_usd`` keyed off the scope budget at sync time.""" # Mock session — function only reads platform settings, so we hand # it a stub that always returns the default template. rules = await _build_default_alert_rules( session=_FakeSession(), monthly_budget_usd=2000.0, ) by_type = {(r["type"], r["threshold_pct"]): r for r in rules} # Daily 25% of 2000 = 500 assert by_type[("daily", 25)]["threshold_usd"] == 500.00 # Monthly 50/80/100% of 2000 = 1000/1600/2000 assert by_type[("monthly", 50)]["threshold_usd"] == 1000.00 assert by_type[("monthly", 80)]["threshold_usd"] == 1600.00 assert by_type[("monthly", 100)]["threshold_usd"] == 2000.00 @pytest.mark.asyncio async def test_alert_rules_only_fire_highest_threshold_per_type( db_session: AsyncSession, ): """When MTD cost crosses 50%/80%/100% in the same run, only the highest-threshold rule actually pages — the lower siblings are suppressed and marked consumed for this period. Reported: an Azure ``tidb-o11y`` subscription whose MTD cost was $3,212.71 against a $1,000 budget triggered three Feishu cards (one per crossed threshold) on the same day. Recipient inbox got bombed with redundant alerts. (2026-05-07.) """ integ = await _make_integration(db_session, provider="azure") owner = await _make_person(db_session, email="owner-tier@p.io") # cloud_admin so the net-basis cards render their full stat tiles — the # ``阈值`` assertions below are how the winner is identified. await _grant_role(db_session, owner, "cloud_admin") scope = await _make_scope(db_session, integration=integ, owner=owner) # Three monthly tiers + one daily tier — daily must NOT be # suppressed by the monthly winner (different rule_type). scope.alert_rules = [ {"type": "monthly", "threshold_usd": 500.0, "active": True}, {"type": "monthly", "threshold_usd": 800.0, "active": True}, {"type": "monthly", "threshold_usd": 1000.0, "active": True}, {"type": "daily", "threshold_usd": 200.0, "active": True}, ] await db_session.commit() sent: list[dict] = [] async def _capture(env, **_kw): # The 2026-05-16 card refresh dropped ``template_key`` in # favour of an inline ``card_stats``/``card_subtitle`` shape # (mirrors the daily-spike card) so the rendered message # carries the explicit period_key in the title — backfill # alerts can no longer masquerade as current-month alerts. # Tests now assert on title + card_subtitle + stat labels. sent.append({ "title": env.title, "card_subtitle": env.card_subtitle, "stat_values": {_t(s.label): _t(s.value) for s in env.card_stats}, }) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired = await evaluate_scope_alert_rules( db_session, scope, today_cost=300.0, today_date=date(2026, 5, 7), month_to_date_cost=3212.71, measure="net", integration=integ, ) # 1 monthly winner ($1000) + 1 daily winner ($200) = 2 rules fired. assert fired == 2 # Recipient set is owner-only (1 person), so dispatch is called # twice — once per fired rule. assert len(sent) == 2 # Each card carries the explicit period_key (YYYY-MM or # YYYY-MM-DD) in BOTH the title and the card_subtitle so the # recipient can tell which period the alert is for without # cross-referencing. monthly_card = next(c for c in sent if "月度" in c["title"]) daily_card = next(c for c in sent if "月度" not in c["title"]) assert "2026-05" in monthly_card["title"] assert "curation-test-scope" in monthly_card["title"] assert "2026-05" in _t(monthly_card["card_subtitle"]) assert "2026-05-07" in daily_card["title"] assert "2026-05-07" in _t(daily_card["card_subtitle"]) # Winner thresholds verified via the rendered ``阈值`` stat tile. assert monthly_card["stat_values"]["阈值"] == "$1,000.00 USD" assert daily_card["stat_values"]["阈值"] == "$200.00 USD" # Lower-threshold monthly siblings ($500, $800) MUST be marked # as consumed for this period — otherwise the next cron tick # would re-evaluate them and re-page after a small cost dip. # The legacy ``last_fired`` JSON map is still written for # back-compat (admin UIs read it); the authoritative dedup is # now ``cmdb_cloud_scope_alert_fired`` rows (asserted below). fired_key = "monthly:2026-05" by_thr = {float(r["threshold_usd"]): r for r in scope.alert_rules} assert by_thr[500.0]["last_fired"][fired_key] is True assert by_thr[800.0]["last_fired"][fired_key] is True assert by_thr[1000.0]["last_fired"][fired_key] is True # Authoritative claim-first dedup rows: each threshold tier (incl. # suppressed ones) MUST have its slot claimed for the period so # the next cron tick doesn't re-evaluate. from app.domains.cmdb.models import CmdbCloudScopeAlertFired fired_rows = (await db_session.execute( select(CmdbCloudScopeAlertFired).where( CmdbCloudScopeAlertFired.scope_id == str(scope.id), ), )).scalars().all() claimed = { (r.rule_type, r.period_key, float(r.threshold_usd)) for r in fired_rows } assert ("monthly", "2026-05", 500.0) in claimed assert ("monthly", "2026-05", 800.0) in claimed assert ("monthly", "2026-05", 1000.0) in claimed assert ("daily", "2026-05-07", 200.0) in claimed # Idempotency: a second run on the same data dispatches nothing. # Claim-first INSERT fails with IntegrityError on the existing # rows, skipping dispatch. sent.clear() with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired2 = await evaluate_scope_alert_rules( db_session, scope, today_cost=300.0, today_date=date(2026, 5, 7), month_to_date_cost=3212.71, measure="net", integration=integ, ) assert fired2 == 0 assert sent == [] @pytest.mark.asyncio async def test_alert_rules_dispatch_false_claims_slot_but_no_feishu( db_session: AsyncSession, ): """Backfill mode (``dispatch=False``) MUST still claim the dedup slot for every threshold that would fire, so the next *live* cron tick doesn't re-page on historical-month data — but it MUST NOT actually send any Feishu/email. Models the 2026-05-16 prod incident root cause: the backfill walker ran ``sync_billing_daily_for_integration`` per 30-day chunk across 6 months. Each chunk's end_date crossed a different monthly threshold key, so the dedup map was always empty and each chunk fired a fresh card — 7+ cards per scope across 60 scopes. With ``dispatch=False`` the rows still land, future live ticks see them, and zero cards go out. """ integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-bf@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.alert_rules = [ {"type": "monthly", "threshold_usd": 1000.0, "active": True}, ] await db_session.commit() sent: list[dict] = [] async def _capture(env, **_kw): sent.append({"title": env.title}) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired = await evaluate_scope_alert_rules( db_session, scope, today_cost=100.0, today_date=date(2025, 11, 30), # historical month month_to_date_cost=5505.80, # crosses $1000 threshold measure="net", integration=integ, dispatch=False, ) # No cards dispatched — the whole point of dispatch=False. assert sent == [] # Function returns 0 since "fired_count" counts dispatches. assert fired == 0 # Slot is claimed so a subsequent live run does NOT re-page. from app.domains.cmdb.models import CmdbCloudScopeAlertFired fired_rows = (await db_session.execute( select(CmdbCloudScopeAlertFired).where( CmdbCloudScopeAlertFired.scope_id == str(scope.id), CmdbCloudScopeAlertFired.period_key == "2025-11", ), )).scalars().all() assert len(fired_rows) == 1 assert float(fired_rows[0].threshold_usd) == 1000.0 # Verify the second (live) run respects the claim — no dispatch # even though dispatch=True. with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired2 = await evaluate_scope_alert_rules( db_session, scope, today_cost=100.0, today_date=date(2025, 11, 30), month_to_date_cost=5505.80, measure="net", integration=integ, dispatch=True, ) assert fired2 == 0 assert sent == [] @pytest.mark.asyncio async def test_alert_rules_honour_legacy_last_fired_json_on_deploy_day( db_session: AsyncSession, ): """Deploy-day back-compat: pre-migration-171 environments tracked dedup in ``rule.last_fired[fired_key]`` JSON. The 2026-05-16 prod hotfix data-stamped this map for every (scope, month) that would have fired so the first post-deploy cron tick wouldn't re-page on a fresh (empty-on-deploy) ``cmdb_cloud_scope_alert_fired`` table. evaluate_scope_alert_rules MUST consult the legacy JSON BEFORE attempting a claim — otherwise the claim succeeds (table empty) and the card re-fires. Also: we deliberately do NOT paper over the legacy state by inserting a fresh ``fired_at=now()`` row that misrepresents when the alert actually went out — the new table simply stays empty until the next genuine threshold cross. """ from app.domains.cmdb.models import CmdbCloudScopeAlertFired integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-legacy@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) # Simulates the post-hotfix prod state: rule already marked as # fired in JSON, but the new dedup table is empty. scope.alert_rules = [ { "type": "monthly", "threshold_usd": 1000.0, "active": True, "last_fired": {"monthly:2026-05": True}, }, ] await db_session.commit() sent: list[dict] = [] async def _capture(env, **_kw): sent.append({"title": env.title}) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired = await evaluate_scope_alert_rules( db_session, scope, today_cost=200.0, today_date=date(2026, 5, 17), month_to_date_cost=5505.80, # > $1000 threshold measure="net", integration=integ, ) # No card dispatched (legacy JSON gate stops the path). assert fired == 0 assert sent == [] # No claim row inserted either — we don't fabricate a fired_at. rows = (await db_session.execute( select(CmdbCloudScopeAlertFired).where( CmdbCloudScopeAlertFired.scope_id == str(scope.id), ), )).scalars().all() assert rows == [] @pytest.mark.asyncio async def test_alert_rules_threshold_change_mid_period_does_not_refire( db_session: AsyncSession, ): """Pre-launch audit P0: admin tuning a threshold mid-period MUST NOT re-page the same scope on the same data. Scenario: monthly rule fires at $1000 on day 1. Day 10 admin lowers threshold to $100 (e.g. realised the original budget was too lax for a sandbox account). MTD is now $5505 — well above both thresholds. Pre-fix, the new $100 threshold opened a NEW dedup slot (UNIQUE includes ``threshold_usd``) and the same scope re-paged on the same period with no new signal — one scope, one period, two cards. The pre-claim ``_period_already_alerted`` gate now blocks the re-fire regardless of which threshold tripped first within the period. """ from app.domains.cmdb.models import CmdbCloudScopeAlertFired integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-thrchg@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.alert_rules = [ {"type": "monthly", "threshold_usd": 1000.0, "active": True}, ] await db_session.commit() sent: list[dict] = [] async def _capture(env, **_kw): sent.append({"title": env.title}) # First run — threshold $1000, MTD $5505 → fires. with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired1 = await evaluate_scope_alert_rules( db_session, scope, today_cost=200.0, today_date=date(2026, 5, 7), month_to_date_cost=5505.80, measure="net", integration=integ, ) assert fired1 == 1 assert len(sent) == 1 # Admin lowers threshold to $100 mid-period. scope.alert_rules = [ {"type": "monthly", "threshold_usd": 100.0, "active": True}, ] await db_session.commit() sent.clear() # Second run — new threshold $100, same MTD — MUST stay silent. with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired2 = await evaluate_scope_alert_rules( db_session, scope, today_cost=200.0, today_date=date(2026, 5, 10), month_to_date_cost=5505.80, measure="net", integration=integ, ) assert fired2 == 0 assert sent == [] # Slot table still carries the ORIGINAL $1000 row (we don't # overwrite history just because admin tuned the threshold). fired_rows = (await db_session.execute( select(CmdbCloudScopeAlertFired).where( CmdbCloudScopeAlertFired.scope_id == str(scope.id), CmdbCloudScopeAlertFired.period_key == "2026-05", ), )).scalars().all() assert len(fired_rows) == 1 assert float(fired_rows[0].threshold_usd) == 1000.0 @pytest.mark.asyncio async def test_grant_request_extra_roles_share_request_group_id( db_session: AsyncSession, ): """``submit_access_request`` with ``extra_role_keys`` writes N+1 grants sharing one ``request_group_id``.""" integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o@p.io") requester = await _make_person(db_session, email="r@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) service = CloudLifecycleService(db_session) primary = await service.submit_access_request( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", reason="t", ttl_days=30, requester_id=requester.id, trace_id="t-1", extra_role_keys=["Storage Blob Data Reader", "Reader"], # dup pruned ) await db_session.commit() rows = list((await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), ) )).scalars().all()) assert len(rows) == 2 group_ids = {r.request_group_id for r in rows} assert len(group_ids) == 1 and None not in group_ids assert primary.role_key == "Reader" assert primary.request_group_id is not None @pytest.mark.asyncio async def test_reject_grant_revokes_whole_request_group( db_session: AsyncSession, ): """Rejecting any one grant in a group revokes every PENDING sibling at once.""" integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o2@p.io") requester = await _make_person(db_session, email="r2@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) service = CloudLifecycleService(db_session) await service.submit_access_request( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", reason=None, ttl_days=30, requester_id=requester.id, trace_id="t-2", extra_role_keys=["Contributor"], ) await db_session.commit() pending = list((await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.status == GRANT_STATUS_PENDING, ) )).scalars().all()) assert len(pending) == 2 # Reject one; the other should also flip to revoked. await service.reject_grant( str(pending[0].id), rejector_id=owner.id, reason="not needed", trace_id="t-3", ) await db_session.commit() rows = list((await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), ) )).scalars().all()) assert all(r.status == GRANT_STATUS_REVOKED for r in rows) @pytest.mark.parametrize( "prior_status,prior_revoke_reason,prior_revoked_at,prior_source", [ ( "revoked", "drift_removed", datetime(2026, 5, 5, tzinfo=UTC), "discovered", ), ("expired", "expired", datetime(2026, 5, 5, tzinfo=UTC), "helix"), ("disabled", None, None, "helix"), ], ids=["revoked", "expired", "disabled"], ) @pytest.mark.asyncio async def test_re_request_resurrects_existing_row_for_every_terminal_status( db_session: AsyncSession, prior_status: str, prior_revoke_reason: str | None, prior_revoked_at: datetime | None, prior_source: str, ): """Re-requesting a role whose prior grant was revoked / expired / disabled MUST reuse the existing row, not insert a new one. Regression: 2026-05-09 a real user (Wang Letian) re-requested ``DBaaS-Prod-DevUser-Role-o11y`` on tidb-o11y after the prior grant was drift-removed (status='revoked'). The submit handler used to filter ``existing_rows`` by ``status IN (active, pending)`` only and then INSERT a fresh row, which tripped the ``uq_cmdb_cloud_grants_scope_person_role`` unique key and 500'd on commit. Fix flips the existing inactive row back to pending in-place; this parametrized test pins behaviour for **every** terminal status (revoked / expired / disabled), and additionally asserts the source-mis-tag fix (a ``source='discovered'`` row that gets resurrected becomes ``source='helix'`` so the next drift sweep doesn't re-revoke it) and the detail-merge fix (a prior ``disable_result`` stash is preserved while the new request fields are layered on top). """ integ = await _make_integration(db_session) owner = await _make_person(db_session, email=f"o-rr-{prior_status}@p.io") requester = await _make_person( db_session, email=f"r-rr-{prior_status}@p.io", ) scope = await _make_scope(db_session, integration=integ, owner=owner) # Seed a pre-existing terminal-state grant. Mirror what the disable # path stashes (``disable_result`` block) so we can verify the # detail-merge fix preserves it through resurrection. revoked = CmdbCloudAccessGrant( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", status=prior_status, granted_at=datetime(2026, 4, 1, tzinfo=UTC), expires_at=datetime(2026, 7, 1, tzinfo=UTC), revoked_at=prior_revoked_at, revoke_reason=prior_revoke_reason, last_notified_stage="t14", last_login_at=datetime(2026, 4, 28, tzinfo=UTC), source=prior_source, # Mark as having been touched by a prior grants_sync run so we # can verify the resurrection clears the bookkeeping that would # otherwise let the next drift sweep re-revoke a discovered row. last_seen_grants_sync_job_id=str(uuid.uuid4()), last_seen_grants_sync_at=datetime(2026, 4, 28, tzinfo=UTC), external_ref="stale-cloud-assignment-id", external_principal_email="r@stale.local", granted_by_person_id=str(owner.id), employee_request_id="prior-emp-req-1", detail={ "disable_result": { "ok": True, "already_absent": False, "reason": "inactive", "at": "2026-05-05T00:00:00+00:00", }, "last_error": "transient driver flap", }, ) db_session.add(revoked) await db_session.commit() revoked_id = revoked.id service = CloudLifecycleService(db_session) primary = await service.submit_access_request( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", reason="re-request after drift", ttl_days=30, requester_id=requester.id, trace_id="t-resurrect", ) await db_session.commit() # Same primary key — row was reused, not inserted. assert primary.id == revoked_id rows = list((await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(requester.id), ) )).scalars().all()) assert len(rows) == 1 only = rows[0] assert only.status == GRANT_STATUS_PENDING # Revoke / expiry / inactivity bookkeeping cleared. assert only.revoked_at is None assert only.revoked_by_person_id is None assert only.revoke_reason is None assert only.expires_at is None assert only.last_renewed_at is None assert only.last_notified_stage is None assert only.last_login_at is None assert only.employee_request_id is None # Source normalization — every resurrected row is Helix-issued # going forward; sync-bookkeeping cleared so the drift sweep can't # re-revoke a discovered row that's now under Helix governance. assert only.source == "helix" assert only.last_seen_grants_sync_job_id is None assert only.last_seen_grants_sync_at is None # Stale cloud-side assignment metadata cleared so dispatch # re-creates the binding cleanly. assert only.external_ref is None assert only.external_principal_email is None # Self-requested resurrection has no granted_by attribution # (approve flow stamps that on the active flip). assert only.granted_by_person_id is None # New submission's request_group_id stamped. assert only.request_group_id is not None # Detail merged, not overwritten — disable_result + last_error # forensic stash preserved while the new request fields layer on. assert only.detail is not None assert only.detail.get("reason") == "re-request after drift" assert only.detail.get("requested_ttl_days") == 30 assert only.detail.get("primary_role_key") == "Reader" assert "disable_result" in only.detail assert only.detail["disable_result"]["ok"] is True assert only.detail.get("last_error") == "transient driver flap" # Resurrection markers stamped for forensic auditability. assert only.detail.get("_resurrected_from_status") == prior_status if prior_revoke_reason is not None: assert ( only.detail.get("_resurrected_from_revoke_reason") == prior_revoke_reason ) if prior_revoked_at is not None: assert only.detail.get("_resurrected_from_revoked_at") is not None @pytest.mark.asyncio async def test_re_request_mixed_primary_new_extra_resurrected( db_session: AsyncSession, ): """Multi-role group where the primary role is fresh and an extra role was previously revoked: the primary INSERTs a new row, the extra resurrects, both share one ``request_group_id``, the audit log carries one ``resurrected_grants`` entry naming the extra.""" integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o-mix@p.io") requester = await _make_person(db_session, email="r-mix@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) revoked_extra = CmdbCloudAccessGrant( scope_id=str(scope.id), person_id=str(requester.id), role_key="Storage Blob Data Reader", role_display_name="Storage Blob Data Reader", status=GRANT_STATUS_REVOKED, granted_at=datetime(2026, 4, 1, tzinfo=UTC), revoked_at=datetime(2026, 5, 1, tzinfo=UTC), revoke_reason="drift_removed", ) db_session.add(revoked_extra) await db_session.commit() extra_id = revoked_extra.id service = CloudLifecycleService(db_session) primary = await service.submit_access_request( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", # fresh role_display_name="Reader", reason="mixed", ttl_days=30, requester_id=requester.id, trace_id="t-mixed", extra_role_keys=["Storage Blob Data Reader"], # resurrected ) await db_session.commit() rows = list((await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(requester.id), ) )).scalars().all()) assert len(rows) == 2 by_role = {r.role_key: r for r in rows} assert by_role["Reader"].id == primary.id assert by_role["Storage Blob Data Reader"].id == extra_id # Both rows share the new request_group_id. rg_ids = {r.request_group_id for r in rows} assert len(rg_ids) == 1 and None not in rg_ids # Both pending. assert all(r.status == GRANT_STATUS_PENDING for r in rows) # Resurrected extra carries the marker; primary doesn't. assert ( by_role["Storage Blob Data Reader"].detail.get( "_resurrected_from_status", ) == GRANT_STATUS_REVOKED ) assert ( "_resurrected_from_status" not in (by_role["Reader"].detail or {}) ) @pytest.mark.asyncio async def test_re_request_blocks_when_existing_grant_is_active( db_session: AsyncSession, ): """Re-requesting on top of an *active* grant still raises ConflictError — resurrection is only for terminal-state rows.""" integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o-block@p.io") requester = await _make_person(db_session, email="r-block@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) db_session.add(CmdbCloudAccessGrant( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", status=GRANT_STATUS_ACTIVE, granted_at=datetime(2026, 4, 1, tzinfo=UTC), expires_at=datetime(2026, 7, 1, tzinfo=UTC), )) await db_session.commit() service = CloudLifecycleService(db_session) with pytest.raises(ConflictError): await service.submit_access_request( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", reason="dup", ttl_days=30, requester_id=requester.id, trace_id="t-dup", ) @pytest.mark.asyncio async def test_submit_blocks_prod_scope_for_every_caller( db_session: AsyncSession, ): """Production-tagged scopes are not self-serviceable. Backend refuses on submit even if the picker were bypassed (cached UI, direct curl), and the block is role-agnostic — IT admins who need to provision prod on behalf of the security team use ``/grants/owner-add`` (scope-owner gated), not the self-request path. This keeps "self-service never reaches prod" a flat invariant. """ integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o-prod@p.io") requester = await _make_person(db_session, email="r-prod@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.environment = "prod" await db_session.commit() service = CloudLifecycleService(db_session) for actor_role in (["employee"], ["cloud_admin"], ["global_admin"]): with pytest.raises(AppException) as ei: await service.submit_access_request( scope_id=str(scope.id), person_id=str(requester.id), role_key="Reader", role_display_name="Reader", reason="prod debug", ttl_days=30, requester_id=requester.id, trace_id=f"t-prod-{actor_role[0]}", actor_role_names=actor_role, ) assert ei.value.code == "CLOUD_SCOPE_PROD_BLOCKED" assert ei.value.status_code == 403 @pytest.mark.asyncio async def test_owner_add_blocks_prod_scope_symmetric_with_submit( db_session: AsyncSession, ): """``owner_add_member`` mirrors the prod-block in ``submit_access_request`` — production cloud access is granted by the R&D security team (Kidd.li), not by any Helix self-service flow including owner-add. Same flat invariant for both entry points; if this drifts, an admin owner-adding on a prod scope becomes a quiet bypass.""" integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o-oadd-prod@p.io") target = await _make_person(db_session, email="t-oadd-prod@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.environment = "prod" await db_session.commit() service = CloudLifecycleService(db_session) with pytest.raises(AppException) as ei: await service.owner_add_member( scope_id=str(scope.id), person_id=str(target.id), role_key="Reader", role_display_name="Reader", ttl_days=30, actor_user_id=owner.id, trace_id="t-oadd-prod", ) assert ei.value.code == "CLOUD_SCOPE_PROD_BLOCKED" assert ei.value.status_code == 403 @pytest.mark.asyncio async def test_owner_add_resurrects_prior_revoked_grant( db_session: AsyncSession, ): """``owner_add_member`` must resurrect a prior revoked / expired / disabled row instead of inserting a fresh one — the unique key ``uq_cmdb_cloud_grants_scope_person_role`` doesn't partition by status, so an INSERT would 500. Same bug class as ``submit_access_request`` pre-fix; the shared ``_reuse_or_create_grant_row`` helper covers both.""" integ = await _make_integration(db_session) owner = await _make_person(db_session, email="o-oadd-rr@p.io") target = await _make_person(db_session, email="t-oadd-rr@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) revoked = CmdbCloudAccessGrant( scope_id=str(scope.id), person_id=str(target.id), role_key="Reader", role_display_name="Reader", status=GRANT_STATUS_REVOKED, granted_at=datetime(2026, 4, 1, tzinfo=UTC), revoked_at=datetime(2026, 5, 1, tzinfo=UTC), revoke_reason="owner", source="helix", ) db_session.add(revoked) await db_session.commit() revoked_id = revoked.id # Patch the dispatch path so we don't need a real driver wired up # for this test — we only care that the row state is correct on # entry to dispatch (i.e. the ``_reuse_or_create_grant_row`` step # succeeded without 500'ing). with patch( "app.domains.cmdb.lifecycle_service." "CloudLifecycleService._dispatch_grant", new=AsyncMock(side_effect=lambda **kw: kw["grant"]), ): service = CloudLifecycleService(db_session) result = await service.owner_add_member( scope_id=str(scope.id), person_id=str(target.id), role_key="Reader", role_display_name="Reader", ttl_days=30, actor_user_id=owner.id, trace_id="t-oadd-rr", ) await db_session.commit() rows = list((await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(target.id), ) )).scalars().all()) assert len(rows) == 1 only = rows[0] # Same primary key — row was reused. assert only.id == revoked_id # Status is pending on entry to dispatch (dispatch flips to active # on driver success — patched away here). assert only.status == GRANT_STATUS_PENDING # Granted_by attribution stamped on resurrection. assert only.granted_by_person_id == str(owner.id) # Direct-add detail merged in (no overwrite). assert only.detail and only.detail.get("direct_add") is True assert ( only.detail.get("_resurrected_from_status") == GRANT_STATUS_REVOKED ) assert result is only @pytest.mark.asyncio async def test_cloud_admin_can_repair_owner_self_grant( db_session: AsyncSession, ): """A cloud admin can restore an owner accidentally left without IAM. A plain owner remains blocked from selecting a named role for themselves, but an admin needs this break-glass path to repair the exact state where ``owner_person_id`` is set while the members table is empty. """ integ = await _make_integration(db_session, provider="gcp") owner = await _make_person(db_session, email="owner-repair@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) await db_session.commit() with patch( "app.domains.cmdb.lifecycle_service." "CloudLifecycleService._dispatch_grant", new=AsyncMock(side_effect=lambda **kw: kw["grant"]), ) as dispatch: result = await CloudLifecycleService(db_session).owner_add_member( scope_id=str(scope.id), person_id=str(owner.id), role_key="roles/owner", role_display_name="Owner", ttl_days=90, actor_user_id=owner.id, trace_id="owner-self-repair", actor_role_names=["global_admin"], ) assert result.person_id == str(owner.id) assert dispatch.await_args.kwargs["force_perpetual"] is True @pytest.mark.asyncio async def test_roles_sync_preserves_admin_promote_flip( db_session: AsyncSession, ): """An admin's promote/demote flip survives the next roles_sync. Driver classification governs ``is_promoted`` only on insert; on update the column is intentionally excluded so the catalog UI is the source of truth.""" integ = await _make_integration(db_session, provider="azure") await _make_scope(db_session, integration=integ) integ_id = str(integ.id) role_key = "Reader" fake = _FakeDriver(role_key) # Pipe the service's ``async_session_factory()`` through this # test's session and the ``_load_cloud_driver`` call through a # fake — same pattern as ``test_grants_sync.py``. Without these # the service opens its own MySQL connection on a different event # loop, which on the CI runner trips "Future attached to a # different loop" intermittently and ALSO makes upserts invisible # to the test (they land in the service's transaction, not ours). with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_integration(integ_id) # Driver said is_promoted=False — confirm it landed. row = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.role_key == role_key, ) ) assert row is not None and row.is_promoted is False # Admin flips it on. row.is_promoted = True await db_session.commit() # Re-sync — the ON DUPLICATE KEY UPDATE branch deliberately omits # is_promoted, so the admin flip persists. with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_integration(integ_id) row2 = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.role_key == role_key, ) ) assert row2 is not None and row2.is_promoted is True, ( "admin promote was clobbered by roles_sync — " "on_duplicate_key_update must NOT include is_promoted" ) @pytest.mark.asyncio async def test_roles_sync_preserves_admin_sort_order_and_description_override( db_session: AsyncSession, ): """Admin-curated ``sort_order`` + ``description_override`` survive the next roles_sync. Both are intentionally excluded from the ON DUPLICATE KEY UPDATE clause (same persist rule as ``is_promoted``) so the catalog UI stays the source of truth — the daily sync keeps refreshing the provider ``description`` without clobbering the admin's display rank or custom blurb. """ integ = await _make_integration(db_session, provider="azure") await _make_scope(db_session, integration=integ) integ_id = str(integ.id) role_key = "Reader" fake = _FakeDriver(role_key) with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_integration(integ_id) row = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.role_key == role_key, ) ) assert row is not None # Fresh insert → column defaults. assert row.sort_order == 0 assert row.description_override is None # Admin curates the display order + a custom employee-facing blurb. row.sort_order = 5 row.description_override = "Read-only access for auditors" await db_session.commit() # Re-sync — the driver re-emits the same role; the override columns # must NOT be touched. with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_integration(integ_id) row2 = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.role_key == role_key, ) ) assert row2 is not None assert row2.sort_order == 5, ( "admin sort_order was clobbered by roles_sync — " "on_duplicate_key_update must NOT include sort_order" ) assert row2.description_override == "Read-only access for auditors", ( "admin description_override was clobbered by roles_sync — " "on_duplicate_key_update must NOT include description_override" ) @pytest.mark.asyncio async def test_roles_sync_does_not_soft_delete_admin_seeded_null_job_role( db_session: AsyncSession, ): """Counter-test for the 2026-05-28 sweep-predicate change in ``sync_roles_for_integration``: a ``CmdbCloudRole`` row with NULL ``last_seen_sync_job_id`` (admin-seeded directly, future migration import) is structurally outside cloud-sync's authority and MUST NOT be auto-soft-deleted on the next sync when the driver doesn't return it. Otherwise the legacy-CMDB → resource-sync collision incident would replicate to the role catalog (curated roles silently disappear from the request modal across the org). """ integ = await _make_integration(db_session, provider="azure") scope = await _make_scope(db_session, integration=integ) integ_id = str(integ.id) # Pre-seed an admin-curated role row directly — no sync has ever # touched it (``last_seen_sync_job_id IS NULL``). admin_role = CmdbCloudRole( id=str(uuid.uuid4()), integration_id=integ_id, scope_id=str(scope.id), role_key="AdminCurated-NotFromDriver", display_name="Admin-curated role", is_custom=True, source="manual", is_promoted=True, raw_data={}, ) db_session.add(admin_role) await db_session.commit() # Driver returns a DIFFERENT role — admin row is "missing" from # the live driver's perspective. fake = _FakeDriver("OtherRole") with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_integration(integ_id) await db_session.refresh(admin_role) assert admin_role.deleted_at is None, ( "admin-seeded role (NULL last_seen_sync_job_id) was " "auto-soft-deleted — the sweep predicate regression " "from pre-2026-05-28 is back." ) @pytest.mark.asyncio async def test_scope_sync_seeds_default_budget_and_alert_rules( db_session: AsyncSession, ): """A brand-new scope picked up by sync_scopes_for_integration lands with the platform-default monthly budget + alert rules.""" integ = await _make_integration(db_session, provider="aws") fake_external_id = f"acct-{uuid.uuid4().hex[:8]}" class _Driver: async def list_scopes(self): return [CloudScopeInfo( scope_type="aws_account", external_id=fake_external_id, display_name="brand-new", status="active", raw={}, )] async def close(self): return None with _patch_session_factory(db_session), _patch_driver(_Driver()): await sync_scopes_for_integration(str(integ.id)) scope = await db_session.scalar( select(CmdbCloudScope).where( CmdbCloudScope.integration_id == str(integ.id), CmdbCloudScope.external_id == fake_external_id, ) ) assert scope is not None assert float(scope.budget_usd_monthly or 0) == 1000.0 assert isinstance(scope.alert_rules, list) and len(scope.alert_rules) >= 4 types = {r["type"] for r in scope.alert_rules} assert types == {"daily", "monthly"} @pytest.mark.asyncio async def test_sync_roles_for_scope_populates_new_scope_without_touching_siblings( db_session: AsyncSession, ): """Post-provision sync must: 1. Insert the driver-returned roles for THE target scope. 2. Leave any role row attached to a sibling scope on the same integration alone — proves we did NOT run mark-and-sweep, which would close those siblings (they wouldn't carry this run's sync_job_id). This is the failure mode that motivated the helper: PR #270's ``na-gtm-eng`` provisioning landed the scope row but left ``cmdb_cloud_roles`` empty for it, so the requester saw "暂无数据" in the role dropdown until the next cron run. """ integ = await _make_integration(db_session, provider="gcp") sibling = await _make_scope(db_session, integration=integ) target = await _make_scope(db_session, integration=integ) # Pre-seed a role row on the sibling to prove sync_roles_for_scope # doesn't sweep it as stale. pre_existing = CmdbCloudRole( id=uuid.uuid4(), integration_id=str(integ.id), scope_id=str(sibling.id), role_key="roles/owner", display_name="Owner", is_custom=False, source="gcp_predefined", is_promoted=True, ) db_session.add(pre_existing) await db_session.commit() fake = _FakeDriver("roles/viewer") with _patch_session_factory(db_session), _patch_driver(fake): result = await sync_roles_for_scope(str(target.id)) assert result.get("upserted") == 1 assert result.get("errors") == [] assert result.get("status") == "success" # Target got the new role. target_row = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.scope_id == str(target.id), CmdbCloudRole.role_key == "roles/viewer", ) ) assert target_row is not None assert target_row.is_promoted is False # driver classification # Sibling row untouched — deleted_at still NULL. sibling_row = await db_session.get(CmdbCloudRole, pre_existing.id) assert sibling_row is not None assert sibling_row.deleted_at is None, ( "sync_roles_for_scope must NOT mark-and-sweep — sibling scope " "rows would be erroneously closed by a per-scope sweep" ) @pytest.mark.asyncio async def test_role_integration_partial_errors_set_failure_status( db_session: AsyncSession, ) -> None: """The production result must let admin chaining stop before grants.""" integ = await _make_integration(db_session, provider="gcp") good_scope = await _make_scope(db_session, integration=integ) bad_scope = await _make_scope(db_session, integration=integ) class _PartialDriver(_FakeDriver): async def list_scope_roles(self, scope: CloudScopeRef): if scope.external_id == bad_scope.external_id: raise RuntimeError("provider page failed") return await super().list_scope_roles(scope) with _patch_session_factory(db_session), _patch_driver( _PartialDriver("roles/viewer"), ): result = await sync_roles_for_integration(str(integ.id)) assert good_scope.external_id != bad_scope.external_id assert result["status"] == "partial_failure" assert result["upserted"] == 1 assert len(result["errors"]) == 1 @pytest.mark.asyncio async def test_sync_roles_for_scope_skips_when_driver_unsupported( db_session: AsyncSession, ): """Driver without ``list_scope_roles`` returns ``status=skipped`` rather than raising — the post-provision background task must never crash because a provider hasn't implemented this capability yet. The scope is already created; the cron is the backstop.""" integ = await _make_integration(db_session, provider="aws") scope = await _make_scope(db_session, integration=integ) class _StubDriver: async def list_scope_roles(self, scope_ref): # noqa: ARG002 raise NotImplementedError() async def close(self): return None with _patch_session_factory(db_session), _patch_driver(_StubDriver()): result = await sync_roles_for_scope(str(scope.id)) assert result.get("status") == "skipped" @pytest.mark.asyncio async def test_sync_roles_for_scope_preserves_admin_promote_flip( db_session: AsyncSession, ): """Re-running sync_roles_for_scope on a scope where an admin has toggled ``is_promoted`` must NOT clobber the flip, mirroring the cron-driven sync_roles_for_integration semantics.""" integ = await _make_integration(db_session, provider="azure") scope = await _make_scope(db_session, integration=integ) fake = _FakeDriver("Reader") # First sync — driver says is_promoted=False. with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_scope(str(scope.id)) row = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.scope_id == str(scope.id), CmdbCloudRole.role_key == "Reader", ) ) assert row is not None and row.is_promoted is False # Admin flips it on. row.is_promoted = True await db_session.commit() # Re-sync — flip must survive. with _patch_session_factory(db_session), _patch_driver(fake): await sync_roles_for_scope(str(scope.id)) row2 = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.scope_id == str(scope.id), CmdbCloudRole.role_key == "Reader", ) ) assert row2 is not None and row2.is_promoted is True @pytest.mark.asyncio async def test_provision_initial_owner_grant_binds_active_grant( db_session: AsyncSession, ): """The post-provision auto-Owner grant inserts an ACTIVE ``CmdbCloudAccessGrant`` for the scope owner and invokes the driver's ``grant_access`` so the requester immediately appears in the member list AND has cloud-side IAM.""" integ = await _make_integration(db_session, provider="gcp") owner = await _make_person(db_session, email="auto-owner@p.io", name="Auto Owner") scope = await _make_scope(db_session, integration=integ, owner=owner) driver_calls: list[dict] = [] class _DriverWithGrant: async def grant_access(self, ctx): driver_calls.append({ "principal_email": ctx.principal_email, "role_key": ctx.role_key, "scope_external_id": ctx.scope.external_id, }) from app.integrations.driver_base import CloudGrantResult return CloudGrantResult( ok=True, external_ref=f"gcp:setIamPolicy:{ctx.scope.external_id}:{ctx.role_key}", ) async def close(self): return None # The dispatch path loads the driver via ``_load_driver_for_scope`` # (NOT ``_load_cloud_driver`` directly) — patch that too. async def _stub_loader(_db, _scope): return integ, _DriverWithGrant() from app.domains.cmdb.lifecycle_service import CloudLifecycleService with _patch_session_factory(db_session), patch( "app.domains.cmdb.lifecycle_service._load_driver_for_scope", side_effect=_stub_loader, ): service = CloudLifecycleService(db_session) outcome = await service.provision_initial_owner_grant( scope_id=str(scope.id), role_key="roles/owner", role_display_name="Owner", trace_id="t-auto-owner-1", ) assert outcome is not None assert outcome.status == GRANT_STATUS_ACTIVE # Driver was actually invoked. assert len(driver_calls) == 1 assert driver_calls[0]["role_key"] == "roles/owner" assert driver_calls[0]["principal_email"] == "auto-owner@p.io" # Grant row landed. rows = (await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), ) )).scalars().all() assert len(rows) == 1 assert rows[0].status == GRANT_STATUS_ACTIVE assert rows[0].role_key == "roles/owner" assert rows[0].person_id == str(owner.id) assert (rows[0].detail or {}).get("auto_owner_grant") is True @pytest.mark.asyncio async def test_provision_initial_owner_grant_skips_unowned_scope( db_session: AsyncSession, ): """Scope without ``owner_person_id`` → helper returns None and writes no grant row. Defensive against a race where the post- provision task fires for a scope whose owner field was cleared by a concurrent admin action between scheduling and execution.""" integ = await _make_integration(db_session, provider="gcp") # Build a scope with no owner. scope = await _make_scope(db_session, integration=integ, owner=None) from app.domains.cmdb.lifecycle_service import CloudLifecycleService service = CloudLifecycleService(db_session) outcome = await service.provision_initial_owner_grant( scope_id=str(scope.id), role_key="roles/owner", role_display_name="Owner", trace_id="t-unowned", ) assert outcome is None rows = (await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), ) )).scalars().all() assert rows == [] @pytest.mark.asyncio async def test_provision_initial_owner_grant_driver_failure_lands_pending( db_session: AsyncSession, ): """If the driver's ``grant_access`` fails (e.g. IAM API blip) the grant row is committed in PENDING state so the requester / admin can retry via the standard owner-add flow. The auto-grant is best-effort — never bubbles to fail the entire provision operation.""" integ = await _make_integration(db_session, provider="gcp") owner = await _make_person(db_session, email="pending@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) class _FailingDriver: async def grant_access(self, ctx): # noqa: ARG002 from app.integrations.driver_base import CloudGrantResult return CloudGrantResult( ok=False, error_message="iam api 503 hiccup", ) async def close(self): return None async def _stub_loader(_db, _scope): return integ, _FailingDriver() from app.domains.cmdb.lifecycle_service import CloudLifecycleService with _patch_session_factory(db_session), patch( "app.domains.cmdb.lifecycle_service._load_driver_for_scope", side_effect=_stub_loader, ): service = CloudLifecycleService(db_session) outcome = await service.provision_initial_owner_grant( scope_id=str(scope.id), role_key="roles/owner", role_display_name="Owner", trace_id="t-pending", ) assert outcome is not None assert outcome.status == GRANT_STATUS_PENDING # Row landed and recorded the last error so a follow-up retry can # surface it to the operator. rows = (await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), ) )).scalars().all() assert len(rows) == 1 assert rows[0].status == GRANT_STATUS_PENDING assert "iam api 503" in str((rows[0].detail or {}).get("last_error") or "") def test_validate_name_accepts_canonical_pattern(): assert _validate_name("tokyo-platform-canary") == "tokyo-platform-canary" assert _validate_name(" spaced name 1 ") == "spaced name 1" def test_validate_name_rejects_special_chars(): with pytest.raises(AppException) as exc: _validate_name("bad/name") assert exc.value.status_code == 400 def test_validate_name_rejects_over_50_chars(): # Becomes the cloud-side DisplayName (Aliyun RD / AWS Orgs cap 50). # 51 chars must be rejected at the source, not surface as a downstream # cloud-provider rejection (2026-06-03 source-validation hardening). assert len(_validate_name("a" * 50)) == 50 # boundary OK with pytest.raises(AppException) as exc: _validate_name("a" * 51) assert exc.value.status_code == 400 def test_validate_name_rejects_no_alphanumeric(): # All separators → the Aliyun driver's alphanumeric-only # AccountNamePrefix would be empty (login-email seed). Block at source # instead of silently falling back to a 'helix' prefix. with pytest.raises(AppException) as exc: _validate_name("- -- -") assert exc.value.status_code == 400 def test_validate_budget_rejects_zero_and_huge(): with pytest.raises(AppException): _validate_budget(0) with pytest.raises(AppException): _validate_budget(20_000_000) assert _validate_budget(1000) == Decimal("1000.00") # ── Helpers ─────────────────────────────────────────────────────── class _FakeSession: """Stand-in for `_build_default_alert_rules`'s ``session`` arg — it only reads platform settings and we want to test the rule-shape transformation, not the storage layer.""" async def get(self, *_args, **_kw): # PlatformSetting lookup return None # Falls through to the in-code default template class _FakeDriver: """Asyncio-flavoured fake that returns a single role with ``is_promoted=False`` so the test can lock in the "driver classification is the insert default" semantic.""" def __init__(self, role_key: str) -> None: self.role_key = role_key async def list_scope_roles(self, scope: CloudScopeRef): return [ CloudRoleInfo( role_key=self.role_key, display_name="Reader", is_custom=False, source="azure_builtin", scope=scope, is_promoted=False, ), ] def classify_role(self, role_key, display_name, scope_type): # noqa: ARG002 from app.integrations.cloud_role_governance import classify_role return classify_role(role_key, display_name) async def close(self): return None class _MultiRoleFakeDriver: """Fake driver that returns N distinct roles on every ``list_scope_roles`` call — used to seed a baseline that the drop-tolerance guard can then trip. """ def __init__(self, role_keys: list[str]) -> None: self.role_keys = list(role_keys) async def list_scope_roles(self, scope: CloudScopeRef): return [ CloudRoleInfo( role_key=k, display_name=k, is_custom=True, source="custom", scope=scope, is_promoted=True, ) for k in self.role_keys ] def classify_role(self, role_key, display_name, scope_type): # noqa: ARG002 from app.integrations.cloud_role_governance import classify_role return classify_role(role_key, display_name) async def close(self): return None class _TenantWideRolesDriver(_MultiRoleFakeDriver): """Returns the same tenant-wide role set from every scope enumeration.""" async def list_scope_roles(self, scope: CloudScopeRef): # noqa: ARG002 return [ CloudRoleInfo( role_key=role_key, display_name=role_key, is_custom=True, source="custom", scope=None, is_promoted=True, ) for role_key in self.role_keys ] @pytest.mark.asyncio async def test_single_scope_prewarm_reuses_tenant_wide_canonical_row( db_session: AsyncSession, ): """Full sync and post-provision prewarm share NULL-scope identity.""" integ = await _make_integration(db_session, provider="aliyun") first_scope = await _make_scope(db_session, integration=integ) second_scope = await _make_scope(db_session, integration=integ) driver = _TenantWideRolesDriver(["tenant-reader"]) with _patch_session_factory(db_session), _patch_driver(driver): await sync_roles_for_integration(str(integ.id)) canonical = await db_session.scalar( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integ.id), CmdbCloudRole.scope_id.is_(None), CmdbCloudRole.role_key == "tenant-reader", CmdbCloudRole.deleted_at.is_(None), ) ) assert canonical is not None canonical_id = canonical.id canonical.is_promoted = False canonical.governance_overridden_by_admin = True canonical.governance_tier = "privileged" canonical.governance_reason = "admin override" await db_session.commit() with _patch_session_factory(db_session), _patch_driver(driver): await sync_roles_for_scope(str(second_scope.id)) await sync_roles_for_scope(str(first_scope.id)) rows = (await db_session.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integ.id), CmdbCloudRole.scope_id.is_(None), CmdbCloudRole.role_key == "tenant-reader", CmdbCloudRole.deleted_at.is_(None), ) )).scalars().all() assert len(rows) == 1 assert rows[0].id == canonical_id assert rows[0].is_promoted is False assert rows[0].governance_tier == "privileged" assert rows[0].governance_reason == "admin override" @pytest.mark.asyncio async def test_full_role_sync_retires_historical_null_scope_duplicates( db_session: AsyncSession, ): """Logical baseline grouping lets the normal sweep clean old duplicates.""" integ = await _make_integration(db_session, provider="aliyun") await _make_scope(db_session, integration=integ) old_job_id = str(uuid.uuid4()) for _ in range(3): db_session.add(CmdbCloudRole( integration_id=str(integ.id), scope_id=None, role_key="tenant-reader", display_name="Tenant reader", is_custom=True, source="custom", raw_data={}, is_promoted=True, last_seen_sync_job_id=old_job_id, )) await db_session.commit() driver = _TenantWideRolesDriver(["tenant-reader", "tenant-auditor"]) with _patch_session_factory(db_session), _patch_driver(driver): result = await sync_roles_for_integration(str(integ.id)) assert result["upserted"] == 2 alive = (await db_session.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integ.id), CmdbCloudRole.scope_id.is_(None), CmdbCloudRole.deleted_at.is_(None), ) )).scalars().all() assert sorted(row.role_key for row in alive) == [ "tenant-auditor", "tenant-reader", ] assert result["deleted"] == 2 @pytest.mark.asyncio async def test_tenant_role_dedupe_merges_admin_fields_from_duplicates( db_session: AsyncSession, ) -> None: """Canonicalization must not sweep away curation on another duplicate.""" integ = await _make_integration(db_session, provider="aliyun") await _make_scope(db_session, integration=integ) old_job_id = str(uuid.uuid4()) uncurated = CmdbCloudRole( integration_id=str(integ.id), scope_id=None, role_key="tenant-curated", display_name="Old provider name", is_custom=True, source="custom", raw_data={}, is_promoted=True, last_seen_sync_job_id=old_job_id, ) curated = CmdbCloudRole( integration_id=str(integ.id), scope_id=None, role_key="tenant-curated", display_name="Duplicate provider name", description_override="Admin description", is_custom=True, source="custom", raw_data={}, is_promoted=False, sort_order=7, governance_tier="privileged", governance_reason="admin decision", governance_overridden_by_admin=True, last_seen_sync_job_id=old_job_id, ) db_session.add_all([uncurated, curated]) await db_session.commit() driver = _TenantWideRolesDriver(["tenant-curated"]) with _patch_session_factory(db_session), _patch_driver(driver): result = await sync_roles_for_integration(str(integ.id)) active_rows = (await db_session.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integ.id), CmdbCloudRole.scope_id.is_(None), CmdbCloudRole.role_key == "tenant-curated", CmdbCloudRole.deleted_at.is_(None), ) )).scalars().all() assert result["status"] == "success" assert len(active_rows) == 1 canonical = active_rows[0] assert canonical.description_override == "Admin description" assert canonical.is_promoted is False assert canonical.sort_order == 7 assert canonical.governance_overridden_by_admin is True assert canonical.governance_tier == "privileged" assert canonical.governance_reason == "admin decision" # ── roles_sync force_sweep override (2026-05-29) ────────────────── @pytest.mark.asyncio async def test_roles_sync_drop_ratio_guard_trips_without_force( db_session: AsyncSession, ): """Baseline test: when the driver returns far fewer roles than the active Helix baseline (>20% drop), ``sync_roles_for_integration`` must raise ``RuntimeError`` and refuse the soft-delete sweep. Mirrors the PR #565 prod incident shape: 10 674 baseline rows vs 288 returned roles. Test uses 10 baseline vs 1 returned so the ratio still trips the 20% floor. """ integ = await _make_integration(db_session, provider="aliyun") await _make_scope(db_session, integration=integ) integ_id = str(integ.id) # Seed a 10-role baseline. seed = _MultiRoleFakeDriver([f"role-{i}" for i in range(10)]) with _patch_session_factory(db_session), _patch_driver(seed): await sync_roles_for_integration(integ_id) baseline_count = await db_session.scalar( select(sa_func.count(CmdbCloudRole.id)).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.deleted_at.is_(None), CmdbCloudRole.last_seen_sync_job_id.is_not(None), ) ) assert baseline_count == 10 # Driver now returns only 1 role — that's a 90% drop, well past # the 20% floor. shrunk = _MultiRoleFakeDriver(["role-0"]) with _patch_session_factory(db_session), _patch_driver(shrunk): with pytest.raises(RuntimeError, match="refusing soft-delete sweep"): await sync_roles_for_integration(integ_id) # The other 9 rows MUST NOT have been soft-deleted. survivors = await db_session.scalar( select(sa_func.count(CmdbCloudRole.id)).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.deleted_at.is_(None), ) ) assert survivors == 10, ( "guard tripped but old rows were soft-deleted anyway — " "the raise happened AFTER the sweep ran" ) @pytest.mark.asyncio async def test_roles_sync_guard_uses_unique_tenant_wide_rows( db_session: AsyncSession, ): """Per-scope fan-out of one tenant-wide role must count as one. Five scope calls each upsert the same ``scope_id=NULL`` unique role. The raw loop counter is five, but only one role was observed; against a five-role baseline this must fail closed instead of deleting four roles. """ integ = await _make_integration(db_session, provider="aliyun") for _ in range(5): await _make_scope(db_session, integration=integ) integ_id = str(integ.id) seed = _TenantWideRolesDriver([f"role-{index}" for index in range(5)]) with _patch_session_factory(db_session), _patch_driver(seed): await sync_roles_for_integration(integ_id) duplicate = _TenantWideRolesDriver(["role-0"]) with _patch_session_factory(db_session), _patch_driver(duplicate): with pytest.raises(RuntimeError, match=r"persisted 1 unique roles \(5 raw upserts\)"): await sync_roles_for_integration(integ_id) survivors = await db_session.scalar( select(sa_func.count(CmdbCloudRole.id)).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.deleted_at.is_(None), ) ) assert survivors == 5 @pytest.mark.asyncio async def test_roles_sync_force_sweep_bypasses_drop_ratio_guard( db_session: AsyncSession, ): """``force_sweep=True`` bypasses the 20% drop floor, sweeps the stale rows, and surfaces ``forced_floor_bypass`` in the result. This is the escape valve for the PR #565 incident shape: an intentional driver-side filter change collapses the catalog and the admin needs a one-click way to accept the new state without the guard staying tripped on every future cron. """ integ = await _make_integration(db_session, provider="aliyun") await _make_scope(db_session, integration=integ) integ_id = str(integ.id) # Seed 10 baseline. seed = _MultiRoleFakeDriver([f"role-{i}" for i in range(10)]) with _patch_session_factory(db_session), _patch_driver(seed): await sync_roles_for_integration(integ_id) # Driver now returns 1 — would normally trip the guard. shrunk = _MultiRoleFakeDriver(["role-0"]) with _patch_session_factory(db_session), _patch_driver(shrunk): result = await sync_roles_for_integration(integ_id, force_sweep=True) # Bypass block recorded in the result so the endpoint can mirror # it into the audit row and the UI can show what was bypassed. bypass = result.get("forced_floor_bypass") assert bypass is not None, "forced_floor_bypass missing from result" assert bypass["trip"] == "drop_ratio" assert bypass["baseline"] == 10 assert bypass["upserted"] == 1 # tolerance_pct + floor come from sweep_settings; just check shape. assert "tolerance_pct" in bypass and "floor" in bypass # 9 stale rows MUST now be soft-deleted; the 1 returned role stays # live with its ``deleted_at`` cleared. alive = await db_session.scalar( select(sa_func.count(CmdbCloudRole.id)).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.deleted_at.is_(None), ) ) assert alive == 1, ( f"force_sweep didn't soft-delete the 9 stale rows ({alive} alive)" ) swept = await db_session.scalar( select(sa_func.count(CmdbCloudRole.id)).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.deleted_at.is_not(None), ) ) assert swept == 9 @pytest.mark.asyncio async def test_roles_sync_force_sweep_bypasses_every_scope_failed_guard( db_session: AsyncSession, ): """``force_sweep=True`` also bypasses the every-scope-failed guard (upserted == 0 with non-zero baseline). The bypass trip type is ``every_scope_failed`` and the result records the error count. Realistic case: driver hits a permanent auth failure on every scope AND admin has decided the integration is being decommissioned so the stale catalog should be cleared anyway. """ integ = await _make_integration(db_session, provider="aliyun") await _make_scope(db_session, integration=integ) integ_id = str(integ.id) # Seed 5 baseline. seed = _MultiRoleFakeDriver([f"role-{i}" for i in range(5)]) with _patch_session_factory(db_session), _patch_driver(seed): await sync_roles_for_integration(integ_id) # Driver that raises on every list_scope_roles call → 0 upserts, # but errors list is populated. class _BrokenDriver: async def list_scope_roles(self, scope: CloudScopeRef): raise RuntimeError("auth expired on every scope") async def close(self): return None broken = _BrokenDriver() with _patch_session_factory(db_session), _patch_driver(broken): # Without force_sweep: guard raises. with pytest.raises(RuntimeError, match="refusing soft-delete sweep"): await sync_roles_for_integration(integ_id) # With force_sweep: sweep proceeds, all 5 rows soft-deleted. result = await sync_roles_for_integration(integ_id, force_sweep=True) bypass = result.get("forced_floor_bypass") assert bypass is not None assert bypass["trip"] == "every_scope_failed" assert bypass["baseline"] == 5 assert bypass["upserted"] == 0 assert bypass["errors"] >= 1 swept = await db_session.scalar( select(sa_func.count(CmdbCloudRole.id)).where( CmdbCloudRole.integration_id == integ_id, CmdbCloudRole.deleted_at.is_not(None), ) ) assert swept == 5 # ── Per-basis tier suppression (2026-07-29 review, F3) ──────────────── @pytest.mark.asyncio async def test_tier_suppression_is_per_measure_not_global( db_session: AsyncSession, ): """Tier suppression ("only the highest crossed threshold of a type pages") must be scoped to the rule's cost BASIS. A list-basis threshold and a net-basis threshold are not comparable — "$1,200 > $900" across bases means nothing — so a list winner must not suppress a net sibling, and vice versa. Both must page (once each), and both must claim their dedup slot. """ from app.domains.cmdb.lifecycle_service import MeasureAggregate integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-tier-measure@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.alert_rules = [ # Net basis: crossed by net $500. {"type": "daily", "threshold_usd": 300.0, "active": True}, # List basis: crossed by list $1,200. Its higher absolute number # must NOT suppress the net rule above. {"type": "daily", "threshold_usd": 900.0, "active": True, "measure": "list"}, ] await db_session.commit() sent: list[str] = [] async def _capture(env, **_kw): sent.append(env.title) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired = await evaluate_scope_alert_rules( db_session, scope, today_cost=500.0, today_date=date(2026, 6, 3), month_to_date_cost=500.0, measure="net", extra_measures={ "list": MeasureAggregate( today_cost=1200.0, month_to_date_cost=1200.0, ), }, integration=integ, ) assert fired == 2, ( "one basis' winner suppressed the other basis' rule — tier " "suppression is comparing thresholds across cost measures" ) assert len(sent) == 2 @pytest.mark.asyncio async def test_rule_without_its_measure_aggregate_is_skipped( db_session: AsyncSession, ): """A rule stamped with a basis the caller could not compute (incomplete data) must be SKIPPED — never silently compared against the other basis, and never claiming its dedup slot on a comparison that did not happen.""" from app.domains.cmdb.models import CmdbCloudScopeAlertFired integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-skip-measure@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.alert_rules = [ {"type": "daily", "threshold_usd": 100.0, "active": True, "measure": "list"}, ] await db_session.commit() sent: list[str] = [] async def _capture(env, **_kw): sent.append(env.title) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): # Only the NET aggregate is supplied, and it is far above the # list-denominated threshold — a fallback would page falsely. fired = await evaluate_scope_alert_rules( db_session, scope, today_cost=5000.0, today_date=date(2026, 6, 4), month_to_date_cost=5000.0, measure="net", integration=integ, ) assert fired == 0 assert sent == [] rows = (await db_session.execute( select(CmdbCloudScopeAlertFired).where( CmdbCloudScopeAlertFired.scope_id == str(scope.id), ), )).scalars().all() assert rows == [], ( "the skipped rule consumed its dedup slot — it would then stay quiet " "for the whole period once the list data DOES arrive" ) # ── Cross-run dedup is per basis too (2026-07-29 review, F2) ────────── # # Tier suppression was made ``(rule_type, basis)``-aware while the CROSS-RUN # dedup slot stayed ``(rule_type, period_key)``. A net monthly alert firing on # the 3rd therefore consumed the month's only monthly slot and permanently # suppressed the list monthly alert — the engineering-facing default — for the # rest of the month. @pytest.mark.asyncio async def test_net_monthly_alert_does_not_swallow_the_list_one( db_session: AsyncSession, ): """Two runs, same month, different bases: each must page once. Run 1 has only the net aggregate (the list column is still incomplete for part of the MTD window), so the list rule is skipped. Run 2 — a later day in the SAME month, list data now complete — must still be able to page the list rule. """ from app.domains.cmdb.lifecycle_service import MeasureAggregate from app.domains.cmdb.models import CmdbCloudScopeAlertFired integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-dedup-measure@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.alert_rules = [ {"type": "monthly", "threshold_usd": 400.0, "active": True}, {"type": "monthly", "threshold_usd": 900.0, "active": True, "measure": "list"}, ] await db_session.commit() sent: list[str] = [] async def _capture(env, **_kw): sent.append(env.title) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): # Run 1 — net only. The net rule pages; the list rule is skipped # (its aggregate is absent) and claims nothing. fired1 = await evaluate_scope_alert_rules( db_session, scope, today_cost=0.0, today_date=date(2026, 6, 3), month_to_date_cost=500.0, measure="net", integration=integ, ) assert fired1 == 1, sent # Run 2 — same month, list window now complete. fired2 = await evaluate_scope_alert_rules( db_session, scope, today_cost=0.0, today_date=date(2026, 6, 11), month_to_date_cost=520.0, measure="net", extra_measures={ "list": MeasureAggregate( today_cost=0.0, month_to_date_cost=1200.0, ), }, integration=integ, ) assert fired2 == 1, ( "the net monthly alert from run 1 swallowed the list monthly alert — " "the cross-run dedup slot is still basis-blind, so whichever basis " "fires first silences the other for the whole period" ) assert len(sent) == 2 claimed = { (r.rule_type, r.period_key) for r in (await db_session.execute( select(CmdbCloudScopeAlertFired).where( CmdbCloudScopeAlertFired.scope_id == str(scope.id), ), )).scalars().all() } # Net keeps the bare rule_type (pre-split rows keep deduping the net # question); list gets its own slot. No migration, no backfill. assert claimed == {("monthly", "2026-06"), ("monthly:list", "2026-06")} @pytest.mark.asyncio async def test_same_basis_repeat_in_the_period_is_still_deduped( db_session: AsyncSession, ): """The basis split must not weaken the guarantee it splits: within ONE basis, one alert per scope per period, even when the admin retunes the threshold mid-period (a different threshold opens a new UNIQUE key, which is exactly what the pre-claim period gate exists to catch).""" from app.domains.cmdb.lifecycle_service import MeasureAggregate integ = await _make_integration(db_session, provider="aws") owner = await _make_person(db_session, email="owner-dedup-same@p.io") scope = await _make_scope(db_session, integration=integ, owner=owner) scope.alert_rules = [ {"type": "monthly", "threshold_usd": 900.0, "active": True, "measure": "list"}, ] await db_session.commit() sent: list[str] = [] async def _capture(env, **_kw): sent.append(env.title) aggregates = { "list": MeasureAggregate(today_cost=0.0, month_to_date_cost=1200.0), } with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): fired1 = await evaluate_scope_alert_rules( db_session, scope, today_cost=0.0, today_date=date(2026, 7, 3), month_to_date_cost=100.0, measure="net", extra_measures=aggregates, integration=integ, ) assert fired1 == 1 # Admin lowers the list threshold mid-month, and the legacy # ``last_fired`` marker is cleared to prove the DB slot (not the JSON # mirror) is what dedups. scope.alert_rules = [ {"type": "monthly", "threshold_usd": 500.0, "active": True, "measure": "list"}, ] await db_session.commit() fired2 = await evaluate_scope_alert_rules( db_session, scope, today_cost=0.0, today_date=date(2026, 7, 20), month_to_date_cost=100.0, measure="net", extra_measures=aggregates, integration=integ, ) assert fired2 == 0, ( "a same-basis repeat re-paged inside one period — the basis-qualified " "slot must still be threshold-agnostic per (scope, type, basis, period)" ) assert len(sent) == 1