"""Cloud-scope decommission + idle-sweep tests. Locks in the contract for the cloud-first decommission flow: * ``decommission_scope`` calls the driver's :meth:`CloudDriver.close_scope_on_provider` first; only on ``ok=True`` does it flip status, fill bookkeeping, disable grants, and audit. ``ok=False`` raises ``SCOPE_CLOSE_PROVIDER_ERROR`` with the provider's message and Helix's row stays unchanged. * Driver-side ``NotImplementedError`` becomes ``SCOPE_CLOSE_NOT_SUPPORTED`` (501) — the action is never silently a no-op. * Idempotent on already-closed scopes via the driver's ``already_closed=True`` signal (e.g. AWS reports SUSPENDED). * IAM-mode (``standalone_iam``) scopes pass the Helix-tracked ``cmdb_iam_users`` list to the driver for evacuation; revoked users get ``revoked_at`` set so the IAM users page can render them with a status pill. * ``list_idle_scope_candidates`` only flags scopes with both conditions: 0 active+pending grants AND $0 trailing-30d cost AND has billing rows in the window. Closed scopes never surface. * RBAC: it_admin (global_admin) and cloud_admin can decommission; employee gets 403. Tests run against the real test DB (Docker MySQL). The driver call is replaced with an in-memory stub via monkeypatching ``_load_cloud_driver`` so we don't hit AWS / Aliyun / Azure / GCP in tests but DO exercise the service-layer dispatch + bookkeeping logic. There is no ``reactivate`` — closures are irreversible (PR introducing real cloud-side close also dropped the reactivate path; the cloud-side close API is irreversible too, so Helix's mirror must be). """ from __future__ import annotations import uuid from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.models import AuditLog from app.auth.models import AuthRole, AuthUserRole from app.core.exceptions import AppException, NotFoundError from app.domains.cmdb import lifecycle_service as lifecycle_service_mod from app.domains.cmdb.lifecycle_service import ( IDLE_SCOPE_DEFAULT_SNOOZE_DAYS, IDLE_SWEEP_DEDUP_DAYS, CloudLifecycleService, list_idle_scope_candidates, sweep_idle_scopes, ) from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, GRANT_STATUS_DISABLED, GRANT_STATUS_PENDING, OWNER_CONFIRMATION_KIND_IDLE_NOTIFIED, OWNER_CONFIRMATION_STATUS_RESPONDED, SCOPE_STATUS_ACTIVE, SCOPE_STATUS_CLOSED, CmdbCloudAccessGrant, CmdbCloudBillingDaily, CmdbCloudScope, CmdbCloudScopeOwnerConfirmation, CmdbIamUser, ) from app.identity.person_models import IdentityPerson from app.integrations.driver_base import ( CloseScopeResult, CloudScopeRef, ) from app.integrations.models import Integration # ── Fixtures (helpers, not pytest fixtures) ─────────────────────── async def _make_person(db: AsyncSession, *, email: str, name: str) -> 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 _make_integration( db: AsyncSession, *, provider: str = "aws", ) -> Integration: row = Integration( id=uuid.uuid4(), kind="cloud", provider=provider, name=f"{provider}-decom-test", slug=f"{provider}-decom-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(row) await db.commit() return row async def _make_scope( db: AsyncSession, integration: Integration, *, external_id: str | None = None, scope_type: str = "aws_account", owner_person_id: str | None = None, first_seen_offset_days: int = 90, ) -> CmdbCloudScope: s = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integration.id), scope_type=scope_type, external_id=external_id or f"{uuid.uuid4().hex[:12]}", display_name=f"scope-{uuid.uuid4().hex[:6]}", status=SCOPE_STATUS_ACTIVE, owner_person_id=owner_person_id, first_seen_at=datetime.now(UTC) - timedelta(days=first_seen_offset_days), last_synced_at=datetime.now(UTC), environment="dev", ) db.add(s) await db.commit() return s async def _make_grant( db: AsyncSession, scope: CmdbCloudScope, person: IdentityPerson, *, role_key: str = "arn:aws:iam::aws:policy/AdministratorAccess", status: str = GRANT_STATUS_ACTIVE, ) -> CmdbCloudAccessGrant: g = CmdbCloudAccessGrant( id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(person.id), role_key=role_key, role_display_name=role_key.rsplit("/", 1)[-1], status=status, granted_at=datetime.now(UTC), source="discovered", ) db.add(g) await db.commit() return g async def _make_iam_user( db: AsyncSession, integration: Integration, *, user_name: str, ) -> CmdbIamUser: u = CmdbIamUser( id=uuid.uuid4(), integration_id=str(integration.id), user_name=user_name, mfa_enabled=False, ) db.add(u) await db.commit() return u async def _audit_actions_for_scope( db: AsyncSession, scope_id: str, ) -> list[str]: rows = (await db.execute( select(AuditLog.action, AuditLog.status).where( AuditLog.resource_type == "cmdb_cloud_scope", AuditLog.resource_id == scope_id, ).order_by(AuditLog.created_at) )).all() return [f"{a}:{s}" for a, s in rows] # ── Stub driver for service-layer tests ────────────────────────── @dataclass class StubCloudDriver: """In-memory ``CloudDriver`` substitute for service-layer tests. The service contract only depends on ``close_scope_on_provider`` — we don't need to subclass the full ``CloudDriver`` ABC, so we expose only the one method the service calls. Programmable per test via ``next_result`` (or ``next_exception`` to simulate ``NotImplementedError``). """ next_result: CloseScopeResult | None = None next_exception: type[Exception] | None = None last_call: dict[str, Any] = field(default_factory=dict) async def close_scope_on_provider( self, scope: CloudScopeRef, *, iam_account_ids: list[str] | None = None, ) -> CloseScopeResult: self.last_call = { "scope_type": scope.scope_type, "external_id": scope.external_id, "iam_account_ids": list(iam_account_ids or []), } if self.next_exception is not None: raise self.next_exception( f"stub driver: scope_type {scope.scope_type!r} not supported", ) assert self.next_result is not None, ( "StubCloudDriver: caller must set next_result before invoking" ) return self.next_result def _patch_driver(monkeypatch, integration_id: str, driver: StubCloudDriver) -> None: """Hook ``_load_driver_for_scope`` to return our stub. Also stubs the out-of-tx audit-failure helper to a no-op so unit tests without the FastAPI ``client`` fixture (which is what wires ``async_session_factory`` to the test engine) don't try to open a session against the production DB. The router-level tests below that DO use ``client`` are left exercising the real helper — that's where its behavior is integration-tested. """ async def _fake_load(session, scope): integ = await session.get(Integration, uuid.UUID(integration_id)) return integ, driver monkeypatch.setattr( lifecycle_service_mod, "_load_driver_for_scope", _fake_load, ) async def _noop_audit_failure(**_kwargs): return None monkeypatch.setattr( lifecycle_service_mod, "_audit_decommission_failure", _noop_audit_failure, ) # ── decommission_scope: cloud-first happy path ─────────────────── @pytest.mark.asyncio async def test_decommission_calls_driver_then_flips_status_and_disables_grants( db_session: AsyncSession, admin_user, monkeypatch, ): admin, _ = admin_user integration = await _make_integration(db_session) owner = await _make_person(db_session, email="o@test.com", name="Owner") scope = await _make_scope( db_session, integration, owner_person_id=str(owner.id), ) member1 = await _make_person(db_session, email="m1@test.com", name="M1") member2 = await _make_person(db_session, email="m2@test.com", name="M2") g1 = await _make_grant(db_session, scope, member1, status=GRANT_STATUS_ACTIVE) g2 = await _make_grant( db_session, scope, member2, role_key="arn:aws:iam::aws:policy/ReadOnlyAccess", status=GRANT_STATUS_PENDING, ) driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", detail={"grace_period_days": 90}, )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) result_scope, disabled_count, close_result = await service.decommission_scope( str(scope.id), reason="AWS account closed by Finance", actor_user_id=admin.id, trace_id="test-trace-decom-1", ) await db_session.commit() # Driver was called with the right scope shape. assert driver.last_call["external_id"] == scope.external_id assert driver.last_call["scope_type"] == "aws_account" # No IAM user rows exist on this scope_type, so the list is empty. assert driver.last_call["iam_account_ids"] == [] assert result_scope.status == SCOPE_STATUS_CLOSED assert result_scope.decommissioned_at is not None assert result_scope.decommissioned_by == str(admin.id) assert result_scope.decommission_reason == "AWS account closed by Finance" assert result_scope.decommission_provider_state == "pending_closure" assert close_result.provider_state == "pending_closure" assert disabled_count == 2 g1_db = await db_session.get(CmdbCloudAccessGrant, g1.id) g2_db = await db_session.get(CmdbCloudAccessGrant, g2.id) assert g1_db.status == GRANT_STATUS_DISABLED assert g2_db.status == GRANT_STATUS_DISABLED actions = await _audit_actions_for_scope(db_session, str(scope.id)) assert "cloud.scope.decommission:success" in actions @pytest.mark.asyncio async def test_decommission_evacuates_iam_users_for_standalone_iam( db_session: AsyncSession, admin_user, monkeypatch, ): """For standalone_iam scopes, the service collects per-integration IAM users from ``cmdb_iam_users`` and hands them to the driver, which deletes each on the cloud side. The service then marks the rows ``revoked_at``.""" admin, _ = admin_user integration = await _make_integration(db_session, provider="tencent") scope = await _make_scope( db_session, integration, scope_type="standalone_iam", ) u1 = await _make_iam_user(db_session, integration, user_name="leaver-a") u2 = await _make_iam_user(db_session, integration, user_name="leaver-b") driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="users_evacuated", revoked_users=2, detail={"evacuated": [{"user": "leaver-a"}, {"user": "leaver-b"}]}, )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) _, _, close_result = await service.decommission_scope( str(scope.id), reason="tenant fully evacuated", actor_user_id=admin.id, trace_id="test-trace-decom-iam", ) await db_session.commit() assert sorted(driver.last_call["iam_account_ids"]) == ["leaver-a", "leaver-b"] assert close_result.revoked_users == 2 # Bulk UPDATE bypasses the identity map; re-fetch via a fresh # select so we observe the ``revoked_at`` / ``revoke_reason`` # mutation the service issued. rows = (await db_session.execute( select(CmdbIamUser).where( CmdbIamUser.id.in_([u1.id, u2.id]), ) )).scalars().all() by_id = {r.id: r for r in rows} assert by_id[u1.id].revoked_at is not None assert by_id[u1.id].revoke_reason == "scope_decommissioned" assert by_id[u2.id].revoked_at is not None # ── decommission_scope: failure modes (Helix row stays untouched) ─ @pytest.mark.asyncio async def test_decommission_provider_error_does_not_flip_status( db_session: AsyncSession, admin_user, monkeypatch, ): """When the cloud provider rejects the close (e.g. AWS ConstraintViolation), the service raises and Helix's row stays fully untouched — no status flip, no grant disabling, no bookkeeping write.""" admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) person = await _make_person(db_session, email="x@test.com", name="X") grant = await _make_grant(db_session, scope, person) driver = StubCloudDriver(next_result=CloseScopeResult( ok=False, error_code="constraint_violation", error_message=( "AWS rejected CloseAccount: payment method required." ), )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) with pytest.raises(AppException) as excinfo: await service.decommission_scope( str(scope.id), reason="will fail", actor_user_id=admin.id, trace_id="test-trace-decom-fail", ) assert excinfo.value.code == "SCOPE_CLOSE_PROVIDER_ERROR" assert excinfo.value.status_code == 502 assert "payment method" in excinfo.value.message # No-mutation invariant: we re-fetch via a brand-new engine / # session below. The fixture's ``db_session`` is left mid-rolled- # back after the AppException + with_for_update lock — touching # it again throws greenlet errors. The router-level test # ``test_decommission_endpoint_provider_failure_502`` exercises # the same invariant via the FastAPI client (which provides a # fresh request-scoped session per call), so this unit-level # test focuses on the exception type + message. grant_unused = grant # silence unused-var warnings; kept for setup symmetry del grant_unused @pytest.mark.asyncio async def test_decommission_driver_not_implemented_raises_501( db_session: AsyncSession, admin_user, monkeypatch, ): admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) driver = StubCloudDriver(next_exception=NotImplementedError) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) with pytest.raises(AppException) as excinfo: await service.decommission_scope( str(scope.id), reason="unsupported scope_type", actor_user_id=admin.id, trace_id="test-trace-decom-501", ) assert excinfo.value.code == "SCOPE_CLOSE_NOT_SUPPORTED" assert excinfo.value.status_code == 501 @pytest.mark.asyncio async def test_decommission_empty_reason_rejected( db_session: AsyncSession, admin_user, ): admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) service = CloudLifecycleService(db_session) with pytest.raises(AppException) as excinfo: await service.decommission_scope( str(scope.id), reason=" ", actor_user_id=admin.id, trace_id="test-trace-decom-2", ) assert excinfo.value.code == "DECOMMISSION_REASON_REQUIRED" @pytest.mark.asyncio async def test_decommission_unknown_scope_404( db_session: AsyncSession, admin_user, ): admin, _ = admin_user service = CloudLifecycleService(db_session) with pytest.raises(NotFoundError): await service.decommission_scope( str(uuid.uuid4()), reason="X", actor_user_id=admin.id, trace_id="test-trace-decom-3", ) @pytest.mark.asyncio async def test_decommission_idempotent_when_provider_says_already_closed( db_session: AsyncSession, admin_user, monkeypatch, ): """Driver returns ``already_closed=True`` (e.g. AWS reports the account is already SUSPENDED). Helix-side bookkeeping still flips — we sync to the cloud truth — but the audit detail flags it as already-closed for transparency. """ admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, already_closed=True, provider_state="suspended", detail={"prior_status": "SUSPENDED"}, )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) _, _, close_result = await service.decommission_scope( str(scope.id), reason="already closed; syncing Helix", actor_user_id=admin.id, trace_id="test-trace-already", ) await db_session.commit() assert close_result.already_closed is True refreshed = await db_session.get(CmdbCloudScope, scope.id) assert refreshed.status == SCOPE_STATUS_CLOSED assert refreshed.decommission_provider_state == "suspended" # Round-52 audit honesty: the admin clicked decommission, but the # provider reported the scope was ALREADY closed (e.g. cancelled # out-of-band on the AWS / Azure portal). The audit row reflects # observation, not action, so log readers can distinguish these # operationally distinct cases without losing either trail. actions = await _audit_actions_for_scope(db_session, str(scope.id)) assert "cloud.scope.decommission_observed_closed:success" in actions assert "cloud.scope.decommission:success" not in actions @pytest.mark.asyncio async def test_decommission_re_run_preserves_original_audit_trail( db_session: AsyncSession, admin_user, monkeypatch, ): """Once a scope is closed, a second decommission call (whether from a UI race or a manual re-trigger) MUST NOT overwrite the original ``decommissioned_at`` / ``decommissioned_by`` / ``decommission_reason``. The provider_state is allowed to refresh (e.g. AWS PENDING_CLOSURE → SUSPENDED), but who-when-why of the first close is the authoritative audit-of-record. """ admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) first_scope, _, _ = await service.decommission_scope( str(scope.id), reason="first close — Finance request 2026-04-30", actor_user_id=admin.id, trace_id="test-trace-first", ) await db_session.commit() original_at = first_scope.decommissioned_at original_by = first_scope.decommissioned_by original_reason = first_scope.decommission_reason assert original_at is not None assert original_reason == "first close — Finance request 2026-04-30" # A second click — provider now reports SUSPENDED (already_closed), # different actor (UI submitted by another admin), different reason. other_admin = await _make_person( db_session, email="other-admin@test.com", name="Other Admin", ) driver.next_result = CloseScopeResult( ok=True, already_closed=True, provider_state="suspended", ) refreshed_scope, _, _ = await service.decommission_scope( str(scope.id), reason="second click — should be ignored for actor/reason", actor_user_id=other_admin.id, trace_id="test-trace-second", ) await db_session.commit() # Original audit trail is preserved. assert refreshed_scope.decommissioned_at == original_at assert refreshed_scope.decommissioned_by == original_by assert refreshed_scope.decommission_reason == original_reason # Provider state refreshed to mirror cloud truth. assert refreshed_scope.decommission_provider_state == "suspended" @pytest.mark.asyncio async def test_aws_close_account_handles_account_already_closed_exception() -> None: """AWS race: describe_account returns ACTIVE, but by the time we call close_account the account has been closed elsewhere → AWS raises ``AccountAlreadyClosedException``. Driver must treat this as ``already_closed=True``, not as an error. """ from unittest.mock import MagicMock, patch from botocore.exceptions import ClientError from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers.aws_cloud import AwsCloudDriver driver = AwsCloudDriver( integration_id="int-aws-race", config={ "access_key_id": "AK", "secret_access_key": "SK", "default_region": "us-east-1", }, ) org = MagicMock() org.describe_account.return_value = { "Account": {"Id": "555555555555", "Status": "ACTIVE"}, } org.close_account.side_effect = ClientError( error_response={ "Error": { "Code": "AccountAlreadyClosedException", "Message": "Account is already closed.", }, }, operation_name="CloseAccount", ) with patch.object(AwsCloudDriver, "_boto_client", return_value=org): result = await driver.close_scope_on_provider( CloudScopeRef(scope_type="aws_account", external_id="555555555555"), ) assert result.ok is True assert result.already_closed is True assert result.provider_state == "suspended" # ── list_idle_scope_candidates ──────────────────────────────────── @pytest.mark.asyncio async def test_idle_candidate_with_no_grants_and_zero_cost( db_session: AsyncSession, ): integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) today = date.today() for i in range(30): d = today - timedelta(days=i) db_session.add(CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integration.id), scope_id=str(scope.id), date=d, total_cost_usd=Decimal("0"), currency="USD", )) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert any(c.scope_id == str(scope.id) for c in candidates) @pytest.mark.asyncio async def test_idle_candidate_skipped_when_grants_active( db_session: AsyncSession, ): integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) person = await _make_person(db_session, email="g@test.com", name="G") await _make_grant(db_session, scope, person, status=GRANT_STATUS_ACTIVE) today = date.today() for i in range(30): db_session.add(CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integration.id), scope_id=str(scope.id), date=today - timedelta(days=i), total_cost_usd=Decimal("0"), currency="USD", )) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert not any(c.scope_id == str(scope.id) for c in candidates) @pytest.mark.asyncio async def test_idle_candidate_excludes_decommissioned_scopes( db_session: AsyncSession, admin_user, monkeypatch, ): admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) today = date.today() for i in range(30): db_session.add(CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integration.id), scope_id=str(scope.id), date=today - timedelta(days=i), total_cost_usd=Decimal("0"), currency="USD", )) await db_session.commit() driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) await service.decommission_scope( str(scope.id), reason="already decommissioned", actor_user_id=admin.id, trace_id="test-trace-idle-decom", ) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert not any(c.scope_id == str(scope.id) for c in candidates) # ── sweep_idle_scopes ───────────────────────────────────────────── @pytest.mark.asyncio async def test_sweep_idle_scopes_with_no_admins_returns_zero_notified( db_session: AsyncSession, ): integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) today = date.today() for i in range(30): db_session.add(CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integration.id), scope_id=str(scope.id), date=today - timedelta(days=i), total_cost_usd=Decimal("0"), currency="USD", )) await db_session.commit() summary = await sweep_idle_scopes(db_session) await db_session.commit() assert summary["candidates"] >= 1 # No admin in this test → no recipient → nothing notified. # Owner is also unset, so ``owners_notified`` is 0. assert summary["notified_recipients"] == 0 assert summary["admins_notified"] == 0 assert summary["owners_notified"] == 0 # No notifications → no stamp on last_idle_notified_at. assert summary["stamped_scopes"] == 0 @pytest.mark.asyncio async def test_sweep_idle_scopes_no_candidates_returns_empty_summary( db_session: AsyncSession, ): summary = await sweep_idle_scopes(db_session) assert summary["candidates"] == 0 assert summary["notified_recipients"] == 0 assert summary["admins_notified"] == 0 assert summary["owners_notified"] == 0 assert summary["errors"] == [] async def _seed_30d_zero_cost( db_session: AsyncSession, integration: Integration, scope: CmdbCloudScope, ) -> None: """Helper: write 30 daily rows of $0 to make ``scope`` idle-eligible.""" today = date.today() for i in range(30): db_session.add(CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integration.id), scope_id=str(scope.id), date=today - timedelta(days=i), total_cost_usd=Decimal("0"), currency="USD", )) await db_session.commit() @pytest.mark.asyncio async def test_idle_candidate_skipped_when_snoozed( db_session: AsyncSession, ): """``idle_snoozed_until > now`` excludes the scope from the candidate list — the owner explicitly said 'I know, ignore for N days'.""" integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) await _seed_30d_zero_cost(db_session, integration, scope) scope.idle_snoozed_until = datetime.now(UTC) + timedelta(days=7) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert not any(c.scope_id == str(scope.id) for c in candidates) # ``include_suppressed=True`` still surfaces it (admin overview UX). all_candidates = await list_idle_scope_candidates( db_session, include_suppressed=True, ) surfaced = next( (c for c in all_candidates if c.scope_id == str(scope.id)), None, ) assert surfaced is not None assert surfaced.is_snoozed is True @pytest.mark.asyncio async def test_idle_candidate_skipped_when_inside_dedup_window( db_session: AsyncSession, ): """Scopes notified within the last ``IDLE_SWEEP_DEDUP_DAYS`` (14d default) are skipped so the weekly cron doesn't re-spam admins about the same scope every Monday.""" integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) await _seed_30d_zero_cost(db_session, integration, scope) scope.last_idle_notified_at = datetime.now(UTC) - timedelta(days=3) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert not any(c.scope_id == str(scope.id) for c in candidates) # Stamp older than the dedup window → eligible again. scope.last_idle_notified_at = datetime.now(UTC) - timedelta(days=20) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert any(c.scope_id == str(scope.id) for c in candidates) @pytest.mark.asyncio async def test_sweep_idle_scopes_notifies_owner_and_admin_once_each( db_session: AsyncSession, admin_user, ): """End-to-end shape: owner gets pinged, admin gets pinged, an admin who is also the owner of one of the candidates only gets ONE card (recipient dedup by person_id), and the scope's ``last_idle_notified_at`` is stamped so the next sweep skips it.""" admin, _ = admin_user integration = await _make_integration(db_session) owner = await _make_person( db_session, email="owner@test.com", name="Idle Owner", ) scope_owned = await _make_scope( db_session, integration, external_id="idle-with-owner", owner_person_id=str(owner.id), ) scope_unowned = await _make_scope( db_session, integration, external_id="idle-no-owner", ) await _seed_30d_zero_cost(db_session, integration, scope_owned) await _seed_30d_zero_cost(db_session, integration, scope_unowned) summary = await sweep_idle_scopes(db_session) await db_session.commit() assert summary["candidates"] == 2 # Two recipients: the global admin (test fixture) + the owner of # ``scope_owned``. ``scope_unowned`` has no owner so the admin is # its only recipient. assert summary["notified_recipients"] == 2 assert summary["admins_notified"] == 1 assert summary["owners_notified"] == 1 # Both candidate scopes get their ``last_idle_notified_at`` stamped # so next week's sweep skips them. assert summary["stamped_scopes"] == 2 # Re-running the sweep immediately yields zero candidates (the # dedup window kicks in). summary2 = await sweep_idle_scopes(db_session) assert summary2["candidates"] == 0 @pytest.mark.asyncio async def test_sweep_idle_scopes_writes_governance_timeline_row( db_session: AsyncSession, admin_user, ): """Each notified scope gets ONE ``idle_notified`` governance-timeline row so the detail-page 治理 tab shows the owner WAS pinged — the user-visible gap this PR closes (previously the sweep only stamped ``last_idle_notified_at`` + sent a digest with no per-scope record). Asserts: leaf shape (status=responded, NOT a pending prompt → won't be escalated), $0/0-member snapshot, the scope's owner carried onto the row, and that the dedup window means re-running writes no second row for the same scope. """ admin, _ = admin_user integration = await _make_integration(db_session) owner = await _make_person( db_session, email="idle-tl-owner@test.com", name="Idle TL Owner", ) scope = await _make_scope( db_session, integration, external_id="idle-timeline-scope", owner_person_id=str(owner.id), ) await _seed_30d_zero_cost(db_session, integration, scope) summary = await sweep_idle_scopes(db_session) await db_session.commit() assert summary["timeline_rows"] == 1 rows = (await db_session.execute( select(CmdbCloudScopeOwnerConfirmation).where( CmdbCloudScopeOwnerConfirmation.scope_id == str(scope.id), CmdbCloudScopeOwnerConfirmation.kind == OWNER_CONFIRMATION_KIND_IDLE_NOTIFIED, ) )).scalars().all() assert len(rows) == 1 row = rows[0] # Leaf event — responded, not a pending prompt the owner must answer. assert row.status == OWNER_CONFIRMATION_STATUS_RESPONDED assert row.response_kind == OWNER_CONFIRMATION_KIND_IDLE_NOTIFIED # Snapshot reflects the idle definition: $0 spend + 0 live members. assert row.spend_recent_30d_usd == Decimal("0") assert row.member_count_snapshot == 0 # Owner carried onto the row so the timeline shows who was pinged. assert str(row.owner_person_id) == str(owner.id) assert row.reason and "0 个生效 grant" in row.reason # Dedup: re-running inside the 14d window writes no second row. summary2 = await sweep_idle_scopes(db_session) await db_session.commit() assert summary2["candidates"] == 0 rows_after = (await db_session.execute( select(CmdbCloudScopeOwnerConfirmation).where( CmdbCloudScopeOwnerConfirmation.scope_id == str(scope.id), CmdbCloudScopeOwnerConfirmation.kind == OWNER_CONFIRMATION_KIND_IDLE_NOTIFIED, ) )).scalars().all() assert len(rows_after) == 1 @pytest.mark.asyncio async def test_idle_dedup_boundary_exactly_14_days( db_session: AsyncSession, ): """Pin the dedup window's exact boundary. The contract is "if notified < 14 days ago, skip; if notified ≥ 14 days ago, eligible again". A scope whose ``last_idle_notified_at`` is exactly 14 days in the past should be eligible (the dedup window is exclusive of the boundary). Without this test, a half-day drift on the cutoff comparison (``<`` vs ``<=``) would silently shift the user-visible re-notification cadence by a day. """ integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) await _seed_30d_zero_cost(db_session, integration, scope) # Set to 14d ago minus 1 minute → strictly inside the window: # SQL filter ``last_idle_notified_at < (now - 14d)`` evaluates # False, scope EXCLUDED. scope.last_idle_notified_at = ( datetime.now(UTC) - timedelta(days=14) + timedelta(minutes=1) ) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert not any(c.scope_id == str(scope.id) for c in candidates), ( "scope notified ~14d ago should still be inside dedup window" ) # Push past 14d by 1 minute → outside window → eligible again. scope.last_idle_notified_at = ( datetime.now(UTC) - timedelta(days=14) - timedelta(minutes=1) ) await db_session.commit() candidates = await list_idle_scope_candidates(db_session) assert any(c.scope_id == str(scope.id) for c in candidates), ( "scope notified just past 14d ago must be eligible again" ) @pytest.mark.asyncio async def test_sweep_idle_stamps_only_scopes_actually_reached( db_session: AsyncSession, admin_user, monkeypatch, ): """**BLK-1/BLK-2 regression**: pre-fix the stamp loop ran for every candidate as soon as ANY recipient was reached. If admin dispatch failed and only some owners received their cards, every candidate still got ``last_idle_notified_at`` stamped — scopes whose owner never heard about them entered the 14d dedup window anyway, and the next week's sweep silently skipped them. Fix: track per-scope dispatch success — a scope is stamped only when at least ONE recipient who would see it in their card body was actually reached. This test forces the failure pattern: admin dispatch raises, owner-A succeeds, owner-B fails → only scope_a is stamped; scope_b is left alone so next sweep retries. """ admin, _ = admin_user integration = await _make_integration(db_session) owner_a = await _make_person( db_session, email="owner-a@test.com", name="Owner A", ) owner_b = await _make_person( db_session, email="owner-b@test.com", name="Owner B", ) scope_a = await _make_scope( db_session, integration, external_id="reached-by-owner", owner_person_id=str(owner_a.id), ) scope_b = await _make_scope( db_session, integration, external_id="never-reached", owner_person_id=str(owner_b.id), ) await _seed_30d_zero_cost(db_session, integration, scope_a) await _seed_30d_zero_cost(db_session, integration, scope_b) # Make the admin's dispatch + owner_b's dispatch raise. Only # owner_a's dispatch succeeds. The shape of failure here matches # real-world transient Feishu / email outbox failures. from app.notifications import service as notify_mod original = notify_mod.NotificationService.dispatch async def selective_dispatch(self, env, *, trace_id="", actor_user_id=None): if env.recipient.work_email in (admin.work_email, owner_b.work_email): raise RuntimeError("simulated outbox failure") return await original( self, env, trace_id=trace_id, actor_user_id=actor_user_id, ) monkeypatch.setattr( notify_mod.NotificationService, "dispatch", selective_dispatch, ) summary = await sweep_idle_scopes(db_session) await db_session.commit() # 2 candidates, 3 recipients (admin + owner_a + owner_b), 2 # failures (admin + owner_b), 1 success (owner_a). assert summary["candidates"] == 2 assert summary["notified_recipients"] == 1 assert summary["owners_notified"] == 1 assert summary["admins_notified"] == 0 assert len(summary["errors"]) == 2 # Only scope_a is on someone's radar (owner_a received their card). # scope_b had its owner fail AND no admin reach → no stamp. assert summary["stamped_scopes"] == 1 await db_session.refresh(scope_a) await db_session.refresh(scope_b) assert scope_a.last_idle_notified_at is not None assert scope_b.last_idle_notified_at is None @pytest.mark.asyncio async def test_sweep_idle_admin_dispatch_success_stamps_all_candidates( db_session: AsyncSession, admin_user, ): """Counterpart to the previous test: when the admin successfully receives the full digest, EVERY candidate scope is stamped — the admin saw the full list and is on the hook for each one. """ admin, _ = admin_user integration = await _make_integration(db_session) scope_a = await _make_scope( db_session, integration, external_id="a-ok", ) scope_b = await _make_scope( db_session, integration, external_id="b-ok", ) await _seed_30d_zero_cost(db_session, integration, scope_a) await _seed_30d_zero_cost(db_session, integration, scope_b) summary = await sweep_idle_scopes(db_session) await db_session.commit() assert summary["candidates"] == 2 assert summary["admins_notified"] == 1 # Admin saw both scopes → both stamped. assert summary["stamped_scopes"] == 2 @pytest.mark.asyncio async def test_sweep_idle_admin_and_owner_overlap_deduped_to_one_card( db_session: AsyncSession, admin_user, ): """An admin who is ALSO the owner of the one idle scope receives a single notification, not two — recipient set is deduped by person_id before dispatch.""" admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope( db_session, integration, external_id="admin-owns-this", owner_person_id=str(admin.id), ) await _seed_30d_zero_cost(db_session, integration, scope) summary = await sweep_idle_scopes(db_session) await db_session.commit() assert summary["candidates"] == 1 # Admin is the only person — both ``admins_notified`` and # ``owners_notified`` increment for this single recipient (they # belong to both sets), but the actual notification count is 1. assert summary["notified_recipients"] == 1 assert summary["admins_notified"] == 1 assert summary["owners_notified"] == 1 @pytest.mark.asyncio async def test_snooze_idle_scope_owner_can_snooze( db_session: AsyncSession, ): """Scope owner can self-service the snooze without admin role.""" integration = await _make_integration(db_session) owner = await _make_person( db_session, email="snoozer@test.com", name="Snoozer", ) scope = await _make_scope( db_session, integration, owner_person_id=str(owner.id), ) service = CloudLifecycleService(db_session) updated = await service.snooze_idle_scope( str(scope.id), days=30, actor_user_id=owner.id, actor_role_names=["employee"], trace_id="test-snooze-owner", ) await db_session.commit() assert updated.idle_snoozed_until is not None expected_lower = datetime.now(UTC) + timedelta(days=29, hours=23) expected_upper = datetime.now(UTC) + timedelta(days=30, hours=1) actual = updated.idle_snoozed_until if actual.tzinfo is None: actual = actual.replace(tzinfo=UTC) assert expected_lower < actual < expected_upper @pytest.mark.asyncio async def test_snooze_idle_scope_admin_can_snooze_any_scope( db_session: AsyncSession, admin_user, ): """Admins can snooze even when they're not the scope owner — on-behalf-of operations are explicitly supported.""" admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) service = CloudLifecycleService(db_session) updated = await service.snooze_idle_scope( str(scope.id), days=45, actor_user_id=admin.id, actor_role_names=["global_admin"], trace_id="test-snooze-admin", ) await db_session.commit() assert updated.idle_snoozed_until is not None @pytest.mark.asyncio async def test_snooze_idle_scope_non_owner_non_admin_forbidden( db_session: AsyncSession, ): """A regular employee who is neither owner nor cloud-admin gets a 403 from the service. Mirrors the HTTP-layer RBAC gate so direct service callers (MCP / scripts) can't bypass it.""" from app.core.exceptions import ForbiddenError integration = await _make_integration(db_session) owner = await _make_person( db_session, email="not-me@test.com", name="Someone Else", ) other = await _make_person( db_session, email="curious@test.com", name="Curious", ) scope = await _make_scope( db_session, integration, owner_person_id=str(owner.id), ) service = CloudLifecycleService(db_session) with pytest.raises(ForbiddenError): await service.snooze_idle_scope( str(scope.id), days=30, actor_user_id=other.id, actor_role_names=["employee"], trace_id="test-snooze-403", ) @pytest.mark.asyncio async def test_snooze_idle_scope_rejects_closed_scope( db_session: AsyncSession, admin_user, ): """Closed (decommissioned) scopes can't be snoozed — they never appear in the sweep candidate list anyway, so snoozing one is nonsensical state we'd rather refuse than let into the audit log.""" admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) scope.status = SCOPE_STATUS_CLOSED await db_session.commit() service = CloudLifecycleService(db_session) with pytest.raises(AppException) as exc: await service.snooze_idle_scope( str(scope.id), days=30, actor_user_id=admin.id, actor_role_names=["global_admin"], trace_id="test-snooze-closed", ) assert exc.value.code == "IDLE_SNOOZE_SCOPE_NOT_ACTIVE" def test_idle_sweep_default_constants_are_pinned(): """Pin the two idle-sweep tunables so an accidental edit triggers a test failure. The defaults are documented in PLATFORM.md + the user-facing notification copy + frontend hardcoded fallbacks (``snoozeIdle: '忽略 30 天'``); changing either without updating the docs / copy is a silent inconsistency we'd rather catch in CI than discover from a user report.""" assert IDLE_SWEEP_DEDUP_DAYS == 14 assert IDLE_SCOPE_DEFAULT_SNOOZE_DAYS == 30 @pytest.mark.asyncio async def test_snooze_idle_scope_rejects_out_of_range_days( db_session: AsyncSession, admin_user, ): admin, _ = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) service = CloudLifecycleService(db_session) for bad in (0, -1, 366, 10_000): with pytest.raises(AppException) as exc: await service.snooze_idle_scope( str(scope.id), days=bad, actor_user_id=admin.id, actor_role_names=["global_admin"], trace_id=f"test-snooze-range-{bad}", ) assert exc.value.code == "IDLE_SNOOZE_DAYS_OUT_OF_RANGE" # ── Router-level RBAC + integration ─────────────────────────────── @pytest.mark.asyncio async def test_decommission_endpoint_requires_admin( client, db_session: AsyncSession, employee_user, ): _, emp_token = employee_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) resp = await client.post( f"/api/v1/cmdb/scopes/{scope.id}/decommission", json={"reason": "should be rejected"}, headers={"Authorization": f"Bearer {emp_token}"}, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_decommission_endpoint_admin_succeeds_with_real_close( client, db_session: AsyncSession, admin_user, monkeypatch, ): """End-to-end through the route: provider returns ok=True, response payload carries scope + provider_state + revoked_users.""" _, admin_token = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", revoked_users=0, )) _patch_driver(monkeypatch, str(integration.id), driver) resp = await client.post( f"/api/v1/cmdb/scopes/{scope.id}/decommission", json={"reason": "test takedown"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text body = resp.json() assert body["data"]["scope"]["status"] == SCOPE_STATUS_CLOSED assert body["data"]["scope"]["decommission_reason"] == "test takedown" assert body["data"]["scope"]["decommission_provider_state"] == "pending_closure" assert body["data"]["disabled_grants"] == 0 assert body["data"]["provider_state"] == "pending_closure" assert body["data"]["already_closed"] is False @pytest.mark.asyncio async def test_decommission_endpoint_provider_failure_502( client, db_session: AsyncSession, admin_user, monkeypatch, ): _, admin_token = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) driver = StubCloudDriver(next_result=CloseScopeResult( ok=False, error_code="access_denied", error_message="AWS denied organizations:CloseAccount", )) _patch_driver(monkeypatch, str(integration.id), driver) resp = await client.post( f"/api/v1/cmdb/scopes/{scope.id}/decommission", json={"reason": "should fail at cloud"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 502 refreshed = await db_session.get(CmdbCloudScope, scope.id) await db_session.refresh(refreshed) assert refreshed.status == SCOPE_STATUS_ACTIVE @pytest.mark.asyncio async def test_decommission_endpoint_cloud_admin_forbidden( client, db_session: AsyncSession, monkeypatch, ): """``cloud_admin`` is the perpetual-grants marker per migration 080 and CLAUDE.md (NOT a substitute for global_admin). Decommission is irreversible cloud-account closure — only global_admin can fire it. Earlier versions of ``can_manage_cloud_scope_lifecycle`` accidentally widened to cloud_admin; this test pins the narrower behaviour. """ from app.core.security import create_access_token user_id = uuid.uuid4() user = IdentityPerson( id=user_id, work_email="cloud-admin@test.com", display_name="Cloud Admin", employment_status="active", source_system="successfactors", hris_person_id="sf-cloud-001", ) db_session.add(user) await db_session.flush() role_id = (await db_session.execute( select(AuthRole.id).where(AuthRole.name == "cloud_admin"), )).scalar_one() db_session.add(AuthUserRole(user_id=user_id, role_id=role_id)) await db_session.commit() token = create_access_token(str(user_id), ["cloud_admin"]) integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", )) _patch_driver(monkeypatch, str(integration.id), driver) resp = await client.post( f"/api/v1/cmdb/scopes/{scope.id}/decommission", json={"reason": "cloud admin takedown"}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_reactivate_endpoint_is_gone_404( client, admin_user, ): """The reactivate endpoint was removed when decommission became irreversible. Hitting the legacy URL returns 404 — there is no way back from a closed scope. """ _, admin_token = admin_user resp = await client.post( f"/api/v1/cmdb/scopes/{uuid.uuid4()}/reactivate", headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_idle_candidates_endpoint_requires_admin( client, db_session: AsyncSession, employee_user, ): _, emp_token = employee_user resp = await client.get( "/api/v1/cmdb/scopes/idle-candidates", headers={"Authorization": f"Bearer {emp_token}"}, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_idle_candidates_endpoint_returns_list( client, db_session: AsyncSession, admin_user, ): _, admin_token = admin_user integration = await _make_integration(db_session) scope = await _make_scope(db_session, integration) today = date.today() for i in range(30): db_session.add(CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integration.id), scope_id=str(scope.id), date=today - timedelta(days=i), total_cost_usd=Decimal("0"), currency="USD", )) await db_session.commit() resp = await client.get( "/api/v1/cmdb/scopes/idle-candidates", headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200 body = resp.json() ids = [c["scope_id"] for c in body["data"]] assert str(scope.id) in ids @pytest.mark.asyncio async def test_list_scopes_default_excludes_closed( client, db_session: AsyncSession, admin_user, monkeypatch, ): """The default ``GET /cmdb/scopes`` payload hides closed scopes; ``?closed_only=true`` returns ONLY closed; ``?include_closed=true`` returns both. This is the contract the new "已下线" tab depends on. """ admin, admin_token = admin_user integration = await _make_integration(db_session) live = await _make_scope(db_session, integration, external_id="live-1") dead_target = await _make_scope( db_session, integration, external_id="dead-1", ) driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) await service.decommission_scope( str(dead_target.id), reason="test", actor_user_id=admin.id, trace_id="t-list-1", ) await db_session.commit() # Default — only live. resp = await client.get( "/api/v1/cmdb/scopes", headers={"Authorization": f"Bearer {admin_token}"}, ) body = resp.json() ids = [r["id"] for r in body["data"]] assert str(live.id) in ids assert str(dead_target.id) not in ids # closed_only — only dead. resp = await client.get( "/api/v1/cmdb/scopes", params={"closed_only": "true"}, headers={"Authorization": f"Bearer {admin_token}"}, ) body = resp.json() ids = [r["id"] for r in body["data"]] assert str(dead_target.id) in ids assert str(live.id) not in ids # include_closed — both. resp = await client.get( "/api/v1/cmdb/scopes", params={"include_closed": "true"}, headers={"Authorization": f"Bearer {admin_token}"}, ) body = resp.json() ids = [r["id"] for r in body["data"]] assert str(live.id) in ids assert str(dead_target.id) in ids @pytest.mark.asyncio async def test_closed_scope_archive_is_ordered_by_decommissioned_at( client, db_session: AsyncSession, admin_user, monkeypatch, ) -> None: """The 已下线 archive answers "what did I take down, and when". Three things are pinned, and each of them broke the tab in a different way before this endpoint took a ``sort``: * **Newest first by default.** The list was ordered by ``display_name``, so the scopes closed this morning were scattered alphabetically across every page — the one question the tab exists for was the one it could not answer. * **Undated rows sort LAST, in both directions.** Closes predating the ``decommissioned_at`` column (prod carries ~1000) have NULL there, and MySQL/TiDB put NULL first ascending — which would open the ascending view on a wall of blank rows. * **``sort_dir=asc`` actually reverses it**, rather than the arrow being decorative. """ admin, admin_token = admin_user integration = await _make_integration(db_session) # display_name is assigned randomly by _make_scope, so name order # and close order are independent — an accidental fallback to name # ordering cannot coincidentally satisfy the assertions below. driver = StubCloudDriver(next_result=CloseScopeResult( ok=True, provider_state="pending_closure", )) _patch_driver(monkeypatch, str(integration.id), driver) service = CloudLifecycleService(db_session) closed_in_order: list[str] = [] for n in range(3): scope = await _make_scope( db_session, integration, external_id=f"archive-order-{n}", ) await service.decommission_scope( str(scope.id), reason="test", actor_user_id=admin.id, trace_id=f"t-order-{n}", ) # Distinct, strictly increasing close timestamps — the service # stamps ``datetime.now`` and three closes inside one test can # land on the same second. scope.decommissioned_at = datetime( 2026, 9, 2, 2, 10 + n, 0, tzinfo=UTC, ) db_session.add(scope) await db_session.commit() closed_in_order.append(str(scope.id)) # A legacy row: closed, but from before the column existed. undated = await _make_scope( db_session, integration, external_id="archive-order-legacy", ) await service.decommission_scope( str(undated.id), reason="test", actor_user_id=admin.id, trace_id="t-order-legacy", ) undated.decommissioned_at = None db_session.add(undated) await db_session.commit() async def _ids(**params: str) -> list[str]: resp = await client.get( "/api/v1/cmdb/scopes", params={"closed_only": "true", "page_size": "100", **params}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200 ours = set(closed_in_order) | {str(undated.id)} return [r["id"] for r in resp.json()["data"] if r["id"] in ours] default = await _ids() assert default == list(reversed(closed_in_order)) + [str(undated.id)] ascending = await _ids(sort_dir="asc") assert ascending == closed_in_order + [str(undated.id)] # An unknown sort key is rejected at the edge rather than silently # falling back to some other ordering. bad = await client.get( "/api/v1/cmdb/scopes", params={"closed_only": "true", "sort": "budget_usd_monthly"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert bad.status_code == 422 @pytest.mark.asyncio async def test_live_scope_list_stays_name_ascending( client, db_session: AsyncSession, admin_user, ) -> None: """Adding ``sort`` must not re-order the list nobody asked to re-order. The live list passes neither ``sort`` nor ``sort_dir``. A single ``sort_dir`` default shared by both keys would be right for exactly one of them — and the first draft of this endpoint defaulted it to ``desc`` (natural for closures), which silently flipped the main scope list to name-DESCENDING. Nothing would have failed; the list would just have been backwards. """ _admin, admin_token = admin_user integration = await _make_integration(db_session) for name in ("zzz-last", "aaa-first", "mmm-middle"): scope = await _make_scope( db_session, integration, external_id=f"order-{name}", ) scope.display_name = name db_session.add(scope) await db_session.commit() resp = await client.get( "/api/v1/cmdb/scopes", params={"page_size": "100"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200 ours = [ r["display_name"] for r in resp.json()["data"] if r["display_name"] in {"zzz-last", "aaa-first", "mmm-middle"} ] assert ours == ["aaa-first", "mmm-middle", "zzz-last"]