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

How to use untar_to_path method in localstack

Best Python code snippet using localstack_python

docker_utils.py

Source:docker_utils.py

...758            t.add(abs_path, arcname=arcname)759        f.seek(0)760        return f761    @staticmethod762    def untar_to_path(tardata, target_path):763        target_path = Path(target_path)764        with tarfile.open(mode="r", fileobj=io.BytesIO(b"".join(b for b in tardata))) as t:765            if target_path.is_dir():766                t.extractall(path=target_path)767            else:768                member = t.next()769                if member:770                    member.name = target_path.name771                    t.extract(member, target_path.parent)772                else:773                    LOG.debug("File to copy empty, ignoring...")774    @staticmethod775    def parse_additional_flags(776        additional_flags: str,777        env_vars: Dict[str, str] = None,778        ports: PortMappings = None,779        mounts: List[Tuple[str, str]] = None,780        network: Optional[str] = None,781    ) -> Tuple[782        Dict[str, str], PortMappings, List[Tuple[str, str]], Optional[Dict[str, str]], Optional[str]783    ]:784        """Parses environment, volume and port flags passed as string785        :param additional_flags: String which contains the flag definitions786        :param env_vars: Dict with env vars. Will be modified in place.787        :param ports: PortMapping object. Will be modified in place.788        :param mounts: List of mount tuples (host_path, container_path). Will be modified in place.789        :param network: Existing network name (optional). Warning will be printed if network is overwritten in flags.790        :return: A tuple containing the env_vars, ports, mount, extra_hosts and network objects. Will return new objects791                if respective parameters were None and additional flags contained a flag for that object, the same which792                are passed otherwise.793        """794        cur_state = None795        extra_hosts = None796        # TODO Use argparse to simplify this logic797        for flag in shlex.split(additional_flags):798            if not cur_state:799                if flag in ["-v", "--volume"]:800                    cur_state = "volume"801                elif flag in ["-p", "--publish"]:802                    cur_state = "port"803                elif flag in ["-e", "--env"]:804                    cur_state = "env"805                elif flag == "--add-host":806                    cur_state = "add-host"807                elif flag == "--network":808                    cur_state = "set-network"809                else:810                    raise NotImplementedError(811                        f"Flag {flag} is currently not supported by this Docker client."812                    )813            else:814                if cur_state == "volume":815                    mounts = mounts if mounts is not None else []816                    match = re.match(817                        r"(?P<host>[\w\s\\\/:\-.]+?):(?P<container>[\w\s\/\-.]+)(?::(?P<arg>ro|rw|z|Z))?",818                        flag,819                    )820                    if not match:821                        LOG.warning("Unable to parse volume mount Docker flags: %s", flag)822                        continue823                    host_path = match.group("host")824                    container_path = match.group("container")825                    rw_args = match.group("arg")826                    if rw_args:827                        LOG.info("Volume options like :ro or :rw are currently ignored.")828                    mounts.append((host_path, container_path))829                elif cur_state == "port":830                    port_split = flag.split(":")831                    protocol = "tcp"832                    if len(port_split) == 2:833                        host_port, container_port = port_split834                    elif len(port_split) == 3:835                        LOG.warning(836                            "Host part of port mappings are ignored currently in additional flags"837                        )838                        _, host_port, container_port = port_split839                    else:840                        raise ValueError("Invalid port string provided: %s", flag)841                    if "/" in container_port:842                        container_port, protocol = container_port.split("/")843                    ports = ports if ports is not None else PortMappings()844                    ports.add(int(host_port), int(container_port), protocol)845                elif cur_state == "env":846                    lhs, _, rhs = flag.partition("=")847                    env_vars = env_vars if env_vars is not None else {}848                    env_vars[lhs] = rhs849                elif cur_state == "add-host":850                    extra_hosts = extra_hosts if extra_hosts is not None else {}851                    hosts_split = flag.split(":")852                    extra_hosts[hosts_split[0]] = hosts_split[1]853                elif cur_state == "set-network":854                    if network:855                        LOG.warning(856                            "Overwriting Docker container network '%s' with new value '%s'",857                            network,858                            flag,859                        )860                    network = flag861                cur_state = None862        return env_vars, ports, mounts, extra_hosts, network863    @staticmethod864    def convert_mount_list_to_dict(865        mount_volumes: List[Tuple[str, str]]866    ) -> Dict[str, Dict[str, str]]:867        """Converts a List of (host_path, container_path) tuples to a Dict suitable as volume argument for docker sdk"""868        return dict(869            map(870                lambda paths: (str(paths[0]), {"bind": paths[1], "mode": "rw"}),871                mount_volumes,872            )873        )874class SdkDockerClient(ContainerClient):875    """Class for managing docker using the python docker sdk"""876    docker_client: Optional[DockerClient]877    def __init__(self):878        try:879            self.docker_client = docker.from_env()880            logging.getLogger("urllib3").setLevel(logging.INFO)881        except DockerException:882            self.docker_client = None883    def client(self):884        if self.docker_client:885            return self.docker_client886        else:887            raise ContainerException("Docker not available")888    def _read_from_sock(self, sock: socket, tty: bool):889        """Reads multiplexed messages from a socket returned by attach_socket.890        Uses the protocol specified here: https://docs.docker.com/engine/api/v1.41/#operation/ContainerAttach891        """892        stdout = b""893        stderr = b""894        for frame_type, frame_data in frames_iter(sock, tty):895            if frame_type == STDOUT:896                stdout += frame_data897            elif frame_type == STDERR:898                stderr += frame_data899            else:900                raise ContainerException("Invalid frame type when reading from socket")901        return stdout, stderr902    def _container_path_info(self, container: Container, container_path: str):903        """904        Get information about a path in the given container905        :param container: Container to be inspected906        :param container_path: Path in container907        :return: Tuple (path_exists, path_is_directory)908        """909        # Docker CLI copy uses go FileMode to determine if target is a dict or not910        # https://github.com/docker/cli/blob/e3dfc2426e51776a3263cab67fbba753dd3adaa9/cli/command/container/cp.go#L260911        # The isDir Bit is the most significant bit in the 32bit struct:912        # https://golang.org/src/os/types.go?s=2650:2683913        stats = {}914        try:915            _, stats = container.get_archive(container_path)916            target_exists = True917        except APIError:918            target_exists = False919        target_is_dir = target_exists and bool(stats["mode"] & SDK_ISDIR)920        return target_exists, target_is_dir921    def get_container_status(self, container_name: str) -> DockerContainerStatus:922        # LOG.debug("Getting container status for container: %s", container_name) #  too verbose923        try:924            container = self.client().containers.get(container_name)925            if container.status == "running":926                return DockerContainerStatus.UP927            else:928                return DockerContainerStatus.DOWN929        except NotFound:930            return DockerContainerStatus.NON_EXISTENT931        except APIError:932            raise ContainerException()933    def get_network(self, container_name: str) -> str:934        LOG.debug("Getting network type for container: %s", container_name)935        try:936            container = self.client().containers.get(container_name)937            return container.attrs["HostConfig"]["NetworkMode"]938        except NotFound:939            raise NoSuchContainer(container_name)940        except APIError:941            raise ContainerException()942    def stop_container(self, container_name: str) -> None:943        LOG.debug("Stopping container: %s", container_name)944        try:945            container = self.client().containers.get(container_name)946            container.stop(timeout=0)947        except NotFound:948            raise NoSuchContainer(container_name)949        except APIError:950            raise ContainerException()951    def remove_container(self, container_name: str, force=True, check_existence=False) -> None:952        LOG.debug("Removing container: %s", container_name)953        if check_existence and container_name not in self.get_running_container_names():954            LOG.debug("Aborting removing due to check_existence check")955            return956        try:957            container = self.client().containers.get(container_name)958            container.remove(force=force)959        except NotFound:960            if not force:961                raise NoSuchContainer(container_name)962        except APIError:963            raise ContainerException()964    def list_containers(self, filter: Union[List[str], str, None] = None, all=True) -> List[dict]:965        if filter:966            filter = [filter] if isinstance(filter, str) else filter967            filter = dict([f.split("=", 1) for f in filter])968        LOG.debug("Listing containers with filters: %s", filter)969        try:970            container_list = self.client().containers.list(filters=filter, all=all)971            return list(972                map(973                    lambda container: {974                        "id": container.id,975                        "image": container.image,976                        "name": container.name,977                        "status": container.status,978                        "labels": container.labels,979                    },980                    container_list,981                )982            )983        except APIError:984            raise ContainerException()985    def copy_into_container(986        self, container_name: str, local_path: str, container_path: str987    ) -> None:  # TODO behave like https://docs.docker.com/engine/reference/commandline/cp/988        LOG.debug("Copying file %s into %s:%s", local_path, container_name, container_path)989        try:990            container = self.client().containers.get(container_name)991            target_exists, target_isdir = self._container_path_info(container, container_path)992            target_path = container_path if target_isdir else os.path.dirname(container_path)993            with Util.tar_path(local_path, container_path, is_dir=target_isdir) as tar:994                container.put_archive(target_path, tar)995        except NotFound:996            raise NoSuchContainer(container_name)997        except APIError:998            raise ContainerException()999    def copy_from_container(1000        self,1001        container_name: str,1002        local_path: str,1003        container_path: str,1004    ) -> None:1005        LOG.debug("Copying file from %s:%s to %s", container_name, container_path, local_path)1006        try:1007            container = self.client().containers.get(container_name)1008            bits, _ = container.get_archive(container_path)1009            Util.untar_to_path(bits, local_path)1010        except NotFound:1011            raise NoSuchContainer(container_name)1012        except APIError:1013            raise ContainerException()1014    def pull_image(self, docker_image: str) -> None:1015        LOG.debug("Pulling image: %s", docker_image)1016        # some path in the docker image string indicates a custom repository1017        path_split = docker_image.rpartition("/")1018        image_split = path_split[2].partition(":")1019        repository = f"{path_split[0]}{path_split[1]}{image_split[0]}"1020        tag = image_split[2]1021        try:1022            LOG.debug("Repository: %s Tag: %s", repository, tag)1023            self.client().images.pull(repository, tag)...

docker_sdk_client.py

Source:docker_sdk_client.py

...176        LOG.debug("Copying file from %s:%s to %s", container_name, container_path, local_path)177        try:178            container = self.client().containers.get(container_name)179            bits, _ = container.get_archive(container_path)180            Util.untar_to_path(bits, local_path)181        except NotFound:182            raise NoSuchContainer(container_name)183        except APIError:184            raise ContainerException()185    def pull_image(self, docker_image: str) -> None:186        LOG.debug("Pulling Docker image: %s", docker_image)187        # some path in the docker image string indicates a custom repository188        try:189            self.client().images.pull(docker_image)190        except ImageNotFound:191            raise NoSuchImage(docker_image)192        except APIError:193            raise ContainerException()194    def push_image(self, docker_image: str) -> None:...

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.