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

How to use get_image_for_runtime method in localstack

Best Python code snippet using localstack_python

runtime_executor.py

Source:runtime_executor.py

...49def get_code_path_for_function(function_version: FunctionVersion) -> Path:50    return get_path_for_function(function_version) / "code"51def get_image_name_for_function(function_version: FunctionVersion) -> str:52    return f"localstack/lambda-{function_version.id.qualified_arn().replace(':', '_').replace('$', '_').lower()}"53def get_image_for_runtime(runtime: str) -> str:54    runtime, version = get_runtime_split(runtime)55    return f"{IMAGE_PREFIX}{runtime}:{version}"56def get_runtime_client_path() -> Path:57    return Path(LAMBDA_RUNTIME_INIT_PATH)58def prepare_image(target_path: Path, function_version: FunctionVersion) -> None:59    if not function_version.config.runtime:60        raise NotImplementedError("Custom images are currently not supported")61    src_init = get_runtime_client_path()62    # copy init file63    target_init = target_path / "aws-lambda-rie"64    shutil.copy(src_init, target_init)65    target_init.chmod(0o755)66    # copy code67    # create dockerfile68    docker_file_path = target_path / "Dockerfile"69    docker_file = LAMBDA_DOCKERFILE.format(70        base_img=get_image_for_runtime(function_version.config.runtime),71        rapid_entrypoint=RAPID_ENTRYPOINT,72    )73    with docker_file_path.open(mode="w") as f:74        f.write(docker_file)75    try:76        CONTAINER_CLIENT.build_image(77            dockerfile_path=str(docker_file_path),78            image_name=get_image_name_for_function(function_version),79        )80    except Exception as e:81        if LOG.isEnabledFor(logging.DEBUG):82            LOG.exception(83                "Error while building prebuilt lambda image for '%s'",84                function_version.qualified_arn,85            )86        else:87            LOG.error(88                "Error while building prebuilt lambda image for '%s', Error: %s",89                function_version.qualified_arn,90                e,91            )92def prepare_version(function_version: FunctionVersion) -> None:93    if not function_version.code.zip_file:94        raise NotImplementedError("Images without zipfile are currently not supported")95    time_before = time.perf_counter()96    target_path = get_path_for_function(function_version)97    target_path.mkdir(parents=True, exist_ok=True)98    # write code to disk99    target_code = get_code_path_for_function(function_version)100    with NamedTemporaryFile() as file:101        file.write(function_version.code.zip_file)102        file.flush()103        unzip(file.name, str(target_code))104    if config.LAMBDA_PREBUILD_IMAGES:105        prepare_image(target_path, function_version)106    LOG.debug("Version preparation took %0.2fms", (time.perf_counter() - time_before) * 1000)107def cleanup_version(function_version: FunctionVersion) -> None:108    function_path = get_path_for_function(function_version)109    try:110        shutil.rmtree(function_path)111    except OSError as e:112        LOG.debug(113            "Could not cleanup function %s due to error %s while deleting file %s",114            function_version.qualified_arn,115            e.strerror,116            e.filename,117        )118    if config.LAMBDA_PREBUILD_IMAGES:119        CONTAINER_CLIENT.remove_image(get_image_name_for_function(function_version))120class LambdaRuntimeException(Exception):121    def __init__(self, message: str):122        super().__init__(message)123class RuntimeExecutor:124    id: str125    function_version: FunctionVersion126    ip: Optional[str]127    executor_endpoint: Optional[ExecutorEndpoint]128    def __init__(129        self, id: str, function_version: FunctionVersion, service_endpoint: ServiceEndpoint130    ) -> None:131        self.id = id132        self.function_version = function_version133        self.ip = None134        self.executor_endpoint = self._build_executor_endpoint(service_endpoint)135    def get_image(self) -> str:136        if not self.function_version.config.runtime:137            raise NotImplementedError("Custom images are currently not supported")138        return (139            get_image_name_for_function(self.function_version)140            if config.LAMBDA_PREBUILD_IMAGES141            else get_image_for_runtime(self.function_version.config.runtime)142        )143    def _build_executor_endpoint(self, service_endpoint: ServiceEndpoint) -> ExecutorEndpoint:144        port = get_free_tcp_port()145        LOG.debug(146            "Creating service endpoint for function %s executor %s",147            self.function_version.qualified_arn,148            self.id,149        )150        executor_endpoint = ExecutorEndpoint(port, service_endpoint=service_endpoint)151        LOG.debug(152            "Finished creating service endpoint for function %s executor %s",153            self.function_version.qualified_arn,154            self.id,155        )...

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.