KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _create_service_map method in localstack

Best Python code snippet using localstack_python

op_router.py

Source:op_router.py

...173    result = arg_key174    for original, substitution in _rule_replacements.items():175        result = result.replace(substitution, original)176    return result177def _create_service_map(service: ServiceModel) -> Map:178    """179    Creates a Werkzeug Map object with all rules necessary for the specific service.180    :param service: botocore service model to create the rules for181    :return: a Map instance which is used to perform the in-service operation routing182    """183    ops = [service.operation_model(op_name) for op_name in service.operation_names]184    rules = []185    # group all operations by their path and method186    path_index: Dict[(str, str), List[_HttpOperation]] = defaultdict(list)187    for op in ops:188        http_op = _HttpOperation.from_operation(op)189        path_index[(http_op.path, http_op.method)].append(http_op)190    # create a matching rule for each (path, method) combination191    for (path, method), ops in path_index.items():192        # translate the requestUri to a Werkzeug rule string193        rule_string = _path_param_regex.sub(_transform_path_params_to_rule_vars, path)194        if len(ops) == 1:195            # if there is only a single operation for a (path, method) combination,196            # the default Werkzeug rule can be used directly (this is the case for most rules)197            op = ops[0]198            rules.append(_StrictMethodRule(string=rule_string, method=method, endpoint=op.operation))  # type: ignore199        else:200            # if there is an ambiguity with only the (path, method) combination,201            # a custom rule - which can use additional request metadata - needs to be used202            rules.append(_RequestMatchingRule(string=rule_string, method=method, operations=ops))203    return Map(204        rules=rules,205        strict_slashes=False,206        merge_slashes=False,207        converters={"path": GreedyPathConverter},208    )209class RestServiceOperationRouter:210    """211    A router implementation which abstracts the (quite complex) routing of incoming HTTP requests to a specific212    operation within a "REST" service (rest-xml, rest-json).213    """214    _map: Map215    def __init__(self, service: ServiceModel):216        self._map = _create_service_map(service)217    def match(self, request: Request) -> Tuple[OperationModel, Mapping[str, Any]]:218        """219        Matches the given request to the operation it targets (or raises an exception if no operation matches).220        :param request: The request of which the targeting operation needs to be found221        :return: A tuple with the matched operation and the (already parsed) path params222        :raises: Werkzeug's NotFound exception in case the given request does not match any operation223        """224        # bind the map to get the actual matcher225        matcher: MapAdapter = self._map.bind(request.host)226        # perform the matching227        rule, args = matcher.match(get_raw_path(request), method=request.method, return_rule=True)228        # if the found rule is a _RequestMatchingRule, the multi rule matching needs to be invoked to perform the229        # fine-grained matching based on the whole request230        if isinstance(rule, _RequestMatchingRule):...

router.py

Source:router.py

...131    # replace any greedy URI params (f.e. /foo/{Bar+}) with the werkzeug notation (/foo/{path:Bar})132    rule_string = _greedy_regex.sub(r"{path:\g<1>}", request_uri_path)133    # replace the spec param notation (f.e. /foo/{Bar}) with the werkzeug path param notation (/foo/<Bar>)134    return rule_string.replace("{", "<").replace("}", ">")135def _create_service_map(service: ServiceModel) -> Map:136    """137    Creates a Werkzeug Map object with all rules necessary for the specific service.138    :param service: botocore service model to create the rules for139    :return: a Map instance which is used to perform the in-service operation routing140             -141    """142    ops = [service.operation_model(op_name) for op_name in service.operation_names]143    rules = []144    # group all operations by their path and method145    path_index: Dict[(str, str), List[_HttpOperation]] = defaultdict(list)146    for op in ops:147        http_op = _HttpOperation.from_operation(op)148        path_index[(http_op.path, http_op.method)].append(http_op)149    # create a matching rule for each (path, method) combination150    for (path, method), ops in path_index.items():151        # translate the requestUri to a Werkzeug rule string152        rule_string = _request_uri_path_to_rule_string(path)153        if len(ops) == 1:154            # if there is only a single operation for a (path, method) combination,155            # the default Werkzeug rule can be used directly (this is the case for most rules)156            op = ops[0]157            rules.append(Rule(rule_string, methods=[method], endpoint=op.operation))  # type: ignore158        else:159            # if there is an ambiguity with only the (path, method) combination,160            # a custom rule - which can use additional request metadata - needs to be used161            rules.append(_RequestMatchingRule(rule_string, methods=[method], operations=ops))162    return Map(rules=rules)163class RestServiceOperationRouter:164    """165    A router implementation which abstracts the (quite complex) routing of incoming HTTP requests to a specific166    operation within a "REST" service (rest-xml, rest-json).167    """168    _map: Map169    def __init__(self, service: ServiceModel):170        self._map = _create_service_map(service)171    def match(self, request: Request) -> Tuple[OperationModel, Mapping[str, Any]]:172        """173        Matches the given request to the operation it targets (or raises an exception if no operation matches).174        :param request: The request of which the targeting operation needs to be found175        :return: A tuple with the matched operation and the (already parsed) path params176        :raises: Werkzeug's NotFound exception in case the given request does not match any operation177        """178        # bind the map to get the actual matcher (use an empty server_name, since there won't be a hostname matching)179        matcher: MapAdapter = self._map.bind("")180        # perform the matching181        rule, args = matcher.match(get_raw_path(request), method=request.method, return_rule=True)182        # if the found rule is a _RequestMatchingRule, the multi rule matching needs to be invoked to perform the183        # fine-grained matching based on the whole request184        if isinstance(rule, _RequestMatchingRule):...

service_store.py

Source:service_store.py

...39    "corrupt": "network",40    "reorder": "network",41    "blackhole": "network",42}43def _create_service_map(ctrs, network):44    services = {}45    for k, v in ctrs.items():46        if network == 'bridge':47            service = v['Name']48        else:49            service = v["Name"].split("_")[1]50        v["Id"] = k51        services[service] = dict([("ctr_info", v),52                                  ("state", State("NORMAL"))])53    return services54def check_service(func):55    def _wrapper(self, service, *args, **kwargs):56        if service not in self.store.services:57            raise NetError("Unknown Service %s, check if the service is "58                           "running" % service)59        device = self._get_device(service)60        if not device:61            self.store.update_network_devices()62            # raise NetError("network interface for service %s could not be "63            #                "found" % service)64        return func(self, service, *args, **kwargs)65    return _wrapper66def update_service_state(func):67    def _wrapper(self, *args, **kwargs):68        services = args[0] if len(args) else kwargs.get('services')69        ret = func(self, *args, **kwargs)70        if ret:71            state = SERVICE_STATES.get(func.__name__)72            category = EVENT_CATEGORY.get(func.__name__)73            for service in services:74                self._update_service_state(service, category, state)75        return ret76    return _wrapper77class State(object):78    def __init__(self, init_state=None, dirty=True):79        self._state = {"network": [init_state],80                       "resource": [init_state],81                       "generic": [init_state]}82        self._dirty = dirty83    @property84    def state(self):85        return self._state86    def update(self, category, value, fresh=False):87        value = [value] if not isinstance(value, list) else value88        if fresh:89            self._state[category] = value90        else:91            if value not in self._state[category]:92                self._state[category].extend(value)93    def current_states(self):94        return list(set([x for v in self.state.values() for x in v]))95    @property96    def dirty(self):97        return self._dirty98    @dirty.setter99    def dirty(self, value):100        self._dirty = value101    def __iter__(self):102        return iter(self.current_states())103    def __contains__(self, state):104        return state in self.current_states()105    def __len__(self):106        return len(self.current_states())107class ServiceStore(object):108    _services = {}109    _network_init = False110    def __init__(self, client, network):111        self.client = client112        self._network = network113        self.update_service_map()114    def get_device(self, service):115        chaos_logger.debug("Getting network interface for '%s'" % service)116        service_info = self._services.get(service, None)117        if service_info:118            return service_info.get("device", None)119    def update_network_devices(self):120        chaos_logger.debug("Getting network interfaces for containers")121        cmd = ["ip", "link"]122        host_res = docker_run(' '.join(cmd), client=self.client, stream=True)123        chaos_logger.debug("IP LINK \n %s" % host_res.decode('utf-8'))124        for service, val in self._services.items():125            chaos_logger.debug("Finding network interface for %s" % service)126            ctr = val['ctr_info']127            try:128                res = self.client.execute_command(129                    ctr, ['ip', 'link', 'show', 'eth0'], split_res=False,130                    stream=False, retry_on_error=False131                )132                if isinstance(res, (list, tuple)):133                    o, e = res134                    o = "".join(o) + "".join(e)135                    res = o136            except DockerClientException as e:137                res = e.message138            res = res.decode('utf-8')139            device = re.search('^([0-9]+):', res)140            if not device:141                chaos_logger.warning("Problem determining host device id"142                                     "for service '%s' - %s" % (service, res))143                self._services.pop(service)144            else:145                peer_idx = int(device.group(1))146                host_idx = peer_idx + 1147                host_rgx = '^%d: ([^:@]+)[:@]' % host_idx148                host_match = re.search(host_rgx, host_res, re.M)149                if not host_match:150                    chaos_logger.warning(151                        "Problem determining host network device "152                        "for service '%s' - %s, could not match host device "153                        "" % (service, res))154                    chaos_logger.debug(155                        "peer -id : %s , host_id: %s" % (peer_idx, host_idx)156                    )157                    self._services.pop(service)158                else:159                    host_device = host_match.group(1)160                    device = host_device  # NetworkDevice(host_device)161                    val['device'] = device162    def update_service_map(self):163        chaos_logger.debug("Creating Serving - container map")164        networks = self.client.list_networks()165        network = [netw for netw in networks if self._network in netw[166            'Name']]167        if not network:168            self._network = "\033[1m '%s' \033[0m" % self._network169            raise NetError("Network not found: %s" % self._network)170        network = network[0]171        ctrs = network.get("Containers", {})172        if not ctrs:173            network = "\033[1m'%s'\033[0m" % network['Name']174            raise NoServiceRunningError(175                "No active containers/services found "176                "for network : %s" % network177            )178        self._services = _create_service_map(ctrs, network.get('Name'))179    def get_state(self, service):180        ser = self.get_service(service)181        x = ser['state'].current_states()182        chaos_logger.debug("Current network state for '%s' is '%s'" % (183            service, x))184        return x185    def get_services(self):186        return self._services187    def get_service(self, service):188        service_info = self.get_services().get(service)189        if not service_info:190            raise UnknownServiceError("Service '%s' not found, make sure the "191                                      "service is running " % service)192        return service_info...

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.