from enum import Enum import inspect from types import GenericAlias, NoneType, UnionType from typing import Any, get_args from warnings import warn from dynamix_sdk import types as sdk_types from dynamix_sdk.base import ( get_alias, name_mapping_dict, ) from dynamix_sdk.utils import JSON, get_nested_value from tests.conftest import SDKFunction expected_inconsistencies: list[str] = [ 'cloudapi.disks.limit_io: API has parameter "iops" but this SDK function doesn\'t have corresponding parameter.', # noqa: E501 'cloudapi.image.list: API has parameter "architecture" but this SDK function doesn\'t have corresponding parameter.', # noqa: E501 'cloudapi.bservice.group_stop: default value of parameter "force" must be None.', # noqa: E501 'cloudapi.bservice.group_stop: annotation of parameter "force" must be Union.', # noqa: E501 'cloudapi.bservice.delete: default value of parameter "permanently" must be None.', # noqa: E501 'cloudapi.bservice.delete: annotation of parameter "permanently" must be Union.', # noqa: E501 'cloudapi.compute.net_attach: result type must be child of BaseAPIResultBool.', # noqa: E501 'cloudapi.compute.net_attach: default value of parameter "mtu" must be 1500.', # noqa: E501 '''cloudapi.compute.update: parameter "cpu_pin", target_annotation = None | bool, expected_annot = ''', # noqa: E501 'cloudapi.disks.create: default value of parameter "size_gb" must be 10.', '''cloudapi.disks.create: API has parameter "ssdSize" but this SDK function doesn't have corresponding parameter.''', # noqa: E501 'cloudbroker.account.create: annotation of parameter "uniq_pools" must contain int.', # noqa: E501 'cloudapi.kvmx86.create: annotation of parameter "ci_user_data" must contain BaseAPIParamsNestedModel.', # noqa: E501 '''cloudapi.image.create: API has parameter "architecture" but this SDK function doesn't have corresponding parameter.''', # noqa: E501 'cloudapi.lb.create: annotation of parameter "sysctl_params" must contain BaseAPIParamsNestedModel.', # noqa: E501 ] def test_with_api_definition( dx_api_definition: JSON, sdk_dx_functions: tuple[SDKFunction, ...], ): inconsistencies: list[str] = [] for sdk_func in sdk_dx_functions: func_full_api_path = f'/restmachine{sdk_func.url_path}' if not isinstance(dx_api_definition, dict): raise TypeError dx_api_definition_paths = dx_api_definition['paths'] if not isinstance(dx_api_definition_paths, dict): raise TypeError if func_full_api_path not in dx_api_definition_paths: inconsistencies.append( f"{'.'.join(sdk_func.call_attrs)}:" f' url path {sdk_func.url_path} not found in' f' DX API definition.' ) continue func_api_definition = dx_api_definition_paths[func_full_api_path] if not isinstance(func_api_definition, dict): raise TypeError http_method_lower = sdk_func.http_method.lower() if http_method_lower not in func_api_definition: api_def_http_methods = [ m.upper() for m in func_api_definition.keys() ] inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' HTTP method {sdk_func.http_method} not found.' f' DX API definition for {sdk_func.url_path} has' f' only these HTTP methods:' f" {', '.join(api_def_http_methods)}." ) continue func_api_def_method = func_api_definition[http_method_lower] if not isinstance(func_api_def_method, dict): raise TypeError type_mappings = { 'integer': int, 'string': str, 'boolean': bool, 'object': sdk_types.BaseAPIParamsNestedModel, 'number': float, } if api_def_result_type := get_nested_value( d=func_api_def_method, keys=('responses', '200', 'schema', 'type'), ): expected_result_type = type_mappings[api_def_result_type] if expected_result_type is bool: expected_result_type = sdk_types.BaseAPIResultBool if not issubclass(sdk_func.result_cls, expected_result_type): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' result type must be child of' f' {expected_result_type.__qualname__}.' ) sdk_func_params: dict[str, inspect.Parameter] = {} for p in inspect.signature(sdk_func.proto_method).parameters.values(): if p.name == 'self': continue alias = get_alias( field_name=p.name, model_cls=sdk_func.params_model_cls, name_mapping_dict=name_mapping_dict, ) sdk_func_params[alias] = p api_definition_params = func_api_def_method.get('parameters') if not api_definition_params: if sdk_func_params: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' this function must not have parameters.' ) continue if not isinstance(api_definition_params, list): raise TypeError dx_param_names = set() for p in api_definition_params: if not isinstance(p, dict): raise TypeError dx_param_names.add(p['name']) if len(sdk_func_params) > len(dx_param_names): unused_sdk_param_names = [] for sdk_param_alias, sdk_param in sdk_func_params.items(): if sdk_param_alias not in dx_param_names: unused_sdk_param_names.append(sdk_param.name) if unused_sdk_param_names: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' unused parameters:' f' {", ".join(sorted(unused_sdk_param_names))}' ) for api_def_param in api_definition_params: if not isinstance(api_def_param, dict): raise TypeError api_def_param_enum = api_def_param.get('enum') if ( api_def_param_enum is not None and not isinstance(api_def_param_enum, list) ): raise TypeError api_def_param_default = api_def_param.get('default') api_def_param_type = api_def_param.get('type') if ( api_def_param_type is not None and not isinstance(api_def_param_type, str) ): raise TypeError api_def_param_items = api_def_param.get('items') if ( api_def_param_items is not None and not isinstance(api_def_param_items, dict) ): raise TypeError necessary_dx_param = not ( api_def_param_enum and len(api_def_param_enum) == 1 and api_def_param_default is not None ) sdk_param_alias = api_def_param['name'] if not isinstance(sdk_param_alias, str): raise TypeError param_exists_in_sdk = sdk_param_alias in sdk_func_params.keys() if not param_exists_in_sdk: if necessary_dx_param: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' API has parameter "{sdk_param_alias}"' f" but this SDK function doesn't have" f' corresponding parameter.' ) continue sdk_param_name = sdk_func_params[sdk_param_alias].name if not necessary_dx_param: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' parameter "{sdk_param_name} is unnecessary in SDK.' ) continue sdk_func_param = sdk_func_params[sdk_param_alias] if sdk_func_param.annotation is Any: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must not be Any.' ) continue if api_def_param_default and api_def_param_default != -1: if sdk_func_param.default != api_def_param_default: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' default value of parameter "{sdk_param_name}"' f' must be {api_def_param_default}.' ) target_annotation = sdk_func_param.annotation if not api_def_param.get('required') and ( api_def_param_default is None or ( api_def_param_default == 0 and api_def_param_default is not False ) or api_def_param_default == '' or api_def_param_default == -1 ): if sdk_func_param.default is not None: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' default value of parameter "{sdk_param_name}"' f' must be None.' ) if not isinstance(sdk_func_param.annotation, UnionType): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must be Union.' ) continue annot_union_args = get_args(sdk_func_param.annotation) if NoneType not in annot_union_args: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must contain None.' ) continue if not len(annot_union_args) == 2: inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must contain two types.' ) continue if annot_union_args[0] is NoneType: target_annotation = annot_union_args[1] else: target_annotation = annot_union_args[0] if api_def_param_type == 'array': if ( not isinstance(target_annotation, GenericAlias) or target_annotation.__origin__ is not list ): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must contain list[...].' ) continue if api_def_param_items is not None: api_def_param_items_type = api_def_param_items.get('type') if api_def_param_items_type is None: continue if not isinstance(api_def_param_items_type, str): raise TypeError target_annotation = get_args(target_annotation)[0] expected_annot = type_mappings.get( api_def_param_items_type ) if expected_annot is None: continue elif api_def_param_type is not None: expected_annot = type_mappings.get(api_def_param_type) if expected_annot is None: continue if not inspect.isclass(target_annotation): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' parameter "{sdk_param_name}",' f' {target_annotation = },' f' {expected_annot = }' ) continue if api_def_param_enum: if not issubclass(target_annotation, Enum): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must contain enum.' ) continue enum_from_annot = target_annotation enum_values = enum_from_annot.__members__.values() if sorted(api_def_param_enum) != sorted(enum_values): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' parameter "{sdk_param_name}":' f' enum {enum_from_annot.__qualname__}' f' must contain these values: {api_def_param_enum}.' ) elif not issubclass(target_annotation, expected_annot): inconsistencies.append( f'{".".join(sdk_func.call_attrs)}:' f' annotation of parameter "{sdk_param_name}"' f' must contain {expected_annot.__qualname__}.' ) continue for expected_inconsistency in expected_inconsistencies: if expected_inconsistency in inconsistencies: inconsistencies.remove(expected_inconsistency) else: inconsistencies.append( 'This expected inconsistency not found:' f' {expected_inconsistency}' ) if expected_inconsistencies: warn( '\nExpected inconsistencies:\n' + '\n'.join(expected_inconsistencies) ) assert not inconsistencies, ( f'found {len(inconsistencies)} inconsistencies:\n' + '\n'.join(inconsistencies) )