#!/usr/bin/env python3 # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Validate the checked-in Control API contract against this checkout.""" import argparse import importlib.util import json from pathlib import Path import re import sys ROOT = Path(__file__).resolve().parents[1] ADMIN_VALIDATOR_PATH = ROOT / 'utils/validate-admin-api.py' MODULE = importlib.util.spec_from_file_location('admin_contract', ADMIN_VALIDATOR_PATH) common = importlib.util.module_from_spec(MODULE) MODULE.loader.exec_module(common) METHODS = common.METHODS def operations(document): return {(path, method) for path, item in document.get('paths', {}).items() for method in METHODS & item.keys()} def core_operations(root): """Extract the built-in v1 route table and expand its wildcard contracts.""" text = (root / 'apisix/control/v1.lua').read_text() route_table = text.rsplit('return {', 1)[1] routes = re.findall( r'\{\s*methods\s*=\s*\{([^}]*)\}\s*,?\s*' r'uris\s*=\s*\{([^}]*)\}', route_table, re.DOTALL, ) method_entries = len(re.findall(r'\bmethods\s*=', route_table)) if len(routes) != method_entries: raise ValueError( f'could not parse {method_entries - len(routes)} Control API route entries' ) expanded = set() wildcard_paths = { '/healthcheck/*': ( '/v1/healthcheck/{src_type}/{src_id}', '/v1/healthcheck/{src_type}/{src_id}/checkers', ), '/route/*': ('/v1/route/{route_id}',), '/service/*': ('/v1/service/{service_id}',), '/upstream/*': ('/v1/upstream/{upstream_id}',), '/plugin_metadata/*': ('/v1/plugin_metadata/{plugin_name}',), } for methods_source, paths_source in routes: methods = re.findall(r'"([A-Z]+)"', methods_source) paths = re.findall(r'"(/[^"]+)"', paths_source) if not methods or not paths: raise ValueError('Control API route entry has no parseable method or URI') for method in methods: for path in paths: for contract_path in wildcard_paths.get(path, ('/v1' + path,)): expanded.add((contract_path, method.lower())) return expanded def optional_operations(root): """Return stable patterns registered by shipped optional components.""" result = set() server_info = (root / 'apisix/plugins/server-info.lua').read_text() if 'function _M.control_api()' in server_info and '"/v1/server_info"' in server_info: result.add(('/v1/server_info', 'get')) discovery_sources = list((root / 'apisix/discovery').glob('*/init.lua')) if any('function _M.dump_data()' in source.read_text() for source in discovery_sources): result.add(('/v1/discovery/{service}/dump', 'get')) if any('function _M.control_api()' in source.read_text() and '"/show_dump_file"' in source.read_text() for source in discovery_sources): result.add(('/v1/discovery/{service}/show_dump_file', 'get')) return result def validate_response_contract(document): """Protect source-verified response details that generic OAS lint cannot infer.""" errors = [] def get(*parts): value = document for part in parts: if not isinstance(value, dict) or part not in value: return None value = value[part] return value schema_example = get('paths', '/v1/schema', 'get', 'responses', '200', 'content', 'application/json', 'examples', 'successful', 'value') or {} schema_example_description = get( 'paths', '/v1/schema', 'get', 'responses', '200', 'content', 'application/json', 'examples', 'successful', 'description') or '' if 'Representative fields' not in schema_example_description: errors.append('schema example must identify its abbreviated main schemas') for resource in ('route', 'upstream'): properties = schema_example.get('main', {}).get(resource, {}).get('properties', {}) if not properties: errors.append(f'schema example main.{resource} must show representative fields') if 'stream_plugins' not in schema_example or 'stream-plugins' in schema_example: errors.append('schema example must use the runtime stream_plugins response key') example_plugin = (schema_example.get('plugins', {}) .get('example-plugin', {})) if set(example_plugin) != {'version', 'priority', 'schema', 'metadata_schema'}: errors.append('schema example must match example-plugin source fields') plugin_schema = example_plugin.get('schema', {}) plugin_properties = plugin_schema.get('properties', {}) if (set(plugin_properties) != {'_meta', 'i', 's', 't', 'ip', 'port'} or plugin_schema.get('required') != ['i'] or plugin_schema.get('$comment') != 'this is a mark for our injected plugin schema'): errors.append('schema example must include the loaded example-plugin schema') plugin_meta = plugin_properties.get('_meta', {}) if (plugin_meta.get('additionalProperties') is not False or set(plugin_meta.get('properties', {})) != { 'disable', 'error_response', 'priority', 'filter', 'pre_function'}): errors.append('schema example must include the injected plugin _meta schema') metadata_schema = example_plugin.get('metadata_schema', {}) if (set(metadata_schema.get('properties', {})) != {'ikey', 'skey'} or metadata_schema.get('required') != ['ikey', 'skey']): errors.append('schema example must include example-plugin metadata fields') mqtt_proxy = schema_example.get('stream_plugins', {}).get('mqtt-proxy', {}) if set(mqtt_proxy) != {'version', 'schema', 'priority'} or mqtt_proxy.get('version') != 0.1: errors.append('schema example must match mqtt-proxy source fields') mqtt_schema = mqtt_proxy.get('schema', {}) mqtt_properties = mqtt_schema.get('properties', {}) if (set(mqtt_properties) != {'_meta', 'protocol_name', 'protocol_level'} or mqtt_schema.get('required') != ['protocol_level'] or mqtt_schema.get('$comment') != 'this is a mark for our injected plugin schema'): errors.append('schema example must include the loaded mqtt-proxy schema') for path in ('/v1/healthcheck', '/v1/healthcheck/{src_type}/{src_id}', '/v1/healthcheck/{src_type}/{src_id}/checkers'): error_schema = get('paths', path, 'get', 'responses', '503', 'content', 'application/json', 'schema', 'properties', 'error_msg') or {} if error_schema.get('pattern') != r'^HTML template rendering: .+$': errors.append(f'GET {path}: HTML error must preserve the runtime prefix') for path in ('/v1/route/{route_id}', '/v1/upstream/{upstream_id}'): example = get('paths', path, 'get', 'responses', '200', 'content', 'application/json', 'examples', '1', 'value') if not isinstance(example, dict) or not isinstance(example.get('value'), dict): errors.append(f'GET {path}: example must preserve the config-entry value envelope') for path in ('/v1/routes', '/v1/services', '/v1/upstreams', '/v1/plugin_metadatas'): schema = get('paths', path, 'get', 'responses', '200', 'content', 'application/json', 'schema') or {} choices = schema.get('oneOf', []) if isinstance(schema, dict) else [] types = {choice.get('type') for choice in choices if isinstance(choice, dict)} empty_object = any(choice.get('type') == 'object' and choice.get('maxProperties') == 0 for choice in choices if isinstance(choice, dict)) if types != {'array', 'object'} or not empty_object: errors.append(f'GET {path}: schema must include the runtime empty-object response') dump_content = get('paths', '/v1/discovery/{service}/show_dump_file', 'get', 'responses', '200', 'content') if not isinstance(dump_content, dict) or set(dump_content) != {'text/plain'}: errors.append('discovery dump-file response must be JSON text with text/plain media type') server_text = ' '.join([ str(get('paths', '/v1/server_info', 'get', 'description') or ''), *(str(tag.get('description', '')) for tag in document.get('tags', []) if tag.get('name') == 'Server Information'), ]) if 'deprecated in APISIX 3.13.0' not in server_text or '3.17.0' in server_text: errors.append('server-info deprecation must identify APISIX 3.13.0') plugin_parameters = get('paths', '/v1/plugin_metadata/{plugin_name}', 'get', 'parameters') or [] plugin_example = next((parameter.get('example') for parameter in plugin_parameters if parameter.get('name') == 'plugin_name'), None) if plugin_example != 'file-logger': errors.append('plugin_name must use a non-empty source-backed example') server_id = get('paths', '/v1/server_info', 'get', 'responses', '200', 'content', 'application/json', 'schema', 'properties', 'id') or {} id_choices = server_id.get('oneOf', []) if isinstance(server_id, dict) else [] if {choice.get('type') for choice in id_choices if isinstance(choice, dict)} != { 'string', 'integer'}: errors.append('server_info.id must allow generated strings and configured integers') reload_response = get('paths', '/v1/plugins/reload', 'put', 'responses', '503') or {} reload_content = reload_response.get('content', {}) reload_description = reload_response.get('description', '').lower() if (set(reload_content) != {'text/plain'} or 'recorded' not in reload_description or 'published' not in reload_description): errors.append('plugin reload 503 must cover both failure paths as text/plain') reload_media = reload_content.get('text/plain', {}) record_example = (reload_media.get('examples', {}) .get('recordFailure', {}).get('value')) if record_example != '{"error_msg":"failed to record plugins reload"}': errors.append('plugin reload 503 must preserve the exact recording-failure body') reload_schema_description = reload_media.get('schema', {}).get('description', '') if 'raw event-system error string' not in reload_schema_description: errors.append('plugin reload 503 must distinguish the raw publication error body') dump_summary = get('paths', '/v1/discovery/{service}/dump', 'get', 'responses', '200', 'content', 'application/json', 'examples', '1', 'summary') if dump_summary != 'Consul Example': errors.append('service discovery dump example must identify its Consul shape') dump_description = get('paths', '/v1/discovery/{service}/dump', 'get', 'description') or '' if 'Response shape varies by module' not in dump_description or 'Consul' not in dump_description: errors.append('service discovery dump must qualify module-specific response shapes') for path in ('/v1/healthcheck', '/v1/healthcheck/{src_type}/{src_id}'): examples = get('paths', path, 'get', 'responses', '200', 'content', 'application/json', 'examples') or {} if '"hostname"' not in json.dumps(examples): errors.append(f'GET {path}: checker node examples must include hostname') expected_errors = { ('/v1/healthcheck/{src_type}/{src_id}', '400'): 'invalid src type route', ('/v1/healthcheck/{src_type}/{src_id}', '404'): 'routes[1] not found', ('/v1/healthcheck/{src_type}/{src_id}/checkers', '400'): 'invalid src type route', ('/v1/healthcheck/{src_type}/{src_id}/checkers', '404'): 'routes[1] not found', ('/v1/route/{route_id}', '404'): 'route[2] not found', ('/v1/service/{service_id}', '404'): 'service[2] not found', ('/v1/upstream/{upstream_id}', '404'): 'upstream[3] not found', ('/v1/plugin_metadata/{plugin_name}', '404'): 'plugin metadata[missing-plugin] not found', } for (path, status), expected in expected_errors.items(): actual = get('paths', path, 'get', 'responses', status, 'content', 'application/json', 'schema', 'properties', 'error_msg', 'example') if actual != expected: errors.append(f'GET {path} {status}: error_msg example must be {expected!r}') return errors def validate_source(document, root): errors = [] expected = core_operations(root) | optional_operations(root) actual = operations(document) if actual != expected: errors.append( f'operation surface differs: missing {sorted(expected - actual)}, ' f'extra {sorted(actual - expected)}' ) description = document.get('info', {}).get('description', '') if 'Enabled plugins and service discovery modules can register additional endpoints' not in description: errors.append('info.description must distinguish configuration-dependent endpoints') if re.search(r'api7|apiseven', json.dumps(document), re.IGNORECASE): errors.append('Control API specification must not contain API7-specific references') if document.get('security') != []: errors.append('Control API must explicitly declare that it has no authentication') servers = document.get('servers', []) if len(servers) != 1 or servers[0].get('url') != 'http://127.0.0.1:9090': errors.append('Control API must document only the default loopback server') conditional = { '/v1/server_info': ('only when', '`server-info` plugin'), '/v1/discovery/{service}/dump': ('only when', 'service discovery module'), '/v1/discovery/{service}/show_dump_file': ('only when', 'service discovery module'), } for path, phrases in conditional.items(): operation = document.get('paths', {}).get(path, {}).get('get', {}) text = operation.get('description', '') for phrase in phrases: if phrase not in text: errors.append(f'GET {path}: description must include {phrase!r}') errors.extend(validate_response_contract(document)) return errors def validate(document, root=ROOT): errors = common.validate(document, common.repository_versions(root)) errors.extend(validate_source(document, root)) return errors def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( 'spec', nargs='?', type=Path, default=ROOT / 'docs/en/latest/control-api.json', ) args = parser.parse_args() document = json.loads(args.spec.read_text()) errors = validate(document) if errors: print('\n'.join(f'- {error}' for error in errors), file=sys.stderr) return 1 operation_count = len(operations(document)) schema_count = len(document.get('components', {}).get('schemas', {})) print(f'Control API contract is valid: {operation_count} operations, {schema_count} schemas') return 0 if __name__ == '__main__': raise SystemExit(main())