"""Feishu external-options webhook — POST /api/v1/open/feishu/software-options. The software-catalog picker Feishu approval fetches to fill an external-options dropdown. Pins: * token-in-body auth on the ``helix_api_`` credential plane; the ``software.catalog:read`` scope gates it * Feishu envelope: ``{code:0, data:{result:{options, i18nResources, hasMore, nextPageToken}}}`` (not the Helix envelope) * option ``id`` = software name; display carried via ``@i18n@`` + i18nResources for every locale * only ``status='active'`` rows offered; ``query`` fuzzy-filters name * pagination via nextPageToken / hasMore * bad/absent token → 401; valid token missing the scope → 403 """ from __future__ import annotations import uuid import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.api_clients.models import ApiClient from app.api_clients.scopes import ( SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_SOFTWARE_CATALOG_READ, SCOPE_SOFTWARE_PERMISSIONS_READ, ) from app.api_clients.tokens import ApiClientTokenService from app.domains.asset.models import AssetSoftwareCatalog from app.domains.permission.models import SoftwarePermission from app.identity.person_models import IdentityPerson pytestmark = pytest.mark.asyncio URL = "/api/v1/open/feishu/software-options" PERM_URL = "/api/v1/open/feishu/permission-options" async def _mk_person(db: AsyncSession) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=f"co-{uuid.uuid4().hex[:8]}@pingcap.com", display_name="Owner", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:10]}", ) db.add(p) await db.flush() return p async def _mk_client(db: AsyncSession, *, scopes: list[str]) -> str: owner = await _mk_person(db) client = ApiClient( id=uuid.uuid4(), name="Feishu Approval", slug=f"feishu-{uuid.uuid4().hex[:6]}", owner_person_id=owner.id, scopes=scopes, ) db.add(client) await db.flush() await db.refresh(client) _row, plaintext = await ApiClientTokenService(db).mint( client=client, name="t", expires_in_days=None, created_by=None, ) await db.commit() return plaintext async def _mk_software( db: AsyncSession, *, name: str, status: str = "active", version: str | None = None, ) -> AssetSoftwareCatalog: row = AssetSoftwareCatalog( id=uuid.uuid4(), name=name, status=status, version=version, ) db.add(row) await db.flush() await db.commit() return row async def _mk_permission( db: AsyncSession, *, software: AssetSoftwareCatalog, key: str, display: str | None = None, is_active: bool = True, sort_order: int = 0, ) -> SoftwarePermission: row = SoftwarePermission( id=uuid.uuid4(), software_id=software.id, permission_key=key, display_name=display or key, is_active=is_active, sort_order=sort_order, ) db.add(row) await db.flush() await db.commit() return row def _body(token: str, **linkage) -> dict: return {"token": token, "linkage_params": linkage} async def test_returns_active_software_in_feishu_envelope(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_CATALOG_READ]) await _mk_software(db_session, name="Zoom") await _mk_software(db_session, name="Slack") resp = await client.post(URL, json=_body(token)) assert resp.status_code == 200 payload = resp.json() assert payload["code"] == 0 result = payload["data"]["result"] names = {o["id"] for o in result["options"]} assert {"Zoom", "Slack"} <= names assert result["hasMore"] is False assert result["nextPageToken"] == "" # display carried via @i18n@ keys resolved in i18nResources opt = next(o for o in result["options"] if o["id"] == "Zoom") assert opt["value"].startswith("@i18n@") locales = {r["locale"] for r in result["i18nResources"]} assert "zh_cn" in locales and "en_us" in locales for res in result["i18nResources"]: assert res["texts"][opt["value"]] == "Zoom" async def test_excludes_retired_software(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_CATALOG_READ]) await _mk_software(db_session, name="LiveTool") await _mk_software(db_session, name="DeadTool", status="retired") resp = await client.post(URL, json=_body(token)) names = {o["id"] for o in resp.json()["data"]["result"]["options"]} assert "LiveTool" in names assert "DeadTool" not in names async def test_query_filters_by_name(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_CATALOG_READ]) await _mk_software(db_session, name="Figma") await _mk_software(db_session, name="Notion") resp = await client.post(URL, json=_body(token, query="fig")) names = {o["id"] for o in resp.json()["data"]["result"]["options"]} assert names == {"Figma"} async def test_pagination_next_page_token(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_CATALOG_READ]) # more than one page (page size 100) for i in range(105): await _mk_software(db_session, name=f"SW-{i:03d}") first = (await client.post(URL, json=_body(token))).json()["data"]["result"] assert first["hasMore"] is True assert first["nextPageToken"] == "100" assert len(first["options"]) == 100 second = (await client.post( URL, json=_body(token, page_token=first["nextPageToken"]) )).json()["data"]["result"] assert second["hasMore"] is False assert len(second["options"]) == 5 async def test_bad_token_rejected(client, db_session): await _mk_software(db_session, name="Zoom") resp = await client.post(URL, json=_body("helix_api_totallybogustoken00000")) assert resp.status_code == 401 async def test_missing_scope_forbidden(client, db_session): # a valid client but without software.catalog:read token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ]) await _mk_software(db_session, name="Zoom") resp = await client.post(URL, json=_body(token)) assert resp.status_code == 403 # ── permission-options (cascade) ─────────────────────────────────────── async def test_permissions_cascade_by_software_name(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_PERMISSIONS_READ]) zoom = await _mk_software(db_session, name="Zoom") figma = await _mk_software(db_session, name="Figma") await _mk_permission(db_session, software=zoom, key="host", display="Host") await _mk_permission(db_session, software=zoom, key="cohost", display="Co-host") await _mk_permission(db_session, software=figma, key="editor", display="Editor") # linked software passed via the recommended dependent-param key resp = await client.post(PERM_URL, json=_body(token, software_name="Zoom")) assert resp.status_code == 200 result = resp.json()["data"]["result"] ids = {o["id"] for o in result["options"]} assert ids == {"host", "cohost"} # only Zoom's, not Figma's editor # display via i18n opt = next(o for o in result["options"] if o["id"] == "host") for res in result["i18nResources"]: assert res["texts"][opt["value"]] == "Host" async def test_permissions_empty_when_no_software_selected(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_PERMISSIONS_READ]) zoom = await _mk_software(db_session, name="Zoom") await _mk_permission(db_session, software=zoom, key="host") resp = await client.post(PERM_URL, json=_body(token)) assert resp.status_code == 200 assert resp.json()["data"]["result"]["options"] == [] async def test_permissions_fallback_linkage_key(client, db_session): # admin named the dependent param something other than software_name; # a single extra param is treated as the link. token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_PERMISSIONS_READ]) zoom = await _mk_software(db_session, name="Zoom") await _mk_permission(db_session, software=zoom, key="host") resp = await client.post(PERM_URL, json=_body(token, my_sw_field="Zoom")) ids = {o["id"] for o in resp.json()["data"]["result"]["options"]} assert ids == {"host"} async def test_permissions_excludes_inactive(client, db_session): token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_PERMISSIONS_READ]) zoom = await _mk_software(db_session, name="Zoom") await _mk_permission(db_session, software=zoom, key="host") await _mk_permission(db_session, software=zoom, key="legacy", is_active=False) resp = await client.post(PERM_URL, json=_body(token, software_name="Zoom")) ids = {o["id"] for o in resp.json()["data"]["result"]["options"]} assert ids == {"host"} async def test_permissions_missing_scope_forbidden(client, db_session): # software.catalog:read is NOT enough for the permission endpoint token = await _mk_client(db_session, scopes=[SCOPE_SOFTWARE_CATALOG_READ]) zoom = await _mk_software(db_session, name="Zoom") await _mk_permission(db_session, software=zoom, key="host") resp = await client.post(PERM_URL, json=_body(token, software_name="Zoom")) assert resp.status_code == 403