KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use format_windows_path method in localstack

Best Python code snippet using localstack_python
lambda_executors.py
Source:lambda_executors.py
...352                dns_str = '--dns="%s"' % dns if dns else ''353                mount_volume = not config.LAMBDA_REMOTE_DOCKER354                lambda_cwd_on_host = Util.get_host_path_for_path_in_docker(lambda_cwd)355                if (':' in lambda_cwd and '\\' in lambda_cwd):356                    lambda_cwd_on_host = Util.format_windows_path(lambda_cwd_on_host)357                mount_volume_str = '-v "%s":/var/task' % lambda_cwd_on_host if mount_volume else ''358                # Create and start the container359                LOG.debug('Creating container: %s' % container_name)360                cmd = (361                    '%s create'362                    ' %s'  # --rm flag363                    ' --name "%s"'364                    ' --entrypoint /bin/bash'  # Load bash when it starts.365                    ' %s'366                    ' --interactive'  # Keeps the container running bash.367                    ' -e AWS_LAMBDA_EVENT_BODY="$AWS_LAMBDA_EVENT_BODY"'368                    ' -e HOSTNAME="$HOSTNAME"'369                    ' -e LOCALSTACK_HOSTNAME="$LOCALSTACK_HOSTNAME"'370                    ' -e EDGE_PORT="$EDGE_PORT"'371                    '  %s'  # env_vars372                    '  %s'  # network373                    '  %s'  # dns374                    ' %s'375                ) % (docker_cmd, rm_flag, container_name, mount_volume_str,376                    env_vars_str, network_str, dns_str, docker_image)377                LOG.debug(cmd)378                run(cmd)379                if not mount_volume:380                    LOG.debug('Copying files to container "%s" from "%s".' % (container_name, lambda_cwd))381                    cmd = (382                        '%s cp'383                        ' "%s/." "%s:/var/task"'384                    ) % (docker_cmd, lambda_cwd, container_name)385                    LOG.debug(cmd)386                    run(cmd)387                LOG.debug('Starting container: %s' % container_name)388                cmd = '%s start %s' % (docker_cmd, container_name)389                LOG.debug(cmd)390                run(cmd)391                # give the container some time to start up392                time.sleep(1)393            # Get the entry point for the image.394            LOG.debug('Getting the entrypoint for image: %s' % (docker_image))395            cmd = (396                '%s image inspect'397                ' --format="{{ .Config.Entrypoint }}"'398                ' %s'399            ) % (docker_cmd, docker_image)400            LOG.debug(cmd)401            run_result = run(cmd)402            entry_point = run_result.strip('[]\n\r ')403            container_network = self.get_docker_container_network(func_arn)404            LOG.debug('Using entrypoint "%s" for container "%s" on network "%s".'405                % (entry_point, container_name, container_network))406            return ContainerInfo(container_name, entry_point)407    def destroy_docker_container(self, func_arn):408        """409        Stops and/or removes a docker container for a specific lambda function ARN.410        :param func_arn: The ARN of the lambda function.411        :return: None412        """413        with self.docker_container_lock:414            status = self.get_docker_container_status(func_arn)415            docker_cmd = self._docker_cmd()416            # Get the container name and id.417            container_name = self.get_container_name(func_arn)418            if status == 1:419                LOG.debug('Stopping container: %s' % container_name)420                cmd = '%s stop -t0 %s' % (docker_cmd, container_name)421                LOG.debug(cmd)422                run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)423                status = self.get_docker_container_status(func_arn)424            if status == -1:425                LOG.debug('Removing container: %s' % container_name)426                cmd = '%s rm %s' % (docker_cmd, container_name)427                LOG.debug(cmd)428                run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)429    def get_all_container_names(self):430        """431        Returns a list of container names for lambda containers.432        :return: A String[] localstack docker container names for each function.433        """434        with self.docker_container_lock:435            LOG.debug('Getting all lambda containers names.')436            cmd = '%s ps -a --filter="name=localstack_lambda_*" --format "{{.Names}}"' % self._docker_cmd()437            LOG.debug(cmd)438            cmd_result = run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE).strip()439            if len(cmd_result) > 0:440                container_names = cmd_result.split('\n')441            else:442                container_names = []443            return container_names444    def destroy_existing_docker_containers(self):445        """446        Stops and/or removes all lambda docker containers for localstack.447        :return: None448        """449        with self.docker_container_lock:450            container_names = self.get_all_container_names()451            LOG.debug('Removing %d containers.' % len(container_names))452            for container_name in container_names:453                cmd = '%s rm -f %s' % (self._docker_cmd(), container_name)454                LOG.debug(cmd)455                run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)456    def get_docker_container_status(self, func_arn):457        """458        Determine the status of a docker container.459        :param func_arn: The ARN of the lambda function.460        :return: 1 If the container is running,461        -1 if the container exists but is not running462        0 if the container does not exist.463        """464        with self.docker_container_lock:465            # Get the container name and id.466            container_name = self.get_container_name(func_arn)467            # Check if the container is already running468            # Note: filtering by *exact* name using regex filter '^...$' seems unstable on some469            # systems. Therefore, we use a combination of filter and grep to get the results.470            cmd = ("docker ps -a --filter name='%s' "471                   '--format "{{ .Status }} - {{ .Names }}" '472                   '| grep -w "%s" | cat') % (container_name, container_name)473            LOG.debug('Getting status for container "%s": %s' % (container_name, cmd))474            cmd_result = run(cmd)475            # If the container doesn't exist. Create and start it.476            container_status = cmd_result.strip()477            if len(container_status) == 0:478                return 0479            if container_status.lower().startswith('up '):480                return 1481            return -1482    def get_docker_container_network(self, func_arn):483        """484        Determine the network of a docker container.485        :param func_arn: The ARN of the lambda function.486        :return: name of the container network487        """488        with self.docker_container_lock:489            status = self.get_docker_container_status(func_arn)490            # container does not exist491            if status == 0:492                return ''493            # Get the container name.494            container_name = self.get_container_name(func_arn)495            docker_cmd = self._docker_cmd()496            # Get the container network497            LOG.debug('Getting container network: %s' % container_name)498            cmd = (499                '%s inspect %s'500                ' --format "{{ .HostConfig.NetworkMode }}"'501            ) % (docker_cmd, container_name)502            LOG.debug(cmd)503            cmd_result = run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)504            container_network = cmd_result.strip()505            return container_network506    def idle_container_destroyer(self):507        """508        Iterates though all the lambda containers and destroys any container that has509        been inactive for longer than MAX_CONTAINER_IDLE_TIME_MS.510        :return: None511        """512        LOG.info('Checking if there are idle containers.')513        current_time = int(time.time() * 1000)514        for func_arn, last_run_time in dict(self.function_invoke_times).items():515            duration = current_time - last_run_time516            # not enough idle time has passed517            if duration < MAX_CONTAINER_IDLE_TIME_MS:518                continue519            # container has been idle, destroy it.520            self.destroy_docker_container(func_arn)521    def start_idle_container_destroyer_interval(self):522        """523        Starts a repeating timer that triggers start_idle_container_destroyer_interval every 60 seconds.524        Thus checking for idle containers and destroying them.525        :return: None526        """527        self.idle_container_destroyer()528        threading.Timer(60.0, self.start_idle_container_destroyer_interval).start()529    def get_container_name(self, func_arn):530        """531        Given a function ARN, returns a valid docker container name.532        :param func_arn: The ARN of the lambda function.533        :return: A docker compatible name for the arn.534        """535        return 'localstack_lambda_' + re.sub(r'[^a-zA-Z0-9_.-]', '_', func_arn)536class LambdaExecutorSeparateContainers(LambdaExecutorContainers):537    def __init__(self):538        super(LambdaExecutorSeparateContainers, self).__init__()539        self.max_port = LAMBDA_API_UNIQUE_PORTS540        self.port_offset = LAMBDA_API_PORT_OFFSET541    def prepare_event(self, environment, event_body):542        # Tell Lambci to use STDIN for the event543        environment['DOCKER_LAMBDA_USE_STDIN'] = '1'544        return event_body.encode()545    def prepare_execution(self, func_arn, env_vars, runtime, command, handler, lambda_cwd):546        entrypoint = ''547        if command:548            entrypoint = ' --entrypoint ""'549        else:550            command = '"%s"' % handler551        # add Docker Lambda env vars552        network = config.LAMBDA_DOCKER_NETWORK553        network_str = '--network="%s"' % network if network else ''554        if network == 'host':555            port = get_free_tcp_port()556            env_vars['DOCKER_LAMBDA_API_PORT'] = port557            env_vars['DOCKER_LAMBDA_RUNTIME_PORT'] = port558        dns = config.LAMBDA_DOCKER_DNS559        dns_str = '--dns="%s"' % dns if dns else ''560        env_vars_string = ' '.join(['-e {}="${}"'.format(k, k) for (k, v) in env_vars.items()])561        debug_docker_java_port = '-p {p}:{p}'.format(p=Util.debug_java_port) if Util.debug_java_port else ''562        docker_cmd = self._docker_cmd()563        docker_image = Util.docker_image_for_runtime(runtime)564        rm_flag = Util.get_docker_remove_flag()565        if config.LAMBDA_REMOTE_DOCKER:566            cmd = (567                'CONTAINER_ID="$(%s create -i'568                ' %s'  # entrypoint569                ' %s'  # debug_docker_java_port570                ' %s'  # env571                ' %s'  # network572                ' %s'  # dns573                ' %s'  # --rm flag574                ' %s %s'  # image and command575                ')";'576                '%s cp "%s/." "$CONTAINER_ID:/var/task"; '577                '%s start -ai "$CONTAINER_ID";'578            ) % (docker_cmd, entrypoint, debug_docker_java_port,579                env_vars_string, network_str, dns_str, rm_flag,580                 docker_image, command,581                 docker_cmd, lambda_cwd,582                 docker_cmd)583        else:584            lambda_cwd_on_host = Util.get_host_path_for_path_in_docker(lambda_cwd)585            cmd = (586                '%s run -i'587                ' %s -v "%s":/var/task'588                ' %s'589                ' %s'  # network590                ' %s'  # dns591                ' %s'  # --rm flag592                ' %s %s'593            ) % (docker_cmd, entrypoint, lambda_cwd_on_host, env_vars_string,594                 network_str, dns_str, rm_flag, docker_image, command)595        return cmd596class LambdaExecutorLocal(LambdaExecutor):597    def _execute(self, func_arn, func_details, event, context=None, version=None):598        lambda_cwd = func_details.cwd599        environment = self._prepare_environment(func_details)600        # execute the Lambda function in a forked sub-process, sync result via queue601        queue = Queue()602        lambda_function = func_details.function(version)603        def do_execute():604            # now we're executing in the child process, safe to change CWD and ENV605            path_before = sys.path606            try:607                if lambda_cwd:608                    os.chdir(lambda_cwd)609                    sys.path = [lambda_cwd] + sys.path610                if environment:611                    os.environ.update(environment)612                result = lambda_function(event, context)613                queue.put(result)614            finally:615                sys.path = path_before616        process = Process(target=do_execute)617        with CaptureOutput() as c:618            process.run()619        result = queue.get()620        # Make sure to keep the log line below, to ensure the log stream gets created621        log_output = 'START: Lambda %s started via "local" executor ...' % func_arn622        # TODO: Interweaving stdout/stderr currently not supported623        for stream in (c.stdout(), c.stderr()):624            if stream:625                log_output += ('\n' if log_output else '') + stream626        # store logs to CloudWatch627        _store_logs(func_details, log_output)628        return result629    def execute_java_lambda(self, event, context, main_file, func_details=None):630        handler = func_details.handler631        opts = config.LAMBDA_JAVA_OPTS if config.LAMBDA_JAVA_OPTS else ''632        event_file = EVENT_FILE_PATTERN.replace('*', short_uid())633        save_file(event_file, json.dumps(json_safe(event)))634        TMP_FILES.append(event_file)635        class_name = handler.split('::')[0]636        classpath = '%s:%s:%s' % (main_file, Util.get_java_classpath(main_file), LAMBDA_EXECUTOR_JAR)637        cmd = 'java %s -cp %s %s %s %s' % (opts, classpath, LAMBDA_EXECUTOR_CLASS, class_name, event_file)638        LOG.warning(cmd)639        result = self.run_lambda_executor(cmd, func_details=func_details)640        return result641class Util:642    debug_java_port = False643    @classmethod644    def get_java_opts(cls):645        opts = config.LAMBDA_JAVA_OPTS or ''646        # Replace _debug_port_ with a random free port647        if '_debug_port_' in opts:648            if not cls.debug_java_port:649                cls.debug_java_port = get_free_tcp_port()650            opts = opts.replace('_debug_port_', ('%s' % cls.debug_java_port))651        else:652            # Parse the debug port from opts653            m = re.match('.*address=(.+:)?(\\d+).*', opts)654            if m is not None:655                cls.debug_java_port = m.groups()[1]656        return opts657    @classmethod658    def get_host_path_for_path_in_docker(cls, path):659        return re.sub(r'^%s/(.*)$' % config.TMP_FOLDER,660                      r'%s/\1' % config.HOST_TMP_FOLDER, path)661    @classmethod662    def format_windows_path(cls, path):663        temp = path.replace(':', '').replace('\\', '/')664        if len(temp) >= 1 and temp[:1] != '/':665            temp = '/' + temp666        temp = '%s%s' % (config.WINDOWS_DOCKER_MOUNT_PREFIX, temp)667        return temp668    @classmethod669    def docker_image_for_runtime(cls, runtime):670        docker_tag = runtime671        docker_image = config.LAMBDA_CONTAINER_REGISTRY672        # TODO: remove prefix once execution issues are fixed with dotnetcore/python lambdas673        # See https://github.com/lambci/docker-lambda/pull/218674        lambdas_to_add_prefix = ['dotnetcore2.0', 'dotnetcore2.1', 'python2.7', 'python3.6', 'python3.7']675        if docker_image == 'lambci/lambda' and any(img in docker_tag for img in lambdas_to_add_prefix):676            docker_tag = '20191117-%s' % docker_tag...
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.