diff --git a/docker/constants.py b/docker/constants.py index 0e39dc2917..25bb82d8a8 100644 --- a/docker/constants.py +++ b/docker/constants.py @@ -1,3 +1,4 @@ +import os import sys from .version import __version__ @@ -11,7 +12,26 @@ ] DEFAULT_HTTP_HOST = "127.0.0.1" -DEFAULT_UNIX_SOCKET = "http+unix:///var/run/docker.sock" + +# Potential Unix socket locations in order of preference +UNIX_SOCKET_PATHS = [ + '/var/run/docker.sock', # Traditional Linux/macOS location + os.path.expanduser('~/.docker/run/docker.sock'), # Docker Desktop v4.x+ on macOS + os.path.expanduser('~/.docker/desktop/docker.sock'), # Older Docker Desktop location +] + + +def _find_available_unix_socket(): + """Find the first available Docker socket from known locations.""" + for path in UNIX_SOCKET_PATHS: + if os.path.exists(path): + return f"http+unix://{path}" + # Fallback to traditional location even if it doesn't exist + return f"http+unix://{UNIX_SOCKET_PATHS[0]}" + + +# Dynamic default socket - checks multiple locations +DEFAULT_UNIX_SOCKET = _find_available_unix_socket() DEFAULT_NPIPE = 'npipe:////./pipe/docker_engine' BYTE_UNITS = { diff --git a/tests/unit/client_test.py b/tests/unit/client_test.py index 5ba712d240..b01763d0c6 100644 --- a/tests/unit/client_test.py +++ b/tests/unit/client_test.py @@ -10,6 +10,7 @@ DEFAULT_DOCKER_API_VERSION, DEFAULT_MAX_POOL_SIZE, DEFAULT_TIMEOUT_SECONDS, + DEFAULT_UNIX_SOCKET, IS_WINDOWS_PLATFORM, ) from docker.utils import kwargs_from_env @@ -89,9 +90,11 @@ def test_default_pool_size_unix(self, mock_obj): client.ping() base_url = f"{client.api.base_url}/v{client.api._version}/_ping" + # Extract socket path from DEFAULT_UNIX_SOCKET (remove http+unix:// prefix) + socket_path = DEFAULT_UNIX_SOCKET.replace('http+unix://', '') mock_obj.assert_called_once_with(base_url, - "/var/run/docker.sock", + socket_path, 60, maxsize=DEFAULT_MAX_POOL_SIZE ) @@ -125,9 +128,11 @@ def test_pool_size_unix(self, mock_obj): client.ping() base_url = f"{client.api.base_url}/v{client.api._version}/_ping" + # Extract socket path from DEFAULT_UNIX_SOCKET (remove http+unix:// prefix) + socket_path = DEFAULT_UNIX_SOCKET.replace('http+unix://', '') mock_obj.assert_called_once_with(base_url, - "/var/run/docker.sock", + socket_path, 60, maxsize=POOL_SIZE ) @@ -198,9 +203,11 @@ def test_default_pool_size_from_env_unix(self, mock_obj): client.ping() base_url = f"{client.api.base_url}/v{client.api._version}/_ping" + # Extract socket path from DEFAULT_UNIX_SOCKET (remove http+unix:// prefix) + socket_path = DEFAULT_UNIX_SOCKET.replace('http+unix://', '') mock_obj.assert_called_once_with(base_url, - "/var/run/docker.sock", + socket_path, 60, maxsize=DEFAULT_MAX_POOL_SIZE ) @@ -233,9 +240,11 @@ def test_pool_size_from_env_unix(self, mock_obj): client.ping() base_url = f"{client.api.base_url}/v{client.api._version}/_ping" + # Extract socket path from DEFAULT_UNIX_SOCKET (remove http+unix:// prefix) + socket_path = DEFAULT_UNIX_SOCKET.replace('http+unix://', '') mock_obj.assert_called_once_with(base_url, - "/var/run/docker.sock", + socket_path, 60, maxsize=POOL_SIZE ) diff --git a/tests/unit/utils_test.py b/tests/unit/utils_test.py index 21da0b58e8..b99245621d 100644 --- a/tests/unit/utils_test.py +++ b/tests/unit/utils_test.py @@ -5,11 +5,17 @@ import shutil import tempfile import unittest +from unittest import mock import pytest from docker.api.client import APIClient -from docker.constants import DEFAULT_DOCKER_API_VERSION, IS_WINDOWS_PLATFORM +from docker.constants import ( + DEFAULT_DOCKER_API_VERSION, + IS_WINDOWS_PLATFORM, + UNIX_SOCKET_PATHS, +) +from docker.constants import _find_available_unix_socket from docker.errors import DockerException from docker.utils import ( compare_version, @@ -32,14 +38,14 @@ TEST_CERT_DIR = os.path.join( os.path.dirname(__file__), - 'testdata/certs', + "testdata/certs", ) class DecoratorsTest(unittest.TestCase): def test_update_headers(self): sample_headers = { - 'X-Docker-Locale': 'en-US', + "X-Docker-Locale": "en-US", } def f(self, headers=None): @@ -51,19 +57,17 @@ def f(self, headers=None): g = update_headers(f) assert g(client, headers=None) is None assert g(client, headers={}) == {} - assert g(client, headers={'Content-type': 'application/json'}) == { - 'Content-type': 'application/json', + assert g(client, headers={"Content-type": "application/json"}) == { + "Content-type": "application/json", } - client._general_configs = { - 'HttpHeaders': sample_headers - } + client._general_configs = {"HttpHeaders": sample_headers} assert g(client, headers=None) == sample_headers assert g(client, headers={}) == sample_headers - assert g(client, headers={'Content-type': 'application/json'}) == { - 'Content-type': 'application/json', - 'X-Docker-Locale': 'en-US', + assert g(client, headers={"Content-type": "application/json"}) == { + "Content-type": "application/json", + "X-Docker-Locale": "en-US", } @@ -76,82 +80,83 @@ def tearDown(self): os.environ.update(self.os_environ) def test_kwargs_from_env_empty(self): - os.environ.update(DOCKER_HOST='', - DOCKER_CERT_PATH='') - os.environ.pop('DOCKER_TLS_VERIFY', None) + os.environ.update(DOCKER_HOST="", DOCKER_CERT_PATH="") + os.environ.pop("DOCKER_TLS_VERIFY", None) kwargs = kwargs_from_env() - assert kwargs.get('base_url') is None - assert kwargs.get('tls') is None + assert kwargs.get("base_url") is None + assert kwargs.get("tls") is None def test_kwargs_from_env_tls(self): - os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376', - DOCKER_CERT_PATH=TEST_CERT_DIR, - DOCKER_TLS_VERIFY='1') + os.environ.update( + DOCKER_HOST="tcp://192.168.59.103:2376", + DOCKER_CERT_PATH=TEST_CERT_DIR, + DOCKER_TLS_VERIFY="1", + ) kwargs = kwargs_from_env() - assert 'tcp://192.168.59.103:2376' == kwargs['base_url'] - assert 'ca.pem' in kwargs['tls'].ca_cert - assert 'cert.pem' in kwargs['tls'].cert[0] - assert 'key.pem' in kwargs['tls'].cert[1] - assert kwargs['tls'].verify is True - - parsed_host = parse_host(kwargs['base_url'], IS_WINDOWS_PLATFORM, True) - kwargs['version'] = DEFAULT_DOCKER_API_VERSION + assert "tcp://192.168.59.103:2376" == kwargs["base_url"] + assert "ca.pem" in kwargs["tls"].ca_cert + assert "cert.pem" in kwargs["tls"].cert[0] + assert "key.pem" in kwargs["tls"].cert[1] + assert kwargs["tls"].verify is True + + parsed_host = parse_host(kwargs["base_url"], IS_WINDOWS_PLATFORM, True) + kwargs["version"] = DEFAULT_DOCKER_API_VERSION try: client = APIClient(**kwargs) assert parsed_host == client.base_url - assert kwargs['tls'].ca_cert == client.verify - assert kwargs['tls'].cert == client.cert + assert kwargs["tls"].ca_cert == client.verify + assert kwargs["tls"].cert == client.cert except TypeError as e: self.fail(e) def test_kwargs_from_env_tls_verify_false(self): - os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376', - DOCKER_CERT_PATH=TEST_CERT_DIR, - DOCKER_TLS_VERIFY='') + os.environ.update( + DOCKER_HOST="tcp://192.168.59.103:2376", + DOCKER_CERT_PATH=TEST_CERT_DIR, + DOCKER_TLS_VERIFY="", + ) kwargs = kwargs_from_env() - assert 'tcp://192.168.59.103:2376' == kwargs['base_url'] - assert 'ca.pem' in kwargs['tls'].ca_cert - assert 'cert.pem' in kwargs['tls'].cert[0] - assert 'key.pem' in kwargs['tls'].cert[1] - assert kwargs['tls'].verify is False - parsed_host = parse_host(kwargs['base_url'], IS_WINDOWS_PLATFORM, True) - kwargs['version'] = DEFAULT_DOCKER_API_VERSION + assert "tcp://192.168.59.103:2376" == kwargs["base_url"] + assert "ca.pem" in kwargs["tls"].ca_cert + assert "cert.pem" in kwargs["tls"].cert[0] + assert "key.pem" in kwargs["tls"].cert[1] + assert kwargs["tls"].verify is False + parsed_host = parse_host(kwargs["base_url"], IS_WINDOWS_PLATFORM, True) + kwargs["version"] = DEFAULT_DOCKER_API_VERSION try: client = APIClient(**kwargs) assert parsed_host == client.base_url - assert kwargs['tls'].cert == client.cert - assert not kwargs['tls'].verify + assert kwargs["tls"].cert == client.cert + assert not kwargs["tls"].verify except TypeError as e: self.fail(e) def test_kwargs_from_env_tls_verify_false_no_cert(self): temp_dir = tempfile.mkdtemp() - cert_dir = os.path.join(temp_dir, '.docker') + cert_dir = os.path.join(temp_dir, ".docker") shutil.copytree(TEST_CERT_DIR, cert_dir) - os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376', - HOME=temp_dir, - DOCKER_TLS_VERIFY='') - os.environ.pop('DOCKER_CERT_PATH', None) + os.environ.update( + DOCKER_HOST="tcp://192.168.59.103:2376", HOME=temp_dir, DOCKER_TLS_VERIFY="" + ) + os.environ.pop("DOCKER_CERT_PATH", None) kwargs = kwargs_from_env() - assert 'tcp://192.168.59.103:2376' == kwargs['base_url'] + assert "tcp://192.168.59.103:2376" == kwargs["base_url"] def test_kwargs_from_env_no_cert_path(self): temp_dir = tempfile.mkdtemp() try: - cert_dir = os.path.join(temp_dir, '.docker') + cert_dir = os.path.join(temp_dir, ".docker") shutil.copytree(TEST_CERT_DIR, cert_dir) - os.environ.update(HOME=temp_dir, - DOCKER_CERT_PATH='', - DOCKER_TLS_VERIFY='1') + os.environ.update(HOME=temp_dir, DOCKER_CERT_PATH="", DOCKER_TLS_VERIFY="1") kwargs = kwargs_from_env() - assert kwargs['tls'].verify - assert cert_dir in kwargs['tls'].ca_cert - assert cert_dir in kwargs['tls'].cert[0] - assert cert_dir in kwargs['tls'].cert[1] + assert kwargs["tls"].verify + assert cert_dir in kwargs["tls"].ca_cert + assert cert_dir in kwargs["tls"].cert[0] + assert cert_dir in kwargs["tls"].cert[1] finally: shutil.rmtree(temp_dir) @@ -159,15 +164,17 @@ def test_kwargs_from_env_alternate_env(self): # Values in os.environ are entirely ignored if an alternate is # provided os.environ.update( - DOCKER_HOST='tcp://192.168.59.103:2376', + DOCKER_HOST="tcp://192.168.59.103:2376", DOCKER_CERT_PATH=TEST_CERT_DIR, - DOCKER_TLS_VERIFY='' + DOCKER_TLS_VERIFY="", ) - kwargs = kwargs_from_env(environment={ - 'DOCKER_HOST': 'http://docker.gensokyo.jp:2581', - }) - assert 'http://docker.gensokyo.jp:2581' == kwargs['base_url'] - assert 'tls' not in kwargs + kwargs = kwargs_from_env( + environment={ + "DOCKER_HOST": "http://docker.gensokyo.jp:2581", + } + ) + assert "http://docker.gensokyo.jp:2581" == kwargs["base_url"] + assert "tls" not in kwargs class ConverVolumeBindsTest(unittest.TestCase): @@ -176,52 +183,31 @@ def test_convert_volume_binds_empty(self): assert convert_volume_binds([]) == [] def test_convert_volume_binds_list(self): - data = ['/a:/a:ro', '/b:/c:z'] + data = ["/a:/a:ro", "/b:/c:z"] assert convert_volume_binds(data) == data def test_convert_volume_binds_complete(self): - data = { - '/mnt/vol1': { - 'bind': '/data', - 'mode': 'ro' - } - } - assert convert_volume_binds(data) == ['/mnt/vol1:/data:ro'] + data = {"/mnt/vol1": {"bind": "/data", "mode": "ro"}} + assert convert_volume_binds(data) == ["/mnt/vol1:/data:ro"] def test_convert_volume_binds_compact(self): - data = { - '/mnt/vol1': '/data' - } - assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw'] + data = {"/mnt/vol1": "/data"} + assert convert_volume_binds(data) == ["/mnt/vol1:/data:rw"] def test_convert_volume_binds_no_mode(self): - data = { - '/mnt/vol1': { - 'bind': '/data' - } - } - assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw'] + data = {"/mnt/vol1": {"bind": "/data"}} + assert convert_volume_binds(data) == ["/mnt/vol1:/data:rw"] def test_convert_volume_binds_unicode_bytes_input(self): - expected = ['/mnt/지연:/unicode/박:rw'] + expected = ["/mnt/지연:/unicode/박:rw"] - data = { - '/mnt/지연'.encode(): { - 'bind': '/unicode/박'.encode(), - 'mode': 'rw' - } - } + data = {"/mnt/지연".encode(): {"bind": "/unicode/박".encode(), "mode": "rw"}} assert convert_volume_binds(data) == expected def test_convert_volume_binds_unicode_unicode_input(self): - expected = ['/mnt/지연:/unicode/박:rw'] + expected = ["/mnt/지연:/unicode/박:rw"] - data = { - '/mnt/지연': { - 'bind': '/unicode/박', - 'mode': 'rw' - } - } + data = {"/mnt/지연": {"bind": "/unicode/박", "mode": "rw"}} assert convert_volume_binds(data) == expected @@ -233,41 +219,36 @@ def generate_tempfile(self, file_content=None): Don't forget to unlink the file with os.unlink() after. """ local_tempfile = tempfile.NamedTemporaryFile(delete=False) - local_tempfile.write(file_content.encode('UTF-8')) + local_tempfile.write(file_content.encode("UTF-8")) local_tempfile.close() return local_tempfile.name def test_parse_env_file_proper(self): - env_file = self.generate_tempfile( - file_content='USER=jdoe\nPASS=secret') + env_file = self.generate_tempfile(file_content="USER=jdoe\nPASS=secret") get_parse_env_file = parse_env_file(env_file) - assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'secret'} + assert get_parse_env_file == {"USER": "jdoe", "PASS": "secret"} os.unlink(env_file) def test_parse_env_file_with_equals_character(self): - env_file = self.generate_tempfile( - file_content='USER=jdoe\nPASS=sec==ret') + env_file = self.generate_tempfile(file_content="USER=jdoe\nPASS=sec==ret") get_parse_env_file = parse_env_file(env_file) - assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'sec==ret'} + assert get_parse_env_file == {"USER": "jdoe", "PASS": "sec==ret"} os.unlink(env_file) def test_parse_env_file_commented_line(self): - env_file = self.generate_tempfile( - file_content='USER=jdoe\n#PASS=secret') + env_file = self.generate_tempfile(file_content="USER=jdoe\n#PASS=secret") get_parse_env_file = parse_env_file(env_file) - assert get_parse_env_file == {'USER': 'jdoe'} + assert get_parse_env_file == {"USER": "jdoe"} os.unlink(env_file) def test_parse_env_file_newline(self): - env_file = self.generate_tempfile( - file_content='\nUSER=jdoe\n\n\nPASS=secret') + env_file = self.generate_tempfile(file_content="\nUSER=jdoe\n\n\nPASS=secret") get_parse_env_file = parse_env_file(env_file) - assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'secret'} + assert get_parse_env_file == {"USER": "jdoe", "PASS": "secret"} os.unlink(env_file) def test_parse_env_file_invalid_line(self): - env_file = self.generate_tempfile( - file_content='USER jdoe') + env_file = self.generate_tempfile(file_content="USER jdoe") with pytest.raises(DockerException): parse_env_file(env_file) os.unlink(env_file) @@ -276,46 +257,43 @@ def test_parse_env_file_invalid_line(self): class ParseHostTest(unittest.TestCase): def test_parse_host(self): invalid_hosts = [ - '0.0.0.0', - 'tcp://', - 'udp://127.0.0.1', - 'udp://127.0.0.1:2375', - 'ssh://:22/path', - 'tcp://netloc:3333/path?q=1', - 'unix:///sock/path#fragment', - 'https://netloc:3333/path;params', - 'ssh://:clearpassword@host:22', + "0.0.0.0", + "tcp://", + "udp://127.0.0.1", + "udp://127.0.0.1:2375", + "ssh://:22/path", + "tcp://netloc:3333/path?q=1", + "unix:///sock/path#fragment", + "https://netloc:3333/path;params", + "ssh://:clearpassword@host:22", ] valid_hosts = { - '0.0.0.1:5555': 'http://0.0.0.1:5555', - ':6666': 'http://127.0.0.1:6666', - 'tcp://:7777': 'http://127.0.0.1:7777', - 'http://:7777': 'http://127.0.0.1:7777', - 'https://kokia.jp:2375': 'https://kokia.jp:2375', - 'unix:///var/run/docker.sock': 'http+unix:///var/run/docker.sock', - 'unix://': 'http+unix:///var/run/docker.sock', - '12.234.45.127:2375/docker/engine': ( - 'http://12.234.45.127:2375/docker/engine' + "0.0.0.1:5555": "http://0.0.0.1:5555", + ":6666": "http://127.0.0.1:6666", + "tcp://:7777": "http://127.0.0.1:7777", + "http://:7777": "http://127.0.0.1:7777", + "https://kokia.jp:2375": "https://kokia.jp:2375", + "unix:///var/run/docker.sock": "http+unix:///var/run/docker.sock", + "12.234.45.127:2375/docker/engine": ( + "http://12.234.45.127:2375/docker/engine" ), - 'somehost.net:80/service/swarm': ( - 'http://somehost.net:80/service/swarm' + "somehost.net:80/service/swarm": ("http://somehost.net:80/service/swarm"), + "npipe:////./pipe/docker_engine": "npipe:////./pipe/docker_engine", + "[fd12::82d1]:2375": "http://[fd12::82d1]:2375", + "https://[fd12:5672::12aa]:1090": "https://[fd12:5672::12aa]:1090", + "[fd12::82d1]:2375/docker/engine": ( + "http://[fd12::82d1]:2375/docker/engine" ), - 'npipe:////./pipe/docker_engine': 'npipe:////./pipe/docker_engine', - '[fd12::82d1]:2375': 'http://[fd12::82d1]:2375', - 'https://[fd12:5672::12aa]:1090': 'https://[fd12:5672::12aa]:1090', - '[fd12::82d1]:2375/docker/engine': ( - 'http://[fd12::82d1]:2375/docker/engine' - ), - 'ssh://[fd12::82d1]': 'ssh://[fd12::82d1]:22', - 'ssh://user@[fd12::82d1]:8765': 'ssh://user@[fd12::82d1]:8765', - 'ssh://': 'ssh://127.0.0.1:22', - 'ssh://user@localhost:22': 'ssh://user@localhost:22', - 'ssh://user@remote': 'ssh://user@remote:22', + "ssh://[fd12::82d1]": "ssh://[fd12::82d1]:22", + "ssh://user@[fd12::82d1]:8765": "ssh://user@[fd12::82d1]:8765", + "ssh://": "ssh://127.0.0.1:22", + "ssh://user@localhost:22": "ssh://user@localhost:22", + "ssh://user@remote": "ssh://user@remote:22", } for host in invalid_hosts: - msg = f'Should have failed to parse invalid host: {host}' + msg = f"Should have failed to parse invalid host: {host}" with self.assertRaises(DockerException, msg=msg): parse_host(host, None) @@ -323,35 +301,42 @@ def test_parse_host(self): self.assertEqual( parse_host(host, None), expected, - msg=f'Failed to parse valid host: {host}', + msg=f"Failed to parse valid host: {host}", ) def test_parse_host_empty_value(self): - unix_socket = 'http+unix:///var/run/docker.sock' - npipe = 'npipe:////./pipe/docker_engine' + from docker.constants import DEFAULT_UNIX_SOCKET + + npipe = "npipe:////./pipe/docker_engine" - for val in [None, '']: - assert parse_host(val, is_win32=False) == unix_socket + for val in [None, ""]: + assert parse_host(val, is_win32=False) == DEFAULT_UNIX_SOCKET assert parse_host(val, is_win32=True) == npipe + def test_parse_host_unix_empty_scheme(self): + """Test that 'unix://' falls back to DEFAULT_UNIX_SOCKET.""" + from docker.constants import DEFAULT_UNIX_SOCKET + + assert parse_host("unix://") == DEFAULT_UNIX_SOCKET + def test_parse_host_tls(self): - host_value = 'myhost.docker.net:3348' - expected_result = 'https://myhost.docker.net:3348' + host_value = "myhost.docker.net:3348" + expected_result = "https://myhost.docker.net:3348" assert parse_host(host_value, tls=True) == expected_result def test_parse_host_tls_tcp_proto(self): - host_value = 'tcp://myhost.docker.net:3348' - expected_result = 'https://myhost.docker.net:3348' + host_value = "tcp://myhost.docker.net:3348" + expected_result = "https://myhost.docker.net:3348" assert parse_host(host_value, tls=True) == expected_result def test_parse_host_trailing_slash(self): - host_value = 'tcp://myhost.docker.net:2376/' - expected_result = 'http://myhost.docker.net:2376' + host_value = "tcp://myhost.docker.net:2376/" + expected_result = "http://myhost.docker.net:2376" assert parse_host(host_value) == expected_result class ParseRepositoryTagTest(unittest.TestCase): - sha = 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855' + sha = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" def test_index_image_no_tag(self): assert parse_repository_tag("root") == ("root", None) @@ -369,77 +354,83 @@ def test_private_reg_image_no_tag(self): assert parse_repository_tag("url:5000/repo") == ("url:5000/repo", None) def test_private_reg_image_tag(self): - assert parse_repository_tag("url:5000/repo:tag") == ( - "url:5000/repo", "tag" - ) + assert parse_repository_tag("url:5000/repo:tag") == ("url:5000/repo", "tag") def test_index_image_sha(self): assert parse_repository_tag(f"root@sha256:{self.sha}") == ( - "root", f"sha256:{self.sha}" + "root", + f"sha256:{self.sha}", ) def test_private_reg_image_sha(self): - assert parse_repository_tag( - f"url:5000/repo@sha256:{self.sha}" - ) == ("url:5000/repo", f"sha256:{self.sha}") + assert parse_repository_tag(f"url:5000/repo@sha256:{self.sha}") == ( + "url:5000/repo", + f"sha256:{self.sha}", + ) class ParseDeviceTest(unittest.TestCase): def test_dict(self): - devices = parse_devices([{ - 'PathOnHost': '/dev/sda1', - 'PathInContainer': '/dev/mnt1', - 'CgroupPermissions': 'r' - }]) + devices = parse_devices( + [ + { + "PathOnHost": "/dev/sda1", + "PathInContainer": "/dev/mnt1", + "CgroupPermissions": "r", + } + ] + ) assert devices[0] == { - 'PathOnHost': '/dev/sda1', - 'PathInContainer': '/dev/mnt1', - 'CgroupPermissions': 'r' + "PathOnHost": "/dev/sda1", + "PathInContainer": "/dev/mnt1", + "CgroupPermissions": "r", } def test_partial_string_definition(self): - devices = parse_devices(['/dev/sda1']) + devices = parse_devices(["/dev/sda1"]) assert devices[0] == { - 'PathOnHost': '/dev/sda1', - 'PathInContainer': '/dev/sda1', - 'CgroupPermissions': 'rwm' + "PathOnHost": "/dev/sda1", + "PathInContainer": "/dev/sda1", + "CgroupPermissions": "rwm", } def test_permissionless_string_definition(self): - devices = parse_devices(['/dev/sda1:/dev/mnt1']) + devices = parse_devices(["/dev/sda1:/dev/mnt1"]) assert devices[0] == { - 'PathOnHost': '/dev/sda1', - 'PathInContainer': '/dev/mnt1', - 'CgroupPermissions': 'rwm' + "PathOnHost": "/dev/sda1", + "PathInContainer": "/dev/mnt1", + "CgroupPermissions": "rwm", } def test_full_string_definition(self): - devices = parse_devices(['/dev/sda1:/dev/mnt1:r']) + devices = parse_devices(["/dev/sda1:/dev/mnt1:r"]) assert devices[0] == { - 'PathOnHost': '/dev/sda1', - 'PathInContainer': '/dev/mnt1', - 'CgroupPermissions': 'r' + "PathOnHost": "/dev/sda1", + "PathInContainer": "/dev/mnt1", + "CgroupPermissions": "r", } def test_hybrid_list(self): - devices = parse_devices([ - '/dev/sda1:/dev/mnt1:rw', - { - 'PathOnHost': '/dev/sda2', - 'PathInContainer': '/dev/mnt2', - 'CgroupPermissions': 'r' - } - ]) + devices = parse_devices( + [ + "/dev/sda1:/dev/mnt1:rw", + { + "PathOnHost": "/dev/sda2", + "PathInContainer": "/dev/mnt2", + "CgroupPermissions": "r", + }, + ] + ) assert devices[0] == { - 'PathOnHost': '/dev/sda1', - 'PathInContainer': '/dev/mnt1', - 'CgroupPermissions': 'rw' + "PathOnHost": "/dev/sda1", + "PathInContainer": "/dev/mnt1", + "CgroupPermissions": "rw", } assert devices[1] == { - 'PathOnHost': '/dev/sda2', - 'PathInContainer': '/dev/mnt2', - 'CgroupPermissions': 'r' + "PathOnHost": "/dev/sda2", + "PathInContainer": "/dev/mnt2", + "CgroupPermissions": "r", } @@ -466,26 +457,26 @@ class UtilsTest(unittest.TestCase): def test_convert_filters(self): tests = [ - ({'dangling': True}, '{"dangling": ["true"]}'), - ({'dangling': "true"}, '{"dangling": ["true"]}'), - ({'exited': 0}, '{"exited": ["0"]}'), - ({'exited': [0, 1]}, '{"exited": ["0", "1"]}'), + ({"dangling": True}, '{"dangling": ["true"]}'), + ({"dangling": "true"}, '{"dangling": ["true"]}'), + ({"exited": 0}, '{"exited": ["0"]}'), + ({"exited": [0, 1]}, '{"exited": ["0", "1"]}'), ] for filters, expected in tests: assert convert_filters(filters) == expected def test_decode_json_header(self): - obj = {'a': 'b', 'c': 1} + obj = {"a": "b", "c": 1} data = None - data = base64.urlsafe_b64encode(bytes(json.dumps(obj), 'utf-8')) + data = base64.urlsafe_b64encode(bytes(json.dumps(obj), "utf-8")) decoded_data = decode_json_header(data) assert obj == decoded_data class SplitCommandTest(unittest.TestCase): def test_split_command_with_unicode(self): - assert split_command('echo μμ') == ['echo', 'μμ'] + assert split_command("echo μμ") == ["echo", "μμ"] class PortsTest(unittest.TestCase): @@ -495,10 +486,8 @@ def test_split_port_with_host_ip(self): assert external_port == [("127.0.0.1", "1000")] def test_split_port_with_protocol(self): - for protocol in ['tcp', 'udp', 'sctp']: - internal_port, external_port = split_port( - f"127.0.0.1:1000:2000/{protocol}" - ) + for protocol in ["tcp", "udp", "sctp"]: + internal_port, external_port = split_port(f"127.0.0.1:1000:2000/{protocol}") assert internal_port == [f"2000/{protocol}"] assert external_port == [("127.0.0.1", "1000")] @@ -538,20 +527,17 @@ def test_split_port_range_no_host_port(self): assert external_port is None def test_split_port_range_with_protocol(self): - internal_port, external_port = split_port( - "127.0.0.1:1000-1001:2000-2001/udp") + internal_port, external_port = split_port("127.0.0.1:1000-1001:2000-2001/udp") assert internal_port == ["2000/udp", "2001/udp"] assert external_port == [("127.0.0.1", "1000"), ("127.0.0.1", "1001")] def test_split_port_with_ipv6_address(self): - internal_port, external_port = split_port( - "2001:abcd:ef00::2:1000:2000") + internal_port, external_port = split_port("2001:abcd:ef00::2:1000:2000") assert internal_port == ["2000"] assert external_port == [("2001:abcd:ef00::2", "1000")] def test_split_port_with_ipv6_square_brackets_address(self): - internal_port, external_port = split_port( - "[2001:abcd:ef00::2]:1000:2000") + internal_port, external_port = split_port("[2001:abcd:ef00::2]:1000:2000") assert internal_port == ["2000"] assert external_port == [("2001:abcd:ef00::2", "1000")] @@ -588,7 +574,7 @@ def test_split_port_empty_string(self): split_port("") def test_split_port_non_string(self): - assert split_port(1243) == (['1243'], None) + assert split_port(1243) == (["1243"], None) def test_build_port_bindings_with_one_port(self): port_bindings = build_port_bindings(["127.0.0.1:1000:1000"]) @@ -596,14 +582,14 @@ def test_build_port_bindings_with_one_port(self): def test_build_port_bindings_with_matching_internal_ports(self): port_bindings = build_port_bindings( - ["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"]) - assert port_bindings["1000"] == [ - ("127.0.0.1", "1000"), ("127.0.0.1", "2000") - ] + ["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"] + ) + assert port_bindings["1000"] == [("127.0.0.1", "1000"), ("127.0.0.1", "2000")] def test_build_port_bindings_with_nonmatching_internal_ports(self): port_bindings = build_port_bindings( - ["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"]) + ["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"] + ) assert port_bindings["1000"] == [("127.0.0.1", "1000")] assert port_bindings["2000"] == [("127.0.0.1", "2000")] @@ -614,47 +600,118 @@ def test_build_port_bindings_with_port_range(self): def test_build_port_bindings_with_matching_internal_port_ranges(self): port_bindings = build_port_bindings( - ["127.0.0.1:1000-1001:1000-1001", "127.0.0.1:2000-2001:1000-1001"]) - assert port_bindings["1000"] == [ - ("127.0.0.1", "1000"), ("127.0.0.1", "2000") - ] - assert port_bindings["1001"] == [ - ("127.0.0.1", "1001"), ("127.0.0.1", "2001") - ] + ["127.0.0.1:1000-1001:1000-1001", "127.0.0.1:2000-2001:1000-1001"] + ) + assert port_bindings["1000"] == [("127.0.0.1", "1000"), ("127.0.0.1", "2000")] + assert port_bindings["1001"] == [("127.0.0.1", "1001"), ("127.0.0.1", "2001")] def test_build_port_bindings_with_nonmatching_internal_port_ranges(self): port_bindings = build_port_bindings( - ["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"]) + ["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"] + ) assert port_bindings["1000"] == [("127.0.0.1", "1000")] assert port_bindings["2000"] == [("127.0.0.1", "2000")] class FormatEnvironmentTest(unittest.TestCase): def test_format_env_binary_unicode_value(self): - env_dict = { - 'ARTIST_NAME': b'\xec\x86\xa1\xec\xa7\x80\xec\x9d\x80' - } - assert format_environment(env_dict) == ['ARTIST_NAME=송지은'] + env_dict = {"ARTIST_NAME": b"\xec\x86\xa1\xec\xa7\x80\xec\x9d\x80"} + assert format_environment(env_dict) == ["ARTIST_NAME=송지은"] def test_format_env_no_value(self): env_dict = { - 'FOO': None, - 'BAR': '', + "FOO": None, + "BAR": "", } - assert sorted(format_environment(env_dict)) == ['BAR=', 'FOO'] + assert sorted(format_environment(env_dict)) == ["BAR=", "FOO"] def test_compare_versions(): - assert compare_version('1.0', '1.1') == 1 - assert compare_version('1.10', '1.1') == -1 - assert compare_version('1.10', '1.10') == 0 - assert compare_version('1.10.0', '1.10.1') == 1 - assert compare_version('1.9', '1.10') == 1 - assert compare_version('1.9.1', '1.10') == 1 + assert compare_version("1.0", "1.1") == 1 + assert compare_version("1.10", "1.1") == -1 + assert compare_version("1.10", "1.10") == 0 + assert compare_version("1.10.0", "1.10.1") == 1 + assert compare_version("1.9", "1.10") == 1 + assert compare_version("1.9.1", "1.10") == 1 # Test comparison helpers - assert version_lt('1.0', '1.27') - assert version_gte('1.27', '1.20') + assert version_lt("1.0", "1.27") + assert version_gte("1.27", "1.20") # Test zero-padding - assert compare_version('1', '1.0') == 0 - assert compare_version('1.10', '1.10.1') == 1 - assert compare_version('1.10.0', '1.10') == 0 + assert compare_version("1", "1.0") == 0 + assert compare_version("1.10", "1.10.1") == 1 + assert compare_version("1.10.0", "1.10") == 0 + + +class UnixSocketDiscoveryTest(unittest.TestCase): + """Tests for the Unix socket discovery logic in constants.py.""" + + def test_find_socket_prefers_traditional_location(self): + """When /var/run/docker.sock exists, it should be preferred.""" + + def mock_exists(path): + # All sockets exist - should prefer the first one + return path in UNIX_SOCKET_PATHS + + with mock.patch("docker.constants.os.path.exists", side_effect=mock_exists): + result = _find_available_unix_socket() + assert result == "http+unix:///var/run/docker.sock" + + def test_find_socket_falls_back_to_docker_desktop(self): + """When only ~/.docker/run/docker.sock exists, it should be used.""" + docker_desktop_socket = os.path.expanduser("~/.docker/run/docker.sock") + + def mock_exists(path): + # Only Docker Desktop v4.x+ socket exists + return path == docker_desktop_socket + + with mock.patch("docker.constants.os.path.exists", side_effect=mock_exists): + result = _find_available_unix_socket() + assert result == f"http+unix://{docker_desktop_socket}" + + def test_find_socket_falls_back_to_older_docker_desktop(self): + """When only ~/.docker/desktop/docker.sock exists, it should be used.""" + older_desktop_socket = os.path.expanduser("~/.docker/desktop/docker.sock") + + def mock_exists(path): + # Only older Docker Desktop socket exists + return path == older_desktop_socket + + with mock.patch("docker.constants.os.path.exists", side_effect=mock_exists): + result = _find_available_unix_socket() + assert result == f"http+unix://{older_desktop_socket}" + + def test_find_socket_fallback_when_none_exist(self): + """When no socket exists, should fall back to traditional location.""" + + def mock_exists(path): + # No sockets exist + return False + + with mock.patch("docker.constants.os.path.exists", side_effect=mock_exists): + result = _find_available_unix_socket() + # Should fall back to traditional location for consistent error messages + assert result == "http+unix:///var/run/docker.sock" + + def test_find_socket_preference_order(self): + """Verify the preference order: traditional > docker desktop v4 > older desktop.""" + docker_desktop_socket = os.path.expanduser("~/.docker/run/docker.sock") + older_desktop_socket = os.path.expanduser("~/.docker/desktop/docker.sock") + + # Test: when docker desktop v4 and older both exist, v4 should win + def mock_exists_v4_and_older(path): + return path in [docker_desktop_socket, older_desktop_socket] + + with mock.patch( + "docker.constants.os.path.exists", side_effect=mock_exists_v4_and_older + ): + result = _find_available_unix_socket() + assert result == f"http+unix://{docker_desktop_socket}" + + def test_unix_socket_paths_order(self): + """Verify UNIX_SOCKET_PATHS contains expected paths in correct order.""" + assert len(UNIX_SOCKET_PATHS) == 3 + assert UNIX_SOCKET_PATHS[0] == "/var/run/docker.sock" + assert UNIX_SOCKET_PATHS[1] == os.path.expanduser("~/.docker/run/docker.sock") + assert UNIX_SOCKET_PATHS[2] == os.path.expanduser( + "~/.docker/desktop/docker.sock" + )