Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _collect_operations method in localstack

Best Python code snippet using localstack_python

core.py

Source:core.py

...167            for s in servers168        ]169        if not self._server and self.servers:170            self._server = self.servers[0]171        self._collect_operations()172    def _collect_operations(self):173        self._operations = {}174        for path, path_spec in self.paths.items():175            for method, op_spec in path_spec.items():176                operation_id = op_spec.get(OpenAPIKeyWord.OPERATION_ID)177                if not operation_id:178                    log.warning(179                        f"'{OpenAPIKeyWord.OPERATION_ID}' not found in: '[{method}] {path}'"180                    )181                    continue182                op = Operation(183                    path,184                    method,185                    op_spec,186                    requestor=self._requestor,187                    req_opts=self._req_opts,188                    server=self._server,189                )190                if operation_id not in self._operations:191                    self._operations[operation_id] = op192                else:193                    log.warning(194                        f"multiple '{operation_id}' found , operation ID should be unique"195                    )196                    v = self._operations[operation_id]197                    if type(v) is not list:198                        self._operations[operation_id] = [v]199                    self._operations[operation_id].append(op)200    def load_spec_from_url(self, url):201        spec = load_spec_from_url(url)202        self.load_spec(spec)203        return204    def load_spec_from_file(self, file_path):205        spec = load_spec_from_file(file_path)206        self.load_spec(spec)207        return208    @property209    def requestor(self):210        return self._requestor211    def set_requestor(self, r: Requestor):212        self._requestor = r213        self._collect_operations()214    @property215    def server(self):216        return self._server217    def set_server(self, s):218        self._server = s219        self._collect_operations()220    def __getattr__(self, op_name):221        if op_name in self._operations:222            return self._operations[op_name]223        raise AttributeError(224            f"'{type(self).__name__}' object has no attribute '{op_name}'"...

test_service_router.py

Source:test_service_router.py

...9from localstack.aws.protocol.service_router import determine_aws_service_name, get_service_catalog10from localstack.http import Request11from localstack.utils.aws import aws_stack12from localstack.utils.run import to_str13def _collect_operations() -> Tuple[ServiceModel, OperationModel]:14    """15    Collects all service<>operation combinations to test.16    """17    service_catalog = get_service_catalog()18    for service in service_catalog.services.values():19        for operation_name in service.operation_names:20            # FIXME try to support more and more services, get these exclusions down!21            # Exclude all operations for the following, currently _not_ supported services22            if service.service_name in [23                "chime",24                "chime-sdk-identity",25                "chime-sdk-media-pipelines",26                "chime-sdk-meetings",27                "chime-sdk-messaging",28                "connect",29                "connect-contact-lens",30                "greengrassv2",31                "iot1click",32                "iot1click-devices",33                "iot1click-projects",34                "kinesis-video-archived",35                "kinesis-video-archived-media",36                "kinesis-video-media",37                "kinesis-video-signaling",38                "kinesisvideo",39                "lex-models",40                "lex-runtime",41                "lexv2-models",42                "lexv2-runtime",43                "personalize",44                "personalize-events",45                "personalize-runtime",46                "pinpoint-sms-voice",47                "sagemaker-a2i-runtime",48                "sagemaker-edge",49                "sagemaker-featurestore-runtime",50                "sagemaker-runtime",51                "sms-voice",52                "sso",53                "sso-oidc",54            ]:55                yield pytest.param(56                    service,57                    service.operation_model(operation_name),58                    marks=pytest.mark.xfail(59                        reason=f"{service.service_name} is currently not supported by the service router"60                    ),61                )62            # Exclude services / operations which have ambiguities and where the service routing needs to resolve those63            elif (64                service.service_name in ["docdb", "neptune"]  # maps to rds65                or service.service_name in "timestream-write"  # maps to timestream-query66                or (67                    service.service_name == "sesv2"68                    and operation_name == "PutEmailIdentityDkimSigningAttributes"69                )70            ):71                yield pytest.param(72                    service,73                    service.operation_model(operation_name),74                    marks=pytest.mark.skip(75                        reason=f"{service.service_name} may differ due to ambiguities in the service specs"76                    ),77                )78            else:79                yield service, service.operation_model(operation_name)80@lru_cache81def _client(service: str):82    """Creates a boto client to create the request for a specific service."""83    config = Config(84        connect_timeout=1_000,85        read_timeout=1_000,86        retries={"total_max_attempts": 1},87        parameter_validation=False,88        user_agent="aws-cli/1.33.7",89    )90    return aws_stack.create_external_boto_client(service, config=config)91def _botocore_request_to_localstack_request(request_object: AWSRequest) -> Request:92    """Converts a botocore request (AWSRequest) to our HTTP framework's Request object based on Werkzeug."""93    split_url = urlsplit(request_object.url)94    path = split_url.path95    query_string = split_url.query96    body = request_object.body97    headers = request_object.headers98    return Request(99        method=request_object.method or "GET",100        path=path,101        query_string=to_str(query_string),102        headers=dict(headers),103        body=body,104        raw_path=path,105    )106# Simple dummy value mapping for the different shape types107_dummy_values = {108    "string": "dummy-value",109    "list": [],110    "integer": 0,111    "long": 0,112    "timestamp": datetime.now(),113}114def _create_dummy_request_args(operation_model: OperationModel) -> Dict:115    """Creates a dummy request param dict for the given operation."""116    input_shape: StructureShape = operation_model.input_shape117    if not input_shape:118        return {}119    result = {}120    for required_member in input_shape.required_members:121        required_shape: Shape = input_shape.members[required_member]122        location = required_shape.serialization.get("location")123        if location in ["uri", "querystring", "header", "headers"]:124            result[required_member] = _dummy_values[required_shape.type_name]125    return result126def _generate_test_name(param: Any):127    """Simple helper function to generate readable test names."""128    if isinstance(param, ServiceModel):129        return param.service_name130    elif isinstance(param, OperationModel):131        return param.name132    return param133@pytest.mark.parametrize(134    "service, operation",135    _collect_operations(),136    ids=_generate_test_name,137)138def test_service_router_works_for_every_service(139    service: ServiceModel, operation: OperationModel, caplog140):141    caplog.set_level("CRITICAL", "botocore")142    # Create a dummy request for the service router143    client = _client(service.service_name)144    request_context = {145        "client_region": client.meta.region_name,146        "client_config": client.meta.config,147        "has_streaming_input": operation.has_streaming_input,148        "auth_type": operation.auth_type,149    }...

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.