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.sep.list_available_sep_and_pools: result type must be child of BaseAPIParamsNestedModel.''', # noqa E501 '''cloudapi.k8s.create: parameter "lb_sysctl_params", target_annotation = dict[str, str], expected_annot = ''', # noqa E501 'cloudbroker.storage_policy.get: result type must be child of BaseAPIParamsNestedModel.', # noqa: E501 'cloudbroker.storage_policy.add_pool: result type must be child of BaseAPIParamsNestedModel.', # noqa: E501 '''cloudapi.lb.create: parameter "sysctl_params", target_annotation = dict[str, str], expected_annot = ''', # noqa E501 'cloudbroker.storage_policy.delete_pool: result type must be child of BaseAPIParamsNestedModel.', # noqa: E501 'cloudbroker.storage_policy.update: result type must be child of BaseAPIParamsNestedModel.', # noqa: E501 'cloudapi.zone.list: result type must be child of BaseAPIParamsNestedModel.', # noqa: E501 'cloudapi.storage_policy.get: result type must be child of BaseAPIParamsNestedModel.', # noqa: E501 '''cloudapi.compute.guest_agent_execute: parameter "args", target_annotation = dict[str, typing.Any], expected_annot = ''', # noqa: E501 '''cloudapi.image.create: API has parameter "asyncMode" but this SDK function doesn't have corresponding parameter.''', # noqa: E501 '''cloudapi.flipgroup.create: API has parameter "clientType" but this SDK function doesn't have corresponding parameter.''', # noqa: E501 '''cloudapi.k8s.create: API has parameter "oidcCertificate" but this SDK function doesn't have corresponding parameter.''', # noqa: E501 'cloudapi.disks.limit_io: API has parameter "iops" 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.compute.net_attach: default value of parameter "mtu" must be 1500.', # noqa: E501 'cloudapi.disks.create: default value of parameter "size_gb" must be 10.', 'cloudapi.kvmx86.create: annotation of parameter "ci_user_data" must contain BaseAPIParamsNestedModel.', # noqa: E501 '''cloudapi.lb.update_sysctl_params: parameter "sysctl_params", target_annotation = dict[str, str], expected_annot = ''', # noqa: E501 '''cloudbroker.compute.start_migration_out: parameter "disk_mapping", target_annotation = dict[str, str], expected_annot = ''', # noqa E501 '''cloudbroker.compute.start_migration_out: parameter "cdrom_mapping", target_annotation = dict[str, str], expected_annot = ''', # noqa E501 '''cloudbroker.compute.start_migration_out: parameter "net_mapping", target_annotation = dict[str, dynamix_sdk.api._nested.params.NetMapConfigAPIParamsNM], expected_annot = ''', # noqa E501 'cloudbroker.user.create: default value of parameter "password" must be strongpassword.', # noqa: E501 ] type_mappings = { 'integer': int, 'string': str, 'boolean': bool, 'object': sdk_types.BaseAPIParamsNestedModel, 'number': float, 'dict': sdk_types.BaseAPIResultModel, 'array': list, } def check_param( *, sdk_func_name: str, sdk_func_params: dict[str, inspect.Parameter], param_name: str, param_def: dict, required: bool, ) -> list[str]: inconsistencies: list[str] = [] api_def_param_enum = param_def.get('enum') if ( api_def_param_enum is not None and not isinstance(api_def_param_enum, list) ): raise TypeError api_def_param_default = param_def.get('default') api_def_param_type = param_def.get('type') if ( api_def_param_type is not None and not isinstance(api_def_param_type, str) ): raise TypeError api_def_param_items = param_def.get('items') if ( api_def_param_items is not None and not isinstance(api_def_param_items, dict) ): raise TypeError necessary_dx_param = required or not ( api_def_param_enum and len(api_def_param_enum) == 1 and api_def_param_default is not None ) if not isinstance(param_name, str): raise TypeError param_exists_in_sdk = param_name in sdk_func_params.keys() if not param_exists_in_sdk: if necessary_dx_param: inconsistencies.append( f'{sdk_func_name}:' f' API has parameter "{param_name}"' f" but this SDK function doesn't have" f' corresponding parameter.' ) return inconsistencies sdk_func_param = sdk_func_params[param_name] sdk_param_name = sdk_func_param.name if not necessary_dx_param: inconsistencies.append( f'{sdk_func_name}:' f' parameter "{sdk_param_name} is unnecessary in SDK.' ) return inconsistencies if sdk_func_param.annotation is Any: inconsistencies.append( f'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must not be Any.' ) return inconsistencies if api_def_param_default and api_def_param_default != -1: if sdk_func_param.default != api_def_param_default: inconsistencies.append( f'{sdk_func_name}:' f' default value of parameter "{sdk_param_name}"' f' must be {api_def_param_default}.' ) target_annotation = sdk_func_param.annotation if not 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 or api_def_param_default == [] ): if sdk_func_param.default is not None: inconsistencies.append( f'{sdk_func_name}:' f' default value of parameter "{sdk_param_name}"' f' must be None.' ) if not isinstance(sdk_func_param.annotation, UnionType): inconsistencies.append( f'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must be Union.' ) return inconsistencies annot_union_args = get_args(sdk_func_param.annotation) if NoneType not in annot_union_args: inconsistencies.append( f'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must contain None.' ) return inconsistencies if not len(annot_union_args) == 2: inconsistencies.append( f'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must contain two types.' ) return inconsistencies 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'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must contain list[...]' ) return inconsistencies 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: return inconsistencies 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: return inconsistencies elif api_def_param_type is not None: expected_annot = type_mappings.get(api_def_param_type) if expected_annot is None: return inconsistencies if ( not inspect.isclass(target_annotation) or isinstance(target_annotation, GenericAlias) ): inconsistencies.append( f'{sdk_func_name}:' f' parameter "{sdk_param_name}",' f' {target_annotation = },' # noqa: E202, E251 f' {expected_annot = }' # noqa: E202, E251 ) return inconsistencies if api_def_param_enum: if not issubclass(target_annotation, Enum): inconsistencies.append( f'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must contain enum.' ) return inconsistencies enum_from_annot = target_annotation enum_values = [e.value for e in enum_from_annot] if sorted(api_def_param_enum) != sorted(enum_values): inconsistencies.append( f'{sdk_func_name}:' f' parameter "{sdk_param_name}":' f' enum {enum_from_annot.__qualname__}' f' must contain these values:' f' {api_def_param_enum}.' ) elif not issubclass(target_annotation, expected_annot): inconsistencies.append( f'{sdk_func_name}:' f' annotation of parameter "{sdk_param_name}"' f' must contain {expected_annot.__qualname__}.' ) return inconsistencies def find_unused_sdk_params( *, sdk_func_name: str, sdk_func_params: dict[str, inspect.Parameter], dx_param_names: set[str], ) -> list[str]: inconsistencies: list[str] = [] unused_sdk_param_names = [ sdk_param.name for alias, sdk_param in sdk_func_params.items() if alias not in dx_param_names ] if unused_sdk_param_names: inconsistencies.append( f'{sdk_func_name}:' f' unused parameters:' f' {", ".join(sorted(unused_sdk_param_names))}' ) return inconsistencies def test_with_api_definition( dx_api_definition: JSON, sdk_dx_functions: tuple[SDKFunction, ...], ): inconsistencies: list[str] = [] for sdk_func in sdk_dx_functions: sdk_func_name = ".".join(sdk_func.call_attrs) 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"{sdk_func_name}:" 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'{sdk_func_name}:' 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 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'{sdk_func_name}:' 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') or [] body_schema = None if not isinstance(api_definition_params, list): raise TypeError for p in api_definition_params: if ( isinstance(p, dict) and p.get('name') == 'body' and 'schema' in p ): body_schema = p['schema'] break dx_param_names = set() param_defs = dict() required_params = set() if isinstance(body_schema, dict) and 'properties' in body_schema: properties = body_schema['properties'] if not isinstance(properties, dict): raise TypeError required = body_schema.get('required', []) if not isinstance(required, list): raise TypeError dx_param_names.update(properties.keys()) param_defs.update(properties) required_params.update(required) else: if not api_definition_params: if sdk_func_params: inconsistencies.append( f'{sdk_func_name}: this function ' f'must not have parameters.' ) continue for p in api_definition_params: if not isinstance(p, dict): raise TypeError param_name = p['name'] if not isinstance(param_name, str): raise TypeError dx_param_names.add(param_name) param_defs[param_name] = p if p.get('required', False): required_params.add(param_name) if not dx_param_names: if sdk_func_params: inconsistencies.append( f"{sdk_func_name}: this function must not have parameters." ) else: inconsistencies.extend( find_unused_sdk_params( sdk_func_name=sdk_func_name, sdk_func_params=sdk_func_params, dx_param_names=dx_param_names, ) ) for param_name, param_def in param_defs.items(): if not isinstance(param_def, dict): raise TypeError inconsistencies.extend( check_param( sdk_func_name=sdk_func_name, sdk_func_params=sdk_func_params, param_name=param_name, param_def=param_def, required=param_name in required_params, ) ) 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) )