Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

parse_host function failing to parse paths. #3310

Closed
lucazz opened this issue Feb 4, 2025 · 1 comment
Closed

parse_host function failing to parse paths. #3310

lucazz opened this issue Feb 4, 2025 · 1 comment

Comments

@lucazz
Copy link

lucazz commented Feb 4, 2025

Hello there everyone, I'm instrumenting a pytest suite to run end-to-end tests against our repository manager and one of the helper functions I have in place is this one, that logs in and attempts to pull a container image:

    def pull_docker_image(self, image) -> bool:
        """
        Pull a Docker image from Nexus repository
        Args:
            image: Full image name including registry, repository and tag
        Returns:
            bool: True if pull successful, False otherwise
        Raises:
            Exception: If image pull fails.
        """
        try:
            registry_url = self.base_url.replace("https://", "")
            home_directory = os.environ.get("HOME")
            docker_socket_path_list = [
                Path(f"unix:///{home_directory}/.orbstack/run/docker.sock"),  # Orbstack
                Path("unix:///run/docker.sock"),  # Linux default
                Path("unix:///var/run/docker.sock"),  # macOS default
            ]
            for docker_socket_path in docker_socket_path_list:
                if docker_socket_path.is_socket():
                    break
            client = docker.DockerClient(base_url=docker_socket_path)
            client.login(
                username=self.username, password=self.password, registry=registry_url
            )
            client.images.pull(image)
            return True
        except Exception as e:
                raise Exception(f"Unexpected error pulling image {image}: {str(e)}")

But it fails to do so, here's the error that pytest bubbles up:

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../.venv/foo/lib/python3.12/site-packages/docker/client.py:45: in __init__
    self.api = APIClient(*args, **kwargs)
../../../.venv/foo/lib/python3.12/site-packages/docker/api/client.py:146: in __init__
    base_url = utils.parse_host(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

addr = PosixPath('unix:/var/run/docker.sock'), is_win32 = False, tls = False

    def parse_host(addr, is_win32=False, tls=False):
        # Sensible defaults
        if not addr and is_win32:
            return DEFAULT_NPIPE
>       if not addr or addr.strip() == 'unix://':
E       AttributeError: 'PosixPath' object has no attribute 'strip'

../../../.venv/foo/lib/python3.12/site-packages/docker/utils/utils.py:237: AttributeError

I'm using docker==7.1.0, btw.
What am I missing here?

@lucazz
Copy link
Author

lucazz commented Feb 5, 2025

Turns out this was an implementation error.
I've broken down this function into a few other functions and it worked.
For future reference, this is what I ended up doing:

    def _get_docker_socket_path(self) -> Optional[str]:
        """Find the first available Docker socket path.
        Returns:
            Optional[Path]: Path to Docker socket if found, None otherwise
        """
        try:
            system = platform.system().lower()
            is_mac = system == "darwin"
            is_orbstack = False
            home_directory = os.environ.get("HOME")
            if is_mac:
                try:
                    subprocess.run(
                        ["orbctl", "version"],
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                        check=True,
                    )
                    is_orbstack = True
                except (subprocess.CalledProcessError, FileNotFoundError):
                    is_orbstack = False
            docker_socket = "unix://var/run/docker.sock"
            if is_orbstack:
                docker_socket = f"unix:///{home_directory}/.orbstack/run/docker.sock"
            return docker_socket
        except Exception as e:
            raise Exception(f"No valid Docker socket found: {str(e)}")

    def _setup_docker_client(self) -> docker.DockerClient:
        try:
            socket_path = self._get_docker_socket_path()
            client = docker.DockerClient(base_url=socket_path)
            return client
        except Exception as e:
            raise Exception(f"Failed to setup docker client: {str(e)}")

    def pull_docker_image(self, image) -> bool:
        """
        Pull a Docker image from Nexus repository
        Args:
            image (str): The container image reference (e.g. "myapp:1.0").
        Returns:
            bool: True if pull successful, False otherwise
        """
        registry = self.base_url.replace("https://", "")
        client = self._setup_docker_client()
        try:
            client.login(
                username=self.username, password=self.password, registry=registry
            )
            client.images.pull(image)
            return True
        except Exception as e:
            raise Exception(f"Failed to pull {image} image: {str(e)}")

@lucazz lucazz closed this as completed Feb 5, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant