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

How to use stop_version method in localstack

Best Python code snippet using localstack_python

lambda_service.py

Source:lambda_service.py

...44            shutdown_futures.append(self.task_executor.submit(version_manager.stop))45        concurrent.futures.wait(shutdown_futures, timeout=5)46        self.task_executor.shutdown()47        # self.task_executor.shutdown(cancel_futures=True)  # TODO: python 3.9+48    def stop_version(self, qualified_arn: str) -> None:49        """50        Stops a specific lambda service version51        :param qualified_arn: Qualified arn for the version to stop52        """53        LOG.debug("Stopping version %s", qualified_arn)54        version_manager = self.lambda_version_managers.pop(qualified_arn)55        if not version_manager:56            raise ValueError(f"Unable to find version manager for {qualified_arn}")57        self.task_executor.submit(version_manager.stop)58    def get_lambda_version_manager(self, function_arn: str) -> LambdaVersionManager:59        """60        Get the lambda version for the given arn61        :param function_arn: qualified arn for the lambda version62        :return: LambdaVersionManager for the arn63        """64        version_manager = self.lambda_version_managers.get(function_arn)65        if not version_manager:66            raise ValueError(f"Could not find version '{function_arn}'. Is it created?")67        return version_manager68    def create_function(69        self,70        account_id: str,71        region_name: str,72        function_name: str,73        function_config: VersionFunctionConfiguration,74        code: Code,75    ) -> FunctionVersion:76        state = LambdaServiceBackend.get(region_name)77        fn = Function(function_name=function_name)78        with self.create_fn_lock:79            arn = VersionIdentifier(function_name, "$LATEST", region_name, account_id)80            zip_file_content = code.zip_file81            code_sha_256 = base64.standard_b64encode(82                hashlib.sha256(zip_file_content).digest()83            ).decode("utf-8")84            version = FunctionVersion(85                id=arn,86                qualifier="$LATEST",87                code=code,88                config_meta=FunctionConfigurationMeta(89                    function_arn=arn.qualified_arn(),90                    revision_id="?",91                    code_size=len(code.zip_file),92                    coda_sha256=code_sha_256,93                    last_modified="asdf",94                    last_update=UpdateStatus(status="Successful"),95                ),96                config=function_config,97            )98            fn.versions["$LATEST"] = version99            state.functions[function_name] = fn100        self.create_function_version(version)101        return version102    # TODO: is this sync?103    def delete_function(self, region_name: str, function_name: str):104        state = LambdaServiceBackend.get(region_name)105        function = state.functions.pop(function_name)106        for version in function.versions.values():107            self.stop_version(qualified_arn=version.id.qualified_arn())108    def delete_version(self, region_name: str, function_name: str, version_qualifier: str):109        state = LambdaServiceBackend.get(region_name)110        version = state.functions[function_name].versions[version_qualifier]111        self.stop_version(qualified_arn=version.id.qualified_arn())112    def get_function_version(113        self, region_name: str, function_name: str, qualifier: Optional[str] = "$LATEST"114    ) -> FunctionVersion:115        state = LambdaServiceBackend.get(region_name)116        return state.functions[function_name].versions[qualifier]117    def list_function_versions(self, region_name: str) -> List[FunctionVersion]:118        state = LambdaServiceBackend.get(region_name)119        return [f.latest() for f in state.functions.values()]  # TODO: qualifier120    def create_function_version(self, function_version: FunctionVersion) -> None:121        with self.lambda_version_manager_lock:122            qualified_arn = function_version.id.qualified_arn()123            version_manager = self.lambda_version_managers.get(qualified_arn)124            if version_manager:125                raise Exception("Version '%s' already created", qualified_arn)...

gae.py

Source:gae.py

...36                    service_id,37                    version_id,38                    self.project39                )40                self.stop_version(service_id, version_id)41        except HttpError as http_error:42            logging.error(http_error)43            return "Error", 50044        return "ok", 20045    @backoff.on_exception(backoff.expo, HttpError, max_tries=8, giveup=utils.fatal_code)46    def stop_version(self, service_id, version_id):47        """48        Stop an instance.49        Args:50            service_id: The App Engine service id51            version_id: The App Engine version id52        Returns:53        """54        # TODO add requestId55        return (56            self.app.apps().services().versions().patch(servicesId=service_id, appsId=self.project,57                                                        versionsId=version_id, updateMask='servingStatus',58                                                        body={"servingStatus": "STOPPED"}).execute()59        )60    @backoff.on_exception(backoff.expo, HttpError, max_tries=8, giveup=utils.fatal_code)...

sxs_layers.py

Source:sxs_layers.py

1import logging2import click3from regparser.index import dependency, entry4from regparser.layer.section_by_section import SectionBySection5logger = logging.getLogger(__name__)6def previous_sxs(cfr_title, cfr_part, stop_version):7    """The SxS layer relies on all notices that came before a particular8    version"""9    sub_entries = entry.FinalVersion(cfr_title, cfr_part).sub_entries()10    version_ids = [e.path[-1] for e in sub_entries]11    for previous_version in version_ids:12        yield entry.SxS(previous_version)13        if previous_version == stop_version:14            break15def is_stale(cfr_title, cfr_part, version_id):16    """Modify and process dependency graph related to a single SxS layer"""17    deps = dependency.Graph()18    layer_entry = entry.Layer(cfr_title, cfr_part, version_id, 'analyses')19    # Layers depend on their associated tree20    deps.add(layer_entry, entry.Tree(cfr_title, cfr_part, version_id))21    # And on all notices which came before22    for sxs_entry in previous_sxs(cfr_title, cfr_part, version_id):23        deps.add(layer_entry, sxs_entry)24    deps.validate_for(layer_entry)25    return deps.is_stale(layer_entry)26@click.command()27@click.argument('cfr_title', type=int)28@click.argument('cfr_part', type=int)29def sxs_layers(cfr_title, cfr_part):30    """Build SxS layers for all known versions."""31    logger.info("Build SxS layers - %s CFR %s", cfr_title, cfr_part)32    for tree_entry in entry.Tree(cfr_title, cfr_part).sub_entries():33        version_id = tree_entry.path[-1]34        if is_stale(cfr_title, cfr_part, version_id):35            tree = tree_entry.read()36            notices = [sxs.read() for sxs in previous_sxs(37                cfr_title, cfr_part, version_id)]38            layer_json = SectionBySection(tree, notices).build()39            entry.Layer.cfr(cfr_title, cfr_part, version_id, 'analyses').write(...

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.