test(api-tools): add unit tests for operation documentation and exposure metadata

LANG-229

Co-authored-by: multica-agent <[email protected]>

tests/unit_tests/api/test_operation_docs.py

@@ -0,0 +1,189 @@
+"""Unit tests for ApplicationAPI operation documentation and exposure metadata."""
+
+from __future__ import annotations
+
+import pytest
+
+from langbot.pkg.api.application.contracts import Operation
+from langbot.pkg.api.application.operation_docs import DOMAINS, OPERATION_DOCS, OperationDoc, get_operation_doc
+from langbot.pkg.api.application.service_operations import operations as service_operations
+from langbot.pkg.api.application.auxiliary import operations as auxiliary_operations
+from langbot.pkg.api.application.plugin_operations import operations as plugin_operations
+from langbot.pkg.api.application.platform_setup import operations as platform_operations
+from langbot.pkg.api.application.connection_operations import operations as connection_operations
+from langbot.pkg.api.application.workspace_streams import operations as workspace_streams
+from langbot.pkg.api.context import PrincipalType
+
+
+@pytest.fixture(scope='module')
+def all_application_operations() -> list[Operation]:
+    return [
+        *service_operations(),
+        *auxiliary_operations(),
+        *plugin_operations(),
+        *platform_operations(),
+        *connection_operations(),
+        *workspace_streams(),
+    ]
+
+
+@pytest.fixture(scope='module')
+def operations_by_id(all_application_operations: list[Operation]) -> dict[str, Operation]:
+    return {op.operation_id: op for op in all_application_operations}
+
+
+@pytest.fixture(scope='module')
+def in_scope_operation_ids(operations_by_id: dict[str, Operation]) -> set[str]:
+    in_scope = set()
+    for op_id in operations_by_id:
+        if op_id.startswith('monitoring.export.'):
+            continue
+        if (
+            op_id.startswith('monitoring.')
+            or op_id == 'stats.basic'
+            or op_id.startswith('bot.')
+            or op_id.startswith('adapter.')
+            or op_id.startswith('agent.')
+            or op_id.startswith('pipeline.')
+            or op_id.startswith('knowledge.')
+            or op_id.startswith('plugin.')
+            or op_id.startswith('mcp.')
+            or op_id.startswith('skill.')
+            or op_id == 'extension.list'
+            or op_id.startswith('tool.')
+            or op_id.startswith('model.')
+            or op_id.startswith('provider.')
+            or op_id.startswith('requester.')
+        ):
+            in_scope.add(op_id)
+    return in_scope
+
+
+def test_in_scope_operations_coverage(in_scope_operation_ids: set[str], operations_by_id: dict[str, Operation]):
+    assert len(in_scope_operation_ids) == 162
+    assert len(OPERATION_DOCS) == 162
+
+    # Check all in-scope operations are documented
+    missing = in_scope_operation_ids - set(OPERATION_DOCS.keys())
+    assert not missing, f'Missing documentation for operations: {missing}'
+
+    # Check there are no extra operation IDs in OPERATION_DOCS
+    extra = set(OPERATION_DOCS.keys()) - in_scope_operation_ids
+    assert not extra, f'Extra undocumented or out-of-scope operation IDs in OPERATION_DOCS: {extra}'
+
+    # Check all documented IDs actually exist in ApplicationAPI operations
+    unknown = set(OPERATION_DOCS.keys()) - set(operations_by_id.keys())
+    assert not unknown, f'Unknown operation IDs in OPERATION_DOCS: {unknown}'
+
+
+def test_out_of_scope_operations_are_none(
+    in_scope_operation_ids: set[str], all_application_operations: list[Operation]
+):
+    out_of_scope_ops = [op for op in all_application_operations if op.operation_id not in in_scope_operation_ids]
+    assert len(out_of_scope_ops) == 72
+
+    for op in out_of_scope_ops:
+        doc = get_operation_doc(op.operation_id, side_effect=op.side_effect)
+        assert doc.exposure == 'none', f"Out of scope operation {op.operation_id} must have exposure='none'"
+        as_dict = op.as_dict()
+        assert as_dict['exposure'] == 'none', (
+            f"Out of scope operation {op.operation_id} as_dict must have exposure='none'"
+        )
+
+
+def test_summary_and_description_constraints():
+    for op_id, doc in OPERATION_DOCS.items():
+        assert isinstance(doc, OperationDoc)
+        assert doc.summary, f'{op_id} summary must not be empty'
+        assert len(doc.summary) <= 120, f"{op_id} summary exceeds 120 chars: {len(doc.summary)} ('{doc.summary}')"
+        assert doc.description, f'{op_id} description must not be empty'
+        assert len(doc.description) > 10, f'{op_id} description too short'
+        assert 'zh_Hans' in doc.label and doc.label['zh_Hans'], f'{op_id} missing zh_Hans label'
+        assert 'en_US' in doc.label and doc.label['en_US'], f'{op_id} missing en_US label'
+        assert doc.risk in {'read', 'write', 'dangerous'}, f'{op_id} invalid risk: {doc.risk}'
+        assert doc.exposure in {'none', 'assistant', 'all'}, f'{op_id} invalid exposure: {doc.exposure}'
+
+
+def test_side_effect_risk_consistency(all_application_operations: list[Operation]):
+    for op in all_application_operations:
+        doc = get_operation_doc(op.operation_id, side_effect=op.side_effect)
+        if op.side_effect:
+            assert doc.risk in {'write', 'dangerous'}, f'{op.operation_id} has side_effect=True but risk is {doc.risk}'
+        else:
+            assert doc.risk == 'read', f'{op.operation_id} has side_effect=False but risk is {doc.risk}'
+
+
+def test_restricted_exposure_operations(all_application_operations: list[Operation]):
+    for op in all_application_operations:
+        doc = get_operation_doc(op.operation_id, side_effect=op.side_effect)
+        op_id = op.operation_id
+
+        if op.streaming:
+            assert doc.exposure == 'none', f"Streaming operation {op_id} must have exposure='none'"
+
+        if (
+            op_id.startswith('assistant.')
+            or op_id.startswith('api_key.')
+            or op_id.startswith('workspace.member.')
+            or op_id.startswith('workspace.invitation.')
+            or op_id.startswith('provider.codex.')
+        ):
+            assert doc.exposure == 'none', f"Operation {op_id} must have exposure='none'"
+
+        if op_id.startswith('pipeline.migration.'):
+            assert doc.exposure == 'assistant', f"Operation {op_id} must have exposure='assistant'"
+
+
+def test_principals_exposure_consistency(all_application_operations: list[Operation]):
+    for op in all_application_operations:
+        doc = get_operation_doc(op.operation_id, side_effect=op.side_effect)
+        if op.principals is not None and PrincipalType.API_KEY not in op.principals:
+            assert doc.exposure != 'all', (
+                f"Operation {op.operation_id} without API_KEY principal cannot be exposure='all'"
+            )
+
+
+def test_domains_metadata(all_application_operations: list[Operation]):
+    for op in all_application_operations:
+        doc = get_operation_doc(op.operation_id, side_effect=op.side_effect)
+        assert doc.domain in DOMAINS, f'Domain {doc.domain} of operation {op.operation_id} not in DOMAINS'
+
+    for domain_name, meta in DOMAINS.items():
+        assert 'label' in meta and 'zh_Hans' in meta['label'] and 'en_US' in meta['label']
+        assert 'summary' in meta and meta['summary']
+
+
+def test_as_dict_incorporates_metadata(all_application_operations: list[Operation]):
+    for op in all_application_operations:
+        data = op.as_dict()
+        assert 'domain' in data
+        assert 'label' in data
+        assert 'summary' in data
+        assert 'description' in data
+        assert 'risk' in data
+        assert 'exposure' in data
+
+        doc = get_operation_doc(op.operation_id, side_effect=op.side_effect)
+        assert data['domain'] == doc.domain
+        assert data['label'] == doc.label
+        assert data['summary'] == doc.summary
+        assert data['description'] == doc.description
+        assert data['risk'] == doc.risk
+        assert data['exposure'] == doc.exposure
+
+
+def test_parameter_descriptions_in_service_contracts(operations_by_id: dict[str, Operation]):
+    # Verify parameter descriptions on service operations
+    bot_get = operations_by_id['bot.get']
+    schema = bot_get.input_type.model_json_schema()
+    assert 'UUID of the bot' in schema['properties']['bot_uuid']['description']
+    assert 'Whether to include secret values' in schema['properties']['include_secret']['description']
+
+    pipeline_get = operations_by_id['pipeline.get']
+    pipeline_schema = pipeline_get.input_type.model_json_schema()
+    assert 'pipeline.list' in pipeline_schema['properties']['pipeline_uuid']['description']
+
+    monitoring_errors = operations_by_id['monitoring.errors']
+    mon_schema = monitoring_errors.input_type.model_json_schema()
+    assert 'Maximum number of items' in mon_schema['properties']['limit']['description']
+    assert 'Number of items to skip' in mon_schema['properties']['offset']['description']