"""MCP-sync: ``list_cloud_scopes`` is the scope_id discovery path (2026-07-30). ``submit_cloud_access_request`` used to tell the client to call ``list_my_cloud_subscriptions`` — a tool that was never registered. An AI client therefore had NO way to discover a ``scope_id``, which made the whole cloud-access self-service flow unreachable over MCP even though the mutating tool worked fine. ``list_cloud_scopes`` is the 1:1 MCP twin of ``GET /cmdb/scopes`` and closes that gap. Two behaviours are pinned here because both are load-bearing for the flow and both are easy to break silently: 1. ``requestable=True`` surfaces active owned non-prod scopes to a caller with no existing membership — without this the picker is empty for exactly the employees who need it — while EXCLUDING production scopes (prod cloud access is granted by the R&D security team outside Helix self-service, and the HTTP twin filters it at the API level so a client can't bypass the picker). 2. The default mode stays membership-scoped for a plain employee, so the new tool is not an org-wide scope-enumeration primitive for anyone holding only ``employee``. """ from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, SCOPE_STATUS_ACTIVE, SCOPE_TYPE_AZURE_RESOURCE_GROUP, CmdbCloudAccessGrant, CmdbCloudScope, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration from app.mcp import context as _mcp_ctx from app.mcp import tools as _tools pytestmark = pytest.mark.asyncio @pytest.fixture() def _patch_factory(db_engine, monkeypatch): from sqlalchemy.ext.asyncio import async_sessionmaker from app.core import database as _db monkeypatch.setattr( _db, "async_session_factory", async_sessionmaker( db_engine, class_=AsyncSession, expire_on_commit=False, ), ) async def _make_integration(db: AsyncSession) -> Integration: integration = Integration( id=uuid.uuid4(), kind="cloud", provider="aws_cloud", name="aws-discovery-test", slug=f"aws-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(integration) await db.commit() return integration async def _make_owner(db: AsyncSession) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=f"scope-owner-{uuid.uuid4().hex[:6]}@test.com", display_name="Scope Owner", employment_status="active", source_system="successfactors", hris_person_id=f"sf-owner-{uuid.uuid4().hex[:6]}", ) db.add(p) await db.commit() return p async def _make_scope( db: AsyncSession, *, integration: Integration, tag: str, environment: str, owner_person_id: str | None, status: str = SCOPE_STATUS_ACTIVE, scope_type: str = "aws_account", ) -> CmdbCloudScope: scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integration.id), scope_type=scope_type, external_id=uuid.uuid4().hex[:16], display_name=tag, status=status, first_seen_at=datetime.now(UTC), last_synced_at=datetime.now(UTC), environment=environment, owner_person_id=owner_person_id, default_grant_ttl_days=30, ) db.add(scope) await db.commit() return scope async def _call(user_id, **kwargs) -> dict: token = _mcp_ctx.mcp_user_id.set(user_id) try: return await _tools.list_cloud_scopes(**kwargs) finally: _mcp_ctx.mcp_user_id.reset(token) async def test_requestable_mode_surfaces_non_prod_and_hides_prod( employee_user, db_session, _patch_factory, ): """The picker an employee needs to get a ``scope_id``: a scope they have no membership on shows up, and the prod scope never does. """ employee, _ = employee_user integration = await _make_integration(db_session) owner = await _make_owner(db_session) # Unique prefix so the assertions don't depend on what other tests # left in the shared test DB. prefix = f"disc-{uuid.uuid4().hex[:8]}" dev = await _make_scope( db_session, integration=integration, tag=f"{prefix}-dev", environment="dev", owner_person_id=str(owner.id), ) prod = await _make_scope( db_session, integration=integration, tag=f"{prefix}-prod", environment="prod", owner_person_id=str(owner.id), ) # An owner-less non-prod scope must NOT be requestable either (no # approval authority to route to, unless it's a GCP billing account). ownerless = await _make_scope( db_session, integration=integration, tag=f"{prefix}-ownerless", environment="dev", owner_person_id=None, ) # Sub-tenant row from the Azure OpenAI auto-grant flow: active, # owned by the requester, env=dev — it satisfies every OTHER # predicate, so only the TENANT_LEVEL_SCOPE_TYPES filter keeps it # out. Its role dropdown would be empty (roles_sync cannot # enumerate roles for an RG) and a grant could only fail at # dispatch with ARM 400 InvalidSubscriptionId. The HTTP twin gained # this filter in #2105; this pins the MCP surface to it. rg = await _make_scope( db_session, integration=integration, tag=f"{prefix}-aoai-rg", environment="dev", owner_person_id=str(owner.id), scope_type=SCOPE_TYPE_AZURE_RESOURCE_GROUP, ) res = await _call(employee.id, requestable=True, search=prefix) ids = {item["scope_id"] for item in res["items"]} assert str(dev.id) in ids, ( "requestable=True must surface an active owned non-prod scope to " "a caller with no existing grant — otherwise the picker is empty " "for exactly the employees who need to request access." ) assert str(prod.id) not in ids, ( "production scopes must never reach the self-service picker." ) assert str(ownerless.id) not in ids assert str(rg.id) not in ids, ( "sub-tenant (azure_resource_group) rows are ungrantable — a " "dead-end picker entry with an empty role dropdown. Keep this " "predicate in lockstep with lifecycle_router.list_scopes." ) # The row carries what submit_cloud_access_request needs next. row = next(i for i in res["items"] if i["scope_id"] == str(dev.id)) assert row["default_grant_ttl_days"] == 30 assert row["owner_person_id"] == str(owner.id) assert row["provider"] == "aws_cloud" assert row["environment"] == "dev" async def test_default_mode_is_membership_scoped_for_employee( employee_user, db_session, _patch_factory, ): """Without ``requestable``, a plain employee sees only scopes they own or hold an active/pending grant on — the new tool must not be an org-wide enumeration primitive. """ employee, _ = employee_user integration = await _make_integration(db_session) owner = await _make_owner(db_session) prefix = f"memb-{uuid.uuid4().hex[:8]}" other = await _make_scope( db_session, integration=integration, tag=f"{prefix}-other", environment="dev", owner_person_id=str(owner.id), ) granted = await _make_scope( db_session, integration=integration, tag=f"{prefix}-granted", environment="dev", owner_person_id=str(owner.id), ) res = await _call(employee.id, search=prefix) assert {i["scope_id"] for i in res["items"]} == set(), ( "employee with no grants must see no scopes in the default mode" ) db_session.add(CmdbCloudAccessGrant( id=uuid.uuid4(), scope_id=str(granted.id), person_id=str(employee.id), role_key="ReadOnlyAccess", role_display_name="ReadOnlyAccess", status=GRANT_STATUS_ACTIVE, granted_at=datetime.now(UTC), )) await db_session.commit() res = await _call(employee.id, search=prefix) ids = {i["scope_id"] for i in res["items"]} assert ids == {str(granted.id)}, ( f"expected only the granted scope, got {ids}" ) assert str(other.id) not in ids row = res["items"][0] assert row["active_member_count"] == 1