From 0f7ee5775ecb7305c79f576faa124398b737b790 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Sat, 31 Oct 2026 10:31:06 +0000 Subject: [PATCH] test(api-tools): add unit tests for operation documentation and exposure metadata LANG-229 Co-authored-by: multica-agent --- tests/unit_tests/api/test_operation_docs.py | 189 ++++++++++++++++++++ 1 file changed, 189 insertions(+) create mode 100644 tests/unit_tests/api/test_operation_docs.py diff --git a/tests/unit_tests/api/test_operation_docs.py b/tests/unit_tests/api/test_operation_docs.py new file mode 100644 index 000000000..2dab50c55 --- /dev/null +++ b/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']