Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use isinstance_union method in localstack

Best Python code snippet using localstack_python

container_client.py

Source:container_client.py

...71    def close(self):72        raise NotImplementedError73# defines the type for port mappings (source->target port range)74PortRange = Union[List, HashableList]75def isinstance_union(obj, class_or_tuple):76    # that's some dirty hack77    if sys.version_info < (3, 10):78        return isinstance(obj, get_args(PortRange))79    else:80        return isinstance(obj, class_or_tuple)81class PortMappings:82    """Maps source to target port ranges for Docker port mappings."""83    # bind host to be used for defining port mappings84    bind_host: str85    # maps `from` port range to `to` port range for port mappings86    mappings: Dict[PortRange, List]87    def __init__(self, bind_host: str = None):88        self.bind_host = bind_host if bind_host else ""89        self.mappings = {}90    def add(91        self,92        port: Union[int, PortRange],93        mapped: Union[int, PortRange] = None,94        protocol: str = "tcp",95    ):96        mapped = mapped or port97        if isinstance_union(port, PortRange):98            for i in range(port[1] - port[0] + 1):99                if isinstance_union(mapped, PortRange):100                    self.add(port[0] + i, mapped[0] + i)101                else:102                    self.add(port[0] + i, mapped)103            return104        if port is None or int(port) <= 0:105            raise Exception(f"Unable to add mapping for invalid port: {port}")106        if self.contains(port):107            return108        bisected_host_port = None109        for from_range, to_range in dict(self.mappings).items():110            if not self.in_expanded_range(port, from_range):111                continue112            if not self.in_expanded_range(mapped, to_range):113                continue...

swagger_documentation.py

Source:swagger_documentation.py

...96TEMPLATE__url = "url: '%s'"97TEMPLATE__spec = "spec: "98DESCRIPTION__security = "A JWT token fetched from /oauth/token"99DOCUMENTATION__allow_all = "ALLOW_ALL"  # Set as the name to allow all input/output100def isinstance_union(arg, check: Union) -> bool:101    return any([isinstance(arg, cls) for cls in check.__args__])102def force_list(potential_list: Union[List[T], T]) -> List[T]:103    return potential_list if isinstance(potential_list, list) else ([] if potential_list is None else [potential_list])104class SwaggerException(Exception):105    def __init__(self, *args, **kwargs):106        super().__init__(args, kwargs)107class SwaggerResponseException:108    def __init__(self, code: HTTPStatus = RESP__default_err_code, message: str = RESP__default_err_message):109        self.code = code110        self.message = message111TYPE__exceptions = Optional[Union[List[SwaggerResponseException], SwaggerResponseException]]112class SwaggerSimpleList:113    def __init__(self, arg_type: type, description: str, example: Optional[TYPE__example] = None,114                 required: bool = False, condition: str = None):115        self.arg_type = arg_type116        self.description = description117        self.example = example118        self.required = required119        self.condition = condition120        if not required and condition is None:121            raise SwaggerException(ERR__condition_needed % "simple list")122        self._validate_example(arg_type, self.example)123    @staticmethod124    def _validate_example(arg_type: type, example: Optional[TYPE__example] = None):125        if example is None:126            raise SwaggerException(ERR__null_example % "simple list")127        if len(example) == 0:128            raise SwaggerException(ERR__empty_example_list % "simple list")129        for ex, ex_id in zip(example, range(len(example))):130            if not isinstance(ex, arg_type):131                raise SwaggerException(ERR__example_type % (str(type(ex)), ex_id + 1, "simple list", str(arg_type)))132        if len(example) != len(set(example)):133            raise SwaggerException(ERR__duplicated_example % "simple list")134class SwaggerArgumentResponse:135    def __init__(self, name: str, description: str,136                 arg_type: Union[type, List['SwaggerArgumentResponse'], 'SwaggerList', SwaggerSimpleList],137                 example: Optional[TYPE__example] = None, required: bool = True, condition: str = None,138                 local_only: bool = False, strip: bool = True, lower: bool = False):139        self.name = name140        self.description = description141        if arg_type == bool and example is None:142            example = [True, False]143        self.example = force_list(example)144        self.required = required145        self.arg_type = arg_type146        self.condition = condition147        self.local_only = local_only148        self.strip = strip149        self.lower = lower150        if not required and condition is None and name != DOCUMENTATION__allow_all:151            raise SwaggerException(ERR__condition_needed % name)152        if required and condition is not None:153            raise SwaggerException(ERR__condition_not_needed % name)154        arg_type_sub_list = isinstance(arg_type, list)155        if arg_type_sub_list:156            arg_type_sub_list = all([isinstance(x, SwaggerArgumentResponse) for x in arg_type])157        if not arg_type_sub_list and not isinstance_union(arg_type, TYPE__swagger_list_like) and not self.is_arg_all():158            self._validate_example(name, arg_type, self.example)159    @staticmethod160    def _validate_example(name: str, arg_type: type, example: Optional[TYPE__example] = None):161        if example is None:162            raise SwaggerException(ERR__null_example % name)163        if len(example) == 0:164            raise SwaggerException(ERR__empty_example_list % name)165        for ex, ex_id in zip(example, range(len(example))):166            if not isinstance(ex, arg_type):167                raise SwaggerException(ERR__example_type % (str(type(ex)), ex_id + 1, name, str(arg_type)))168        if len(example) != len(set(example)):169            raise SwaggerException(ERR__duplicated_example % name)170    def is_arg_all(self):171        if isinstance(self.arg_type, SwaggerArgumentResponse):172            if self.arg_type.name == DOCUMENTATION__allow_all:173                return True174        return self.name == DOCUMENTATION__allow_all175class SwaggerList:176    def __init__(self, *args: Union[SwaggerArgumentResponse, 'SwaggerList']):177        self.responses = list(args)178TYPE__swagger_list_like = Union[SwaggerSimpleList, SwaggerList]179ARG_RESP__allow_all = SwaggerArgumentResponse(DOCUMENTATION__allow_all, DOCUMENTATION__allow_all, str)180BODY__file = "JAAQL_IS_FILE"181TYPE__argument_response = Optional[Union[List[SwaggerArgumentResponse], SwaggerArgumentResponse, SwaggerList,182                                         SwaggerSimpleList]]183TYPE__flat_argument_response = Optional[Union[List[SwaggerArgumentResponse], SwaggerArgumentResponse]]184TYPE__listed_argument_response = Union[List[SwaggerArgumentResponse], SwaggerList, SwaggerSimpleList]185def validate_argument_responses(arg_responses: TYPE__argument_response):186    if isinstance(arg_responses, SwaggerList):187        arg_responses = arg_responses.responses188    if arg_responses is not None and isinstance(arg_responses, list):189        found_names = []190        for arg_resp in arg_responses:191            if isinstance(arg_resp, SwaggerList):192                validate_argument_responses(arg_resp.responses)193            else:194                try:195                    is_list = isinstance(arg_resp.arg_type, SwaggerList)196                    is_resp = isinstance(arg_resp.arg_type, SwaggerArgumentResponse)197                    is_resp_list = isinstance(arg_resp.arg_type, List)198                    if is_list or is_resp or is_resp_list:199                        validate_argument_responses(arg_resp.arg_type)200                    if arg_resp.name in found_names:201                        raise SwaggerException(ERR__duplicated_argument_response_name % arg_resp.name)202                    found_names.append(arg_resp.name)203                except:204                    pass205class SwaggerFlatResponse:206    def __init__(self, description: str = "HTTP OK", code: Union[HTTPStatus, int] = HTTPStatus.OK, resp_type: type = str,207                 body: str = RESPONSE__OK):208        self.code = code209        self.description = description210        self.body = body211        self.resp_type = resp_type212        if not isinstance(body, resp_type):213            raise SwaggerException(ERR__response_type % (code, str(resp_type)))214MOCK__description = ""215class SwaggerResponse:216    def __init__(self, description: str, code: HTTPStatus = HTTPStatus.OK, response: TYPE__argument_response = None,217                 error_on_unexpected_field: bool = False):218        self.code = code219        self.description = description220        self.error_on_unexpected_field = error_on_unexpected_field221        validate_argument_responses(response)222        self.responses = response if isinstance_union(response, TYPE__swagger_list_like) else force_list(response)223RES__allow_all = SwaggerResponse(224    description="Allows all",225    response=ARG_RESP__allow_all226)227TYPE__response = Union[SwaggerFlatResponse, SwaggerResponse]228TYPE__responses = Optional[Union[TYPE__response, List[TYPE__response]]]229class SwaggerMethod:230    def __init__(self, name: str, description: str, method: str, arguments: TYPE__flat_argument_response = None,231                 body: TYPE__argument_response = None, response: TYPE__responses = None,232                 exceptions: TYPE__exceptions = None):233        self._validate_exceptions(exceptions)234        self._validate_responses(response)235        validate_argument_responses(arguments)236        validate_argument_responses(body)237        self.name = name238        self.description = description239        self.method = method240        self.responses = response if isinstance_union(response, TYPE__swagger_list_like) else force_list(response)241        self.arguments = force_list(arguments)242        self._validate_arguments(arguments)243        self.body = force_list(body)244        self.exceptions = force_list(exceptions)245        found_500 = any([_http_status_to_integer(ex.code) == RESP__default_err_code for ex in self.exceptions])246        if not found_500:247            self.exceptions.append(SwaggerResponseException())248        if self.method == REST__DELETE and len(self.body) != 0:249            raise Exception(ERR__delete_not_allowed_body)250        if self.method == REST__GET and len(self.body) != 0:251            raise Exception(ERR__get_not_allowed_body)252    @staticmethod253    def _validate_exceptions(exceptions: TYPE__exceptions):254        if exceptions is not None and isinstance(exceptions, list):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.