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

How to use _container_path_info method in localstack

Best Python code snippet using localstack_python

docker_sdk_client.py

Source:docker_sdk_client.py

...53                stderr += frame_data54            else:55                raise ContainerException("Invalid frame type when reading from socket")56        return stdout, stderr57    def _container_path_info(self, container: Container, container_path: str):58        """59        Get information about a path in the given container60        :param container: Container to be inspected61        :param container_path: Path in container62        :return: Tuple (path_exists, path_is_directory)63        """64        # Docker CLI copy uses go FileMode to determine if target is a dict or not65        # https://github.com/docker/cli/blob/e3dfc2426e51776a3263cab67fbba753dd3adaa9/cli/command/container/cp.go#L26066        # The isDir Bit is the most significant bit in the 32bit struct:67        # https://golang.org/src/os/types.go?s=2650:268368        stats = {}69        try:70            _, stats = container.get_archive(container_path)71            target_exists = True72        except APIError:73            target_exists = False74        target_is_dir = target_exists and bool(stats["mode"] & SDK_ISDIR)75        return target_exists, target_is_dir76    def get_container_status(self, container_name: str) -> DockerContainerStatus:77        # LOG.debug("Getting container status for container: %s", container_name) #  too verbose78        try:79            container = self.client().containers.get(container_name)80            if container.status == "running":81                return DockerContainerStatus.UP82            elif container.status == "paused":83                return DockerContainerStatus.PAUSED84            else:85                return DockerContainerStatus.DOWN86        except NotFound:87            return DockerContainerStatus.NON_EXISTENT88        except APIError:89            raise ContainerException()90    def stop_container(self, container_name: str, timeout: int = None) -> None:91        if timeout is None:92            timeout = self.STOP_TIMEOUT93        LOG.debug("Stopping container: %s", container_name)94        try:95            container = self.client().containers.get(container_name)96            container.stop(timeout=timeout)97        except NotFound:98            raise NoSuchContainer(container_name)99        except APIError:100            raise ContainerException()101    def pause_container(self, container_name: str) -> None:102        LOG.debug("Pausing container: %s", container_name)103        try:104            container = self.client().containers.get(container_name)105            container.pause()106        except NotFound:107            raise NoSuchContainer(container_name)108        except APIError:109            raise ContainerException()110    def unpause_container(self, container_name: str) -> None:111        LOG.debug("Unpausing container: %s", container_name)112        try:113            container = self.client().containers.get(container_name)114            container.unpause()115        except NotFound:116            raise NoSuchContainer(container_name)117        except APIError:118            raise ContainerException()119    def remove_container(self, container_name: str, force=True, check_existence=False) -> None:120        LOG.debug("Removing container: %s", container_name)121        if check_existence and container_name not in self.get_running_container_names():122            LOG.debug("Aborting removing due to check_existence check")123            return124        try:125            container = self.client().containers.get(container_name)126            container.remove(force=force)127        except NotFound:128            if not force:129                raise NoSuchContainer(container_name)130        except APIError:131            raise ContainerException()132    def list_containers(self, filter: Union[List[str], str, None] = None, all=True) -> List[dict]:133        if filter:134            filter = [filter] if isinstance(filter, str) else filter135            filter = dict([f.split("=", 1) for f in filter])136        LOG.debug("Listing containers with filters: %s", filter)137        try:138            container_list = self.client().containers.list(filters=filter, all=all)139            result = []140            for container in container_list:141                try:142                    result.append(143                        {144                            "id": container.id,145                            "image": container.image,146                            "name": container.name,147                            "status": container.status,148                            "labels": container.labels,149                        }150                    )151                except Exception as e:152                    LOG.error(f"Error checking container {container}: {e}")153            return result154        except APIError:155            raise ContainerException()156    def copy_into_container(157        self, container_name: str, local_path: str, container_path: str158    ) -> None:  # TODO behave like https://docs.docker.com/engine/reference/commandline/cp/159        LOG.debug("Copying file %s into %s:%s", local_path, container_name, container_path)160        try:161            container = self.client().containers.get(container_name)162            target_exists, target_isdir = self._container_path_info(container, container_path)163            target_path = container_path if target_isdir else os.path.dirname(container_path)164            with Util.tar_path(local_path, container_path, is_dir=target_isdir) as tar:165                container.put_archive(target_path, tar)166        except NotFound:167            raise NoSuchContainer(container_name)168        except APIError:169            raise ContainerException()170    def copy_from_container(171        self,172        container_name: str,173        local_path: str,174        container_path: str,175    ) -> None:176        LOG.debug("Copying file from %s:%s to %s", container_name, container_path, local_path)...

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.