From 2c75d03bcbc039ebcb9d37ea263fdba4e6b38c7b Mon Sep 17 00:00:00 2001
From: huanghuoguoguo <youhuanghe0525@gmail.com>
Date: Sat, 10 Oct 2026 23:53:26 +0800
Subject: [PATCH] test(api): cover versioned application API boundary

---
 tests/unit_tests/api/test_application_api.py | 50 ++++++++++++++++++++
 1 file changed, 50 insertions(+)
 create mode 100644 tests/unit_tests/api/test_application_api.py

diff --git a/tests/unit_tests/api/test_application_api.py b/tests/unit_tests/api/test_application_api.py
new file mode 100644
index 000000000..2c7253105
--- /dev/null
+++ b/tests/unit_tests/api/test_application_api.py
@@ -0,0 +1,50 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from langbot.pkg.api.application import ApplicationAPI
+from langbot.pkg.api.http.authz import Permission
+
+
+def _context(*permissions: str):
+    return SimpleNamespace(
+        workspace=SimpleNamespace(permissions=frozenset(permissions)),
+    )
+
+
+@pytest.mark.asyncio
+async def test_list_pipelines_delegates_with_workspace_authorization():
+    service = SimpleNamespace(get_pipelines=AsyncMock(return_value=[{'uuid': 'p1'}]))
+    api = ApplicationAPI(SimpleNamespace(pipeline_service=service))
+    context = _context(Permission.RESOURCE_VIEW.value)
+
+    result = await api.list_pipelines(context, 'updated_at', 'ASC')
+
+    assert result == [{'uuid': 'p1'}]
+    service.get_pipelines.assert_awaited_once_with(
+        context,
+        'updated_at',
+        'ASC',
+        include_secret=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_pipeline_operation_rejects_missing_permission():
+    service = SimpleNamespace(get_pipeline=AsyncMock())
+    api = ApplicationAPI(SimpleNamespace(pipeline_service=service))
+    context = _context()
+
+    with pytest.raises(PermissionError):
+        await api.get_pipeline(context, 'p1')
+
+    service.get_pipeline.assert_not_awaited()
+
+
+def test_catalog_is_versioned_and_workspace_scoped():
+    catalog = {item['id']: item for item in ApplicationAPI.operation_catalog()}
+
+    assert catalog['pipeline.list']['version'] == 'v1'
+    assert catalog['pipeline.list']['scope'] == 'workspace'
+    assert catalog['pipeline.extensions.update']['side_effect'] is True