You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
352 lines
14 KiB
352 lines
14 KiB
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 = <class 'bool'>''', # 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)
|
|
)
|