diff options
Diffstat (limited to 'tests/unit/utils_test.py')
| -rw-r--r-- | tests/unit/utils_test.py | 107 |
1 files changed, 57 insertions, 50 deletions
diff --git a/tests/unit/utils_test.py b/tests/unit/utils_test.py index 8880cfe..802d919 100644 --- a/tests/unit/utils_test.py +++ b/tests/unit/utils_test.py @@ -1,30 +1,22 @@ -# -*- coding: utf-8 -*- - import base64 import json import os import os.path import shutil -import sys import tempfile import unittest - +import pytest from docker.api.client import APIClient +from docker.constants import IS_WINDOWS_PLATFORM, DEFAULT_DOCKER_API_VERSION from docker.errors import DockerException -from docker.utils import ( - convert_filters, convert_volume_binds, decode_json_header, kwargs_from_env, - parse_bytes, parse_devices, parse_env_file, parse_host, - parse_repository_tag, split_command, update_headers, -) - +from docker.utils import (convert_filters, convert_volume_binds, + decode_json_header, kwargs_from_env, parse_bytes, + parse_devices, parse_env_file, parse_host, + parse_repository_tag, split_command, update_headers) from docker.utils.ports import build_port_bindings, split_port from docker.utils.utils import format_environment -import pytest - -import six - TEST_CERT_DIR = os.path.join( os.path.dirname(__file__), 'testdata/certs', @@ -40,7 +32,7 @@ class DecoratorsTest(unittest.TestCase): def f(self, headers=None): return headers - client = APIClient() + client = APIClient(version=DEFAULT_DOCKER_API_VERSION) client._general_configs = {} g = update_headers(f) @@ -83,15 +75,18 @@ class KwargsFromEnvTest(unittest.TestCase): DOCKER_CERT_PATH=TEST_CERT_DIR, DOCKER_TLS_VERIFY='1') kwargs = kwargs_from_env(assert_hostname=False) - assert 'https://192.168.59.103:2376' == kwargs['base_url'] + 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'].assert_hostname is False assert kwargs['tls'].verify + + parsed_host = parse_host(kwargs['base_url'], IS_WINDOWS_PLATFORM, True) + kwargs['version'] = DEFAULT_DOCKER_API_VERSION try: client = APIClient(**kwargs) - assert kwargs['base_url'] == client.base_url + assert parsed_host == client.base_url assert kwargs['tls'].ca_cert == client.verify assert kwargs['tls'].cert == client.cert except TypeError as e: @@ -102,15 +97,17 @@ class KwargsFromEnvTest(unittest.TestCase): DOCKER_CERT_PATH=TEST_CERT_DIR, DOCKER_TLS_VERIFY='') kwargs = kwargs_from_env(assert_hostname=True) - assert 'https://192.168.59.103:2376' == kwargs['base_url'] + 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'].assert_hostname is True 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 kwargs['base_url'] == client.base_url + assert parsed_host == client.base_url assert kwargs['tls'].cert == client.cert assert not kwargs['tls'].verify except TypeError as e: @@ -195,22 +192,22 @@ class ConverVolumeBindsTest(unittest.TestCase): assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw'] def test_convert_volume_binds_unicode_bytes_input(self): - expected = [u'/mnt/지연:/unicode/박:rw'] + expected = ['/mnt/지연:/unicode/박:rw'] data = { - u'/mnt/지연'.encode('utf-8'): { - 'bind': u'/unicode/박'.encode('utf-8'), + '/mnt/지연'.encode(): { + 'bind': '/unicode/박'.encode(), 'mode': 'rw' } } assert convert_volume_binds(data) == expected def test_convert_volume_binds_unicode_unicode_input(self): - expected = [u'/mnt/지연:/unicode/박:rw'] + expected = ['/mnt/지연:/unicode/박:rw'] data = { - u'/mnt/지연': { - 'bind': u'/unicode/박', + '/mnt/지연': { + 'bind': '/unicode/박', 'mode': 'rw' } } @@ -272,6 +269,11 @@ class ParseHostTest(unittest.TestCase): '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 = { @@ -281,7 +283,7 @@ class ParseHostTest(unittest.TestCase): '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', + 'unix://': 'http+unix:///var/run/docker.sock', '12.234.45.127:2375/docker/engine': ( 'http://12.234.45.127:2375/docker/engine' ), @@ -294,6 +296,9 @@ class ParseHostTest(unittest.TestCase): '[fd12::82d1]:2375/docker/engine': ( 'http://[fd12::82d1]:2375/docker/engine' ), + '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: @@ -304,7 +309,7 @@ class ParseHostTest(unittest.TestCase): assert parse_host(host, None) == expected def test_parse_host_empty_value(self): - unix_socket = 'http+unix://var/run/docker.sock' + unix_socket = 'http+unix:///var/run/docker.sock' npipe = 'npipe:////./pipe/docker_engine' for val in [None, '']: @@ -351,14 +356,14 @@ class ParseRepositoryTagTest(unittest.TestCase): ) def test_index_image_sha(self): - assert parse_repository_tag("root@sha256:{0}".format(self.sha)) == ( - "root", "sha256:{0}".format(self.sha) + assert parse_repository_tag(f"root@sha256:{self.sha}") == ( + "root", f"sha256:{self.sha}" ) def test_private_reg_image_sha(self): assert parse_repository_tag( - "url:5000/repo@sha256:{0}".format(self.sha) - ) == ("url:5000/repo", "sha256:{0}".format(self.sha)) + f"url:5000/repo@sha256:{self.sha}" + ) == ("url:5000/repo", f"sha256:{self.sha}") class ParseDeviceTest(unittest.TestCase): @@ -435,11 +440,7 @@ class ParseBytesTest(unittest.TestCase): parse_bytes("127.0.0.1K") def test_parse_bytes_float(self): - with pytest.raises(DockerException): - parse_bytes("1.5k") - - def test_parse_bytes_maxint(self): - assert parse_bytes("{0}k".format(sys.maxsize)) == sys.maxsize * 1024 + assert parse_bytes("1.5k") == 1536 class UtilsTest(unittest.TestCase): @@ -449,8 +450,8 @@ class UtilsTest(unittest.TestCase): tests = [ ({'dangling': True}, '{"dangling": ["true"]}'), ({'dangling': "true"}, '{"dangling": ["true"]}'), - ({'exited': 0}, '{"exited": [0]}'), - ({'exited': [0, 1]}, '{"exited": [0, 1]}'), + ({'exited': 0}, '{"exited": ["0"]}'), + ({'exited': [0, 1]}, '{"exited": ["0", "1"]}'), ] for filters, expected in tests: @@ -459,20 +460,13 @@ class UtilsTest(unittest.TestCase): def test_decode_json_header(self): obj = {'a': 'b', 'c': 1} data = None - if six.PY3: - data = base64.urlsafe_b64encode(bytes(json.dumps(obj), 'utf-8')) - else: - data = base64.urlsafe_b64encode(json.dumps(obj)) + 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(u'echo μμ') == ['echo', 'μμ'] - - @pytest.mark.skipif(six.PY3, reason="shlex doesn't support bytes in py3") - def test_split_command_with_bytes(self): assert split_command('echo μμ') == ['echo', 'μμ'] @@ -483,9 +477,12 @@ class PortsTest(unittest.TestCase): assert external_port == [("127.0.0.1", "1000")] def test_split_port_with_protocol(self): - internal_port, external_port = split_port("127.0.0.1:1000:2000/udp") - assert internal_port == ["2000/udp"] - assert external_port == [("127.0.0.1", "1000")] + for protocol in ['tcp', 'udp', 'sctp']: + internal_port, external_port = split_port( + "127.0.0.1:1000:2000/" + protocol + ) + assert internal_port == ["2000/" + protocol] + assert external_port == [("127.0.0.1", "1000")] def test_split_port_with_host_ip_no_port(self): internal_port, external_port = split_port("127.0.0.1::2000") @@ -534,10 +531,20 @@ class PortsTest(unittest.TestCase): 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") + assert internal_port == ["2000"] + assert external_port == [("2001:abcd:ef00::2", "1000")] + def test_split_port_invalid(self): with pytest.raises(ValueError): split_port("0.0.0.0:1000:2000:tcp") + def test_split_port_invalid_protocol(self): + with pytest.raises(ValueError): + split_port("0.0.0.0:1000:2000/ftp") + def test_non_matching_length_port_ranges(self): with pytest.raises(ValueError): split_port("0.0.0.0:1000-1010:2000-2002/tcp") @@ -609,7 +616,7 @@ class FormatEnvironmentTest(unittest.TestCase): env_dict = { 'ARTIST_NAME': b'\xec\x86\xa1\xec\xa7\x80\xec\x9d\x80' } - assert format_environment(env_dict) == [u'ARTIST_NAME=송지은'] + assert format_environment(env_dict) == ['ARTIST_NAME=송지은'] def test_format_env_no_value(self): env_dict = { |