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

How to use _docker_cmd method in localstack

Best Python code snippet using localstack_python

docker_cmd_client.py

Source:docker_cmd_client.py

...26LOG = logging.getLogger(__name__)27class CmdDockerClient(ContainerClient):28    """Class for managing docker containers using the command line executable"""29    default_run_outfile: Optional[str] = None30    def _docker_cmd(self) -> List[str]:31        """Return the string to be used for running Docker commands."""32        return config.DOCKER_CMD.split()33    def get_container_status(self, container_name: str) -> DockerContainerStatus:34        cmd = self._docker_cmd()35        cmd += [36            "ps",37            "-a",38            "--filter",39            f"name={container_name}",40            "--format",41            "{{ .Status }} - {{ .Names }}",42        ]43        cmd_result = safe_run(cmd)44        # filter empty / invalid lines from docker ps output45        cmd_result = next((line for line in cmd_result.splitlines() if container_name in line), "")46        container_status = cmd_result.strip().lower()47        if len(container_status) == 0:48            return DockerContainerStatus.NON_EXISTENT49        elif "(paused)" in container_status:50            return DockerContainerStatus.PAUSED51        elif container_status.startswith("up "):52            return DockerContainerStatus.UP53        else:54            return DockerContainerStatus.DOWN55    def stop_container(self, container_name: str, timeout: int = None) -> None:56        if timeout is None:57            timeout = self.STOP_TIMEOUT58        cmd = self._docker_cmd()59        cmd += ["stop", "--time", str(timeout), container_name]60        LOG.debug("Stopping container with cmd %s", cmd)61        try:62            safe_run(cmd)63        except subprocess.CalledProcessError as e:64            if "No such container" in to_str(e.stdout):65                raise NoSuchContainer(container_name, stdout=e.stdout, stderr=e.stderr)66            else:67                raise ContainerException(68                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr69                )70    def pause_container(self, container_name: str) -> None:71        cmd = self._docker_cmd()72        cmd += ["pause", container_name]73        LOG.debug("Pausing container with cmd %s", cmd)74        try:75            safe_run(cmd)76        except subprocess.CalledProcessError as e:77            if "No such container" in to_str(e.stdout):78                raise NoSuchContainer(container_name, stdout=e.stdout, stderr=e.stderr)79            else:80                raise ContainerException(81                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr82                )83    def unpause_container(self, container_name: str) -> None:84        cmd = self._docker_cmd()85        cmd += ["unpause", container_name]86        LOG.debug("Unpausing container with cmd %s", cmd)87        try:88            safe_run(cmd)89        except subprocess.CalledProcessError as e:90            if "No such container" in to_str(e.stdout):91                raise NoSuchContainer(container_name, stdout=e.stdout, stderr=e.stderr)92            else:93                raise ContainerException(94                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr95                )96    def remove_image(self, image: str, force: bool = True) -> None:97        cmd = self._docker_cmd()98        cmd += ["rmi", image]99        if force:100            cmd += ["--force"]101        LOG.debug("Removing image %s %s", image, "(forced)" if force else "")102        try:103            safe_run(cmd)104        except subprocess.CalledProcessError as e:105            if "No such image" in to_str(e.stdout):106                raise NoSuchImage(image, stdout=e.stdout, stderr=e.stderr)107            else:108                raise ContainerException(109                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr110                )111    def commit(112        self,113        container_name_or_id: str,114        image_name: str,115        image_tag: str,116    ):117        cmd = self._docker_cmd()118        cmd += ["commit", container_name_or_id, f"{image_name}:{image_tag}"]119        LOG.debug(120            "Creating image from container %s as %s:%s", container_name_or_id, image_name, image_tag121        )122        try:123            safe_run(cmd)124        except subprocess.CalledProcessError as e:125            if "No such container" in to_str(e.stdout):126                raise NoSuchContainer(container_name_or_id, stdout=e.stdout, stderr=e.stderr)127            else:128                raise ContainerException(129                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr130                )131    def remove_container(self, container_name: str, force=True, check_existence=False) -> None:132        if check_existence and container_name not in self.get_running_container_names():133            return134        cmd = self._docker_cmd() + ["rm"]135        if force:136            cmd.append("-f")137        cmd.append(container_name)138        LOG.debug("Removing container with cmd %s", cmd)139        try:140            safe_run(cmd)141        except subprocess.CalledProcessError as e:142            if "No such container" in to_str(e.stdout):143                raise NoSuchContainer(container_name, stdout=e.stdout, stderr=e.stderr)144            else:145                raise ContainerException(146                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr147                )148    def list_containers(self, filter: Union[List[str], str, None] = None, all=True) -> List[dict]:149        filter = [filter] if isinstance(filter, str) else filter150        cmd = self._docker_cmd()151        cmd.append("ps")152        if all:153            cmd.append("-a")154        options = []155        if filter:156            options += [y for filter_item in filter for y in ["--filter", filter_item]]157        cmd += options158        cmd.append("--format")159        cmd.append("{{json . }}")160        try:161            cmd_result = safe_run(cmd).strip()162        except subprocess.CalledProcessError as e:163            raise ContainerException(164                "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr165            )166        container_list = []167        if cmd_result:168            container_list = [json.loads(line) for line in cmd_result.splitlines()]169        result = []170        for container in container_list:171            result.append(172                {173                    "id": container["ID"],174                    "image": container["Image"],175                    "name": container["Names"],176                    "status": container["State"],177                    "labels": container["Labels"],178                }179            )180        return result181    def copy_into_container(182        self, container_name: str, local_path: str, container_path: str183    ) -> None:184        cmd = self._docker_cmd()185        cmd += ["cp", local_path, f"{container_name}:{container_path}"]186        LOG.debug("Copying into container with cmd: %s", cmd)187        try:188            safe_run(cmd)189        except subprocess.CalledProcessError as e:190            if "No such container" in to_str(e.stdout):191                raise NoSuchContainer(container_name)192            raise ContainerException(193                f"Docker process returned with errorcode {e.returncode}", e.stdout, e.stderr194            )195    def copy_from_container(196        self, container_name: str, local_path: str, container_path: str197    ) -> None:198        cmd = self._docker_cmd()199        cmd += ["cp", f"{container_name}:{container_path}", local_path]200        LOG.debug("Copying from container with cmd: %s", cmd)201        try:202            safe_run(cmd)203        except subprocess.CalledProcessError as e:204            if "No such container" in to_str(e.stdout):205                raise NoSuchContainer(container_name)206            raise ContainerException(207                "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr208            )209    def pull_image(self, docker_image: str) -> None:210        cmd = self._docker_cmd()211        cmd += ["pull", docker_image]212        LOG.debug("Pulling image with cmd: %s", cmd)213        try:214            safe_run(cmd)215        except subprocess.CalledProcessError as e:216            if "pull access denied" in to_str(e.stdout):217                raise NoSuchImage(docker_image)218            raise ContainerException(219                "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr220            )221    def push_image(self, docker_image: str) -> None:222        cmd = self._docker_cmd()223        cmd += ["push", docker_image]224        LOG.debug("Pushing image with cmd: %s", cmd)225        try:226            safe_run(cmd)227        except subprocess.CalledProcessError as e:228            if "is denied" in to_str(e.stdout):229                raise AccessDenied(docker_image)230            if "does not exist" in to_str(e.stdout):231                raise NoSuchImage(docker_image)232            if "connection refused" in to_str(e.stdout):233                raise RegistryConnectionError(e.stdout)234            raise ContainerException(235                f"Docker process returned with errorcode {e.returncode}", e.stdout, e.stderr236            ) from e237    def build_image(self, dockerfile_path: str, image_name: str, context_path: str = None):238        cmd = self._docker_cmd()239        dockerfile_path = Util.resolve_dockerfile_path(dockerfile_path)240        context_path = context_path or os.path.dirname(dockerfile_path)241        cmd += ["build", "-t", image_name, "-f", dockerfile_path, context_path]242        LOG.debug("Building Docker image: %s", cmd)243        try:244            safe_run(cmd)245        except subprocess.CalledProcessError as e:246            raise ContainerException(247                f"Docker build process returned with error code {e.returncode}", e.stdout, e.stderr248            ) from e249    def tag_image(self, source_ref: str, target_name: str) -> None:250        cmd = self._docker_cmd()251        cmd += ["tag", source_ref, target_name]252        LOG.debug("Tagging Docker image %s as %s", source_ref, target_name)253        try:254            safe_run(cmd)255        except subprocess.CalledProcessError as e:256            if "No such image" in to_str(e.stdout):257                raise NoSuchImage(source_ref)258            raise ContainerException(259                f"Docker process returned with error code {e.returncode}", e.stdout, e.stderr260            ) from e261    def get_docker_image_names(self, strip_latest=True, include_tags=True):262        format_string = "{{.Repository}}:{{.Tag}}" if include_tags else "{{.Repository}}"263        cmd = self._docker_cmd()264        cmd += ["images", "--format", format_string]265        try:266            output = safe_run(cmd)267            image_names = output.splitlines()268            if strip_latest:269                Util.append_without_latest(image_names)270            return image_names271        except Exception as e:272            LOG.info('Unable to list Docker images via "%s": %s', cmd, e)273            return []274    def get_container_logs(self, container_name_or_id: str, safe=False) -> str:275        cmd = self._docker_cmd()276        cmd += ["logs", container_name_or_id]277        try:278            return safe_run(cmd)279        except subprocess.CalledProcessError as e:280            if safe:281                return ""282            if "No such container" in to_str(e.stdout):283                raise NoSuchContainer(container_name_or_id, stdout=e.stdout, stderr=e.stderr)284            else:285                raise ContainerException(286                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr287                )288    def _inspect_object(self, object_name_or_id: str) -> Dict[str, Union[Dict, str]]:289        cmd = self._docker_cmd()290        cmd += ["inspect", "--format", "{{json .}}", object_name_or_id]291        try:292            cmd_result = safe_run(cmd)293        except subprocess.CalledProcessError as e:294            if "No such object" in to_str(e.stdout):295                raise NoSuchObject(object_name_or_id, stdout=e.stdout, stderr=e.stderr)296            else:297                raise ContainerException(298                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr299                )300        image_data = json.loads(cmd_result.strip())301        return image_data302    def inspect_container(self, container_name_or_id: str) -> Dict[str, Union[Dict, str]]:303        try:304            return self._inspect_object(container_name_or_id)305        except NoSuchObject as e:306            raise NoSuchContainer(container_name_or_id=e.object_id)307    def inspect_image(self, image_name: str, pull: bool = True) -> Dict[str, Union[Dict, str]]:308        try:309            return self._inspect_object(image_name)310        except NoSuchObject as e:311            if pull:312                self.pull_image(image_name)313                return self.inspect_image(image_name, pull=False)314            raise NoSuchImage(image_name=e.object_id)315    def inspect_network(self, network_name: str) -> Dict[str, Union[Dict, str]]:316        try:317            return self._inspect_object(network_name)318        except NoSuchObject as e:319            raise NoSuchNetwork(network_name=e.object_id)320    def connect_container_to_network(321        self, network_name: str, container_name_or_id: str, aliases: Optional[List] = None322    ) -> None:323        LOG.debug(324            "Connecting container '%s' to network '%s' with aliases '%s'",325            container_name_or_id,326            network_name,327            aliases,328        )329        cmd = self._docker_cmd()330        cmd += ["network", "connect"]331        if aliases:332            cmd += ["--alias", ",".join(aliases)]333        cmd += [network_name, container_name_or_id]334        try:335            safe_run(cmd)336        except subprocess.CalledProcessError as e:337            stdout_str = to_str(e.stdout)338            if re.match(r".*network (.*) not found.*", stdout_str):339                raise NoSuchNetwork(network_name=network_name)340            elif "No such container" in stdout_str:341                raise NoSuchContainer(container_name_or_id, stdout=e.stdout, stderr=e.stderr)342            else:343                raise ContainerException(344                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr345                )346    def disconnect_container_from_network(347        self, network_name: str, container_name_or_id: str348    ) -> None:349        LOG.debug(350            "Disconnecting container '%s' from network '%s'", container_name_or_id, network_name351        )352        cmd = self._docker_cmd() + ["network", "disconnect", network_name, container_name_or_id]353        try:354            safe_run(cmd)355        except subprocess.CalledProcessError as e:356            stdout_str = to_str(e.stdout)357            if re.match(r".*network (.*) not found.*", stdout_str):358                raise NoSuchNetwork(network_name=network_name)359            elif "No such container" in stdout_str:360                raise NoSuchContainer(container_name_or_id, stdout=e.stdout, stderr=e.stderr)361            else:362                raise ContainerException(363                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr364                )365    def get_container_ip(self, container_name_or_id: str) -> str:366        cmd = self._docker_cmd()367        cmd += [368            "inspect",369            "--format",370            "{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}",371            container_name_or_id,372        ]373        try:374            result = safe_run(cmd).strip()375            return result.split(" ")[0] if result else ""376        except subprocess.CalledProcessError as e:377            if "No such object" in to_str(e.stdout):378                raise NoSuchContainer(container_name_or_id, stdout=e.stdout, stderr=e.stderr)379            else:380                raise ContainerException(381                    "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr382                )383    def has_docker(self) -> bool:384        try:385            safe_run(self._docker_cmd() + ["ps"])386            return True387        except subprocess.CalledProcessError:388            return False389    def create_container(self, image_name: str, **kwargs) -> str:390        cmd, env_file = self._build_run_create_cmd("create", image_name, **kwargs)391        LOG.debug("Create container with cmd: %s", cmd)392        try:393            container_id = safe_run(cmd)394            # Note: strip off Docker warning messages like "DNS setting (--dns=127.0.0.1) may fail in containers"395            container_id = container_id.strip().split("\n")[-1]396            return container_id.strip()397        except subprocess.CalledProcessError as e:398            if "Unable to find image" in to_str(e.stdout):399                raise NoSuchImage(image_name, stdout=e.stdout, stderr=e.stderr)400            raise ContainerException(401                "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr402            )403        finally:404            Util.rm_env_vars_file(env_file)405    def run_container(self, image_name: str, stdin=None, **kwargs) -> Tuple[bytes, bytes]:406        cmd, env_file = self._build_run_create_cmd("run", image_name, **kwargs)407        LOG.debug("Run container with cmd: %s", cmd)408        result = self._run_async_cmd(cmd, stdin, kwargs.get("name") or "", image_name)409        Util.rm_env_vars_file(env_file)410        return result411    def exec_in_container(412        self,413        container_name_or_id: str,414        command: Union[List[str], str],415        interactive=False,416        detach=False,417        env_vars: Optional[Dict[str, Optional[str]]] = None,418        stdin: Optional[bytes] = None,419        user: Optional[str] = None,420        workdir: Optional[str] = None,421    ) -> Tuple[bytes, bytes]:422        env_file = None423        cmd = self._docker_cmd()424        cmd.append("exec")425        if interactive:426            cmd.append("--interactive")427        if detach:428            cmd.append("--detach")429        if user:430            cmd += ["--user", user]431        if workdir:432            cmd += ["--workdir", workdir]433        if env_vars:434            env_flag, env_file = Util.create_env_vars_file_flag(env_vars)435            cmd += env_flag436        cmd.append(container_name_or_id)437        cmd += command if isinstance(command, List) else [command]438        LOG.debug("Execute in container cmd: %s", cmd)439        result = self._run_async_cmd(cmd, stdin, container_name_or_id)440        Util.rm_env_vars_file(env_file)441        return result442    def start_container(443        self,444        container_name_or_id: str,445        stdin=None,446        interactive: bool = False,447        attach: bool = False,448        flags: Optional[str] = None,449    ) -> Tuple[bytes, bytes]:450        cmd = self._docker_cmd() + ["start"]451        if flags:452            cmd.append(flags)453        if interactive:454            cmd.append("--interactive")455        if attach:456            cmd.append("--attach")457        cmd.append(container_name_or_id)458        LOG.debug("Start container with cmd: %s", cmd)459        return self._run_async_cmd(cmd, stdin, container_name_or_id)460    def _run_async_cmd(461        self, cmd: List[str], stdin: bytes, container_name: str, image_name=None462    ) -> Tuple[bytes, bytes]:463        kwargs = {464            "inherit_env": True,465            "asynchronous": True,466            "stderr": subprocess.PIPE,467            "outfile": self.default_run_outfile or subprocess.PIPE,468        }469        if stdin:470            kwargs["stdin"] = True471        try:472            process = safe_run(cmd, **kwargs)473            stdout, stderr = process.communicate(input=stdin)474            if process.returncode != 0:475                raise subprocess.CalledProcessError(476                    process.returncode,477                    cmd,478                    stdout,479                    stderr,480                )481            else:482                return stdout, stderr483        except subprocess.CalledProcessError as e:484            stderr_str = to_str(e.stderr)485            if "Unable to find image" in stderr_str:486                raise NoSuchImage(image_name or "", stdout=e.stdout, stderr=e.stderr)487            if "No such container" in stderr_str:488                raise NoSuchContainer(container_name, stdout=e.stdout, stderr=e.stderr)489            raise ContainerException(490                "Docker process returned with errorcode %s" % e.returncode, e.stdout, e.stderr491            )492    def _build_run_create_cmd(493        self,494        action: str,495        image_name: str,496        *,497        name: Optional[str] = None,498        entrypoint: Optional[str] = None,499        remove: bool = False,500        interactive: bool = False,501        tty: bool = False,502        detach: bool = False,503        command: Optional[Union[List[str], str]] = None,504        mount_volumes: Optional[List[SimpleVolumeBind]] = None,505        ports: Optional[PortMappings] = None,506        env_vars: Optional[Dict[str, str]] = None,507        user: Optional[str] = None,508        cap_add: Optional[List[str]] = None,509        cap_drop: Optional[List[str]] = None,510        network: Optional[str] = None,511        dns: Optional[str] = None,512        additional_flags: Optional[str] = None,513        workdir: Optional[str] = None,514    ) -> Tuple[List[str], str]:515        env_file = None516        cmd = self._docker_cmd() + [action]517        if remove:518            cmd.append("--rm")519        if name:520            cmd += ["--name", name]521        if entrypoint is not None:  # empty string entrypoint can be intentional522            cmd += ["--entrypoint", entrypoint]523        if mount_volumes:524            cmd += [525                volume526                for host_path, docker_path in dict(mount_volumes).items()527                for volume in ["-v", f"{host_path}:{docker_path}"]528            ]529        if interactive:530            cmd.append("--interactive")...

stack.py

Source:stack.py

...4    pass5class StackManager:6    def __init__(self, sock='unix:///run/docker.sock'):7        self.sock = sock8    def _docker_cmd(self, args, *e_args, parse=True, **kwargs):9        try:10            output = subprocess.check_output(['docker', '--host', self.sock] + args, *e_args, stderr=subprocess.STDOUT,11            encoding='utf-8', **kwargs)12        except subprocess.CalledProcessError as ex:13            raise StackError(f'Docker exited with non-zero exit code {ex.returncode}: {ex.output}')14        lines = filter(lambda l: l, output.split('\n'))15        if parse:16            return map(json.loads, lines)17        return lines18    def ls(self):19        return list(map(lambda s: s['Name'], self._docker_cmd(['stack', 'ls', '--format', '{{json .}}'])))20    def services(self, id_):21        return list(self._docker_cmd(['stack', 'services', '--format', '{{json .}}', id_]))22    def deploy(self, name, spec, prune=False, registry_auth=False):23        args = ['stack', 'deploy', '--compose-file', '-']24        if prune:25            args.append('--prune')26        if registry_auth:27            args.append('--with-registry-auth')28        args.append(name)29        self._docker_cmd(args, input=json.dumps(spec), parse=False)30    def rm(self, name):...

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.