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

How to use get_docker_container_status method in localstack

Best Python code snippet using localstack_python

lambda_executors.py

Source:lambda_executors.py

...226        """227        with self.docker_container_lock:228            # Get the container name and id.229            container_name = self.get_container_name(func_arn)230            status = self.get_docker_container_status(func_arn)231            LOG.debug('Priming docker container (status "%s"): %s' % (status, container_name))232            # Container is not running or doesn't exist.233            if status < 1:234                # Make sure the container does not exist in any form/state.235                self.destroy_docker_container(func_arn)236                env_vars_str = ' '.join(['-e {}={}'.format(k, cmd_quote(v)) for (k, v) in env_vars])237                network = config.LAMBDA_DOCKER_NETWORK238                network_str = ' --network="%s" ' % network if network else ''239                # Create and start the container240                LOG.debug('Creating container: %s' % container_name)241                cmd = (242                    'docker create'243                    ' --rm'244                    ' --name "%s"'245                    ' --entrypoint /bin/bash'  # Load bash when it starts.246                    ' --interactive'  # Keeps the container running bash.247                    ' -e AWS_LAMBDA_EVENT_BODY="$AWS_LAMBDA_EVENT_BODY"'248                    ' -e HOSTNAME="$HOSTNAME"'249                    ' -e LOCALSTACK_HOSTNAME="$LOCALSTACK_HOSTNAME"'250                    '  %s'  # env_vars251                    '  %s'  # network252                    ' lambci/lambda:%s'253                ) % (container_name, env_vars_str, network_str, runtime)254                LOG.debug(cmd)255                run(cmd)256                LOG.debug('Copying files to container "%s" from "%s".' % (container_name, lambda_cwd))257                cmd = (258                    'docker cp'259                    ' "%s/." "%s:/var/task"'260                ) % (lambda_cwd, container_name)261                LOG.debug(cmd)262                run(cmd)263                LOG.debug('Starting container: %s' % container_name)264                cmd = 'docker start %s' % (container_name)265                LOG.debug(cmd)266                run(cmd)267                # give the container some time to start up268                time.sleep(1)269            # Get the entry point for the image.270            LOG.debug('Getting the entrypoint for image: lambci/lambda:%s' % runtime)271            cmd = (272                'docker image inspect'273                ' --format="{{ .ContainerConfig.Entrypoint }}"'274                ' lambci/lambda:%s'275            ) % (runtime)276            LOG.debug(cmd)277            run_result = run(cmd)278            entry_point = run_result.strip('[]\n\r ')279            container_network = self.get_docker_container_network(func_arn)280            LOG.debug('Using entrypoint "%s" for container "%s" on network "%s".'281                % (entry_point, container_name, container_network))282            return ContainerInfo(container_name, entry_point)283    def destroy_docker_container(self, func_arn):284        """285        Stops and/or removes a docker container for a specific lambda function ARN.286        :param func_arn: The ARN of the lambda function.287        :return: None288        """289        with self.docker_container_lock:290            status = self.get_docker_container_status(func_arn)291            # Get the container name and id.292            container_name = self.get_container_name(func_arn)293            if status == 1:294                LOG.debug('Stopping container: %s' % container_name)295                cmd = (296                    'docker stop -t0 %s'297                ) % (container_name)298                LOG.debug(cmd)299                run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)300                status = self.get_docker_container_status(func_arn)301            if status == -1:302                LOG.debug('Removing container: %s' % container_name)303                cmd = (304                    'docker rm %s'305                ) % (container_name)306                LOG.debug(cmd)307                run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)308    def get_all_container_names(self):309        """310        Returns a list of container names for lambda containers.311        :return: A String[] localstack docker container names for each function.312        """313        with self.docker_container_lock:314            LOG.debug('Getting all lambda containers names.')315            cmd = 'docker ps -a --filter="name=localstack_lambda_*" --format "{{.Names}}"'316            LOG.debug(cmd)317            cmd_result = run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE).strip()318            if len(cmd_result) > 0:319                container_names = cmd_result.split('\n')320            else:321                container_names = []322            return container_names323    def destroy_existing_docker_containers(self):324        """325        Stops and/or removes all lambda docker containers for localstack.326        :return: None327        """328        with self.docker_container_lock:329            container_names = self.get_all_container_names()330            LOG.debug('Removing %d containers.' % len(container_names))331            for container_name in container_names:332                cmd = 'docker rm -f %s' % container_name333                LOG.debug(cmd)334                run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)335    def get_docker_container_status(self, func_arn):336        """337        Determine the status of a docker container.338        :param func_arn: The ARN of the lambda function.339        :return: 1 If the container is running,340        -1 if the container exists but is not running341        0 if the container does not exist.342        """343        with self.docker_container_lock:344            # Get the container name and id.345            container_name = self.get_container_name(func_arn)346            # Check if the container is already running347            # Note: filtering by *exact* name using regex filter '^...$' seems unstable on some348            # systems. Therefore, we use a combination of filter and grep to get the results.349            cmd = ('docker ps -a --filter name=\'%s\' '350                   '--format "{{ .Status }} - {{ .Names }}" '351                   '| grep -w "%s" | cat') % (container_name, container_name)352            LOG.debug('Getting status for container "%s": %s' % (container_name, cmd))353            cmd_result = run(cmd)354            # If the container doesn't exist. Create and start it.355            container_status = cmd_result.strip()356            if len(container_status) == 0:357                return 0358            if container_status.lower().startswith('up '):359                return 1360            return -1361    def get_docker_container_network(self, func_arn):362        """363        Determine the network of a docker container.364        :param func_arn: The ARN of the lambda function.365        :return: name of the container network366        """367        with self.docker_container_lock:368            status = self.get_docker_container_status(func_arn)369            # container does not exist370            if status == 0:371                return ''372            # Get the container name.373            container_name = self.get_container_name(func_arn)374            # Get the container network375            LOG.debug('Getting container network: %s' % container_name)376            cmd = (377                'docker inspect %s'378                ' --format "{{ .HostConfig.NetworkMode }}"'379            ) % (container_name)380            LOG.debug(cmd)381            cmd_result = run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)382            container_network = cmd_result.strip()...

aws_linux_host.py

Source:aws_linux_host.py

...42        @return: boolean service status43        """44        return self._host.service(service_name).is_running4546    def get_docker_container_status(self) -> list:47        """Returns running containers."""48        containers = []49        info = self._host.run('sudo docker ps')50        if info.exit_status == 0:51            list_of_containers = info.stdout.split()52            return list_of_containers53        else:54            if info.stderr:55                logging.error(info.stderr)56            else:57                logging.error("Can't get list of running containers")58            return containers5960    def get_root_volume_size(self) -> str:61        """Returns root volume size."""62        mount_point = self._host.mount_point("/").device63        info = self._host.run(f'lsblk {mount_point}')64        if info.exit_status == 0:65            list_of_point = info.stdout.split()66            return list_of_point[-4]67        else:68            if info.stderr:69                logging.error(info.stderr)70            else:71                logging.error("Can't get volume size")72            return None7374host = '3.68.105.182'75user_name ='ec2-user'76ssh_config_path = '/home/pavel/PycharmProjects/testinfra/ssh_config'77# a = AWSLinuxHost(user_name, host, ssh_config_path)78a = testinfra.get_host("ssh://ec2-user@3.68.105.182", ssh_config="/home/pavel/PycharmProjects/testinfra/ssh_config")79print(a)80print(dir(a))81print(a.__getattribute__)82print(a.backend)83print(a.socket.get_listening_sockets())8485# print(a.get_list_of_listening_ports())86# print(a.get_listening_ports_status(22))87# print(a.get_package_installation_status('docker'))88# print(a.get_service_running_status('docker'))89# print(a.get_docker_container_status())
...

view.py

Source:view.py

...9    containers = docker_.docker_find_container({"name": name})10    if containers is False:11        source = update_source({"code": -1, "msg": "è¯¥macå°åå®¹å¨æªæ¾å°"})12    else:13        _, status = docker_.get_docker_container_status(containers)14        logs = docker_.get_docker_container_logs(containers)15        read_state = 1 if status == "running" else 016        source = update_source({"msg": logs, "read_state": read_state})...

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.