# Copyright 2013 dotCloud inc. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # http://www.apache.org/licenses/LICENSE-2.0 # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import base64 import datetime import gzip import io import json import os import re import shutil import signal import socket import sys import tarfile import tempfile import threading import time import unittest import warnings import random import docker import requests import six import base import fake_api try: from unittest import mock except ImportError: import mock DEFAULT_TIMEOUT_SECONDS = docker.client.constants.DEFAULT_TIMEOUT_SECONDS warnings.simplefilter('error') warnings.filterwarnings('error') def response(status_code=200, content='', headers=None, reason=None, elapsed=0, request=None): res = requests.Response() res.status_code = status_code if not isinstance(content, six.binary_type): content = json.dumps(content).encode('ascii') res._content = content res.headers = requests.structures.CaseInsensitiveDict(headers or {}) res.reason = reason res.elapsed = datetime.timedelta(elapsed) res.request = request return res def fake_resolve_authconfig(authconfig, registry=None): return None def fake_inspect_container(self, container, tty=False): return fake_api.get_fake_inspect_container(tty=tty)[1] def fake_inspect_container_tty(self, container): return fake_inspect_container(self, container, tty=True) def fake_resp(url, data=None, **kwargs): status_code, content = fake_api.fake_responses[url]() return response(status_code=status_code, content=content) fake_request = mock.Mock(side_effect=fake_resp) url_prefix = 'http+docker://localunixsocket/v{0}/'.format( docker.client.constants.DEFAULT_DOCKER_API_VERSION) class Cleanup(object): if sys.version_info < (2, 7): # Provide a basic implementation of addCleanup for Python < 2.7 def __init__(self, *args, **kwargs): super(Cleanup, self).__init__(*args, **kwargs) self._cleanups = [] def tearDown(self): super(Cleanup, self).tearDown() ok = True while self._cleanups: fn, args, kwargs = self._cleanups.pop(-1) try: fn(*args, **kwargs) except KeyboardInterrupt: raise except: ok = False if not ok: raise def addCleanup(self, function, *args, **kwargs): self._cleanups.append((function, args, kwargs)) @mock.patch.multiple('docker.Client', get=fake_request, post=fake_request, put=fake_request, delete=fake_request) class DockerClientTest(Cleanup, base.BaseTestCase): def setUp(self): self.client = docker.Client() # Force-clear authconfig to avoid tampering with the tests self.client._cfg = {'Configs': {}} def tearDown(self): self.client.close() def assertIn(self, object, collection): if six.PY2 and sys.version_info[1] <= 6: return self.assertTrue(object in collection) return super(DockerClientTest, self).assertIn(object, collection) def base_create_payload(self, img='busybox', cmd=None): if not cmd: cmd = ['true'] return {"Tty": False, "Image": img, "Cmd": cmd, "AttachStdin": False, "AttachStderr": True, "AttachStdout": True, "StdinOnce": False, "OpenStdin": False, "NetworkDisabled": False, } def test_ctor(self): try: docker.Client(version=1.12) except Exception as e: self.assertTrue(isinstance(e, docker.errors.DockerException)) if not six.PY3: self.assertEqual( str(e), 'Version parameter must be a string or None. Found float' ) ######################### # INFORMATION TESTS # ######################### def test_version(self): self.client.version() fake_request.assert_called_with( url_prefix + 'version', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_retrieve_server_version(self): client = docker.Client(version="auto") self.assertTrue(isinstance(client._version, six.string_types)) self.assertFalse(client._version == "auto") client.close() def test_auto_retrieve_server_version(self): version = self.client._retrieve_server_version() self.assertTrue(isinstance(version, six.string_types)) def test_info(self): self.client.info() fake_request.assert_called_with( url_prefix + 'info', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_search(self): self.client.search('busybox') fake_request.assert_called_with( url_prefix + 'images/search', params={'term': 'busybox'}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_image_viz(self): try: self.client.images('busybox', viz=True) self.fail('Viz output should not be supported!') except Exception: pass def test_events(self): self.client.events() fake_request.assert_called_with( url_prefix + 'events', params={'since': None, 'until': None, 'filters': None}, stream=True ) def test_events_with_since_until(self): ts = 1356048000 now = datetime.datetime.utcfromtimestamp(ts) since = now - datetime.timedelta(seconds=10) until = now + datetime.timedelta(seconds=10) self.client.events(since=since, until=until) fake_request.assert_called_with( url_prefix + 'events', params={ 'since': ts - 10, 'until': ts + 10, 'filters': None }, stream=True ) def test_events_with_filters(self): filters = {'event': ['die', 'stop'], 'container': fake_api.FAKE_CONTAINER_ID} self.client.events(filters=filters) expected_filters = docker.utils.convert_filters(filters) fake_request.assert_called_with( url_prefix + 'events', params={ 'since': None, 'until': None, 'filters': expected_filters }, stream=True ) ################### # LISTING TESTS # ################### def test_images(self): self.client.images(all=True) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 0, 'all': 1}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_images_quiet(self): self.client.images(all=True, quiet=True) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 1, 'all': 1}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_image_ids(self): self.client.images(quiet=True) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 1, 'all': 0}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_images_filters(self): self.client.images(filters={'dangling': True}) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 0, 'all': 0, 'filters': '{"dangling": ["true"]}'}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_list_containers(self): self.client.containers(all=True) fake_request.assert_called_with( url_prefix + 'containers/json', params={ 'all': 1, 'since': None, 'size': 0, 'limit': -1, 'trunc_cmd': 0, 'before': None }, timeout=DEFAULT_TIMEOUT_SECONDS ) ##################### # CONTAINER TESTS # ##################### def test_create_container(self): self.client.create_container('busybox', 'true') args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["true"], "AttachStdin": false, "AttachStderr": true, "AttachStdout": true, "StdinOnce": false, "OpenStdin": false, "NetworkDisabled": false}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_binds(self): mount_dest = '/mnt' self.client.create_container('busybox', ['ls', mount_dest], volumes=[mount_dest]) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["ls", "/mnt"], "AttachStdin": false, "Volumes": {"/mnt": {}}, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_volume_string(self): mount_dest = '/mnt' self.client.create_container('busybox', ['ls', mount_dest], volumes=mount_dest) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["ls", "/mnt"], "AttachStdin": false, "Volumes": {"/mnt": {}}, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_ports(self): self.client.create_container('busybox', 'ls', ports=[1111, (2222, 'udp'), (3333,)]) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["ls"], "AttachStdin": false, "ExposedPorts": { "1111/tcp": {}, "2222/udp": {}, "3333/tcp": {} }, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_entrypoint(self): self.client.create_container('busybox', 'hello', entrypoint='cowsay entry') args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["hello"], "AttachStdin": false, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "Entrypoint": ["cowsay", "entry"]}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_cpu_shares(self): self.client.create_container('busybox', 'ls', cpu_shares=5) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["ls"], "AttachStdin": false, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "CpuShares": 5}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_cpuset(self): self.client.create_container('busybox', 'ls', cpuset='0,1') args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["ls"], "AttachStdin": false, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "Cpuset": "0,1", "CpusetCpus": "0,1"}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_cgroup_parent(self): self.client.create_container( 'busybox', 'ls', host_config=self.client.create_host_config( cgroup_parent='test' ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') data = json.loads(args[1]['data']) self.assertIn('HostConfig', data) self.assertIn('CgroupParent', data['HostConfig']) self.assertEqual(data['HostConfig']['CgroupParent'], 'test') def test_create_container_with_working_dir(self): self.client.create_container('busybox', 'ls', working_dir='/root') args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["ls"], "AttachStdin": false, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "WorkingDir": "/root"}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_stdin_open(self): self.client.create_container('busybox', 'true', stdin_open=True) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["true"], "AttachStdin": true, "AttachStderr": true, "AttachStdout": true, "StdinOnce": true, "OpenStdin": true, "NetworkDisabled": false}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_volumes_from(self): vol_names = ['foo', 'bar'] try: self.client.create_container('busybox', 'true', volumes_from=vol_names) except docker.errors.DockerException: self.assertTrue( docker.utils.compare_version('1.10', self.client._version) >= 0 ) return args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data'])['VolumesFrom'], ','.join(vol_names)) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_empty_volumes_from(self): self.client.create_container('busybox', 'true', volumes_from=[]) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertTrue('VolumesFrom' not in data) def test_create_named_container(self): self.client.create_container('busybox', 'true', name='marisa-kirisame') args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), json.loads(''' {"Tty": false, "Image": "busybox", "Cmd": ["true"], "AttachStdin": false, "AttachStderr": true, "AttachStdout": true, "StdinOnce": false, "OpenStdin": false, "NetworkDisabled": false}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual(args[1]['params'], {'name': 'marisa-kirisame'}) def test_create_container_with_mem_limit_as_int(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( mem_limit=128.0 ) ) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['HostConfig']['Memory'], 128.0) def test_create_container_with_mem_limit_as_string(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( mem_limit='128' ) ) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['HostConfig']['Memory'], 128.0) def test_create_container_with_mem_limit_as_string_with_k_unit(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( mem_limit='128k' ) ) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['HostConfig']['Memory'], 128.0 * 1024) def test_create_container_with_mem_limit_as_string_with_m_unit(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( mem_limit='128m' ) ) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['HostConfig']['Memory'], 128.0 * 1024 * 1024) def test_create_container_with_mem_limit_as_string_with_g_unit(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( mem_limit='128g' ) ) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual( data['HostConfig']['Memory'], 128.0 * 1024 * 1024 * 1024 ) def test_create_container_with_mem_limit_as_string_with_wrong_value(self): self.assertRaises( docker.errors.DockerException, self.client.create_host_config, mem_limit='128p' ) self.assertRaises( docker.errors.DockerException, self.client.create_host_config, mem_limit='1f28' ) def test_start_container(self): self.client.start(fake_api.FAKE_CONTAINER_ID) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual(json.loads(args[1]['data']), {}) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_start_container_none(self): try: self.client.start(container=None) except ValueError as e: self.assertEqual(str(e), 'image or container param is undefined') else: self.fail('Command should raise ValueError') try: self.client.start(None) except ValueError as e: self.assertEqual(str(e), 'image or container param is undefined') else: self.fail('Command should raise ValueError') def test_start_container_regression_573(self): self.client.start(**{'container': fake_api.FAKE_CONTAINER_ID}) def test_create_container_with_lxc_conf(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( lxc_conf={'lxc.conf.k': 'lxc.conf.value'} ) ) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/create' ) expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['LxcConf'] = [ {"Value": "lxc.conf.value", "Key": "lxc.conf.k"} ] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_lxc_conf_compat(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( lxc_conf=[{'Key': 'lxc.conf.k', 'Value': 'lxc.conf.value'}] ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['LxcConf'] = [ {"Value": "lxc.conf.value", "Key": "lxc.conf.k"} ] self.assertEqual( json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_binds_ro(self): mount_dest = '/mnt' mount_origin = '/tmp' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( binds={mount_origin: { "bind": mount_dest, "ro": True }} ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Binds'] = ["/tmp:/mnt:ro"] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_binds_rw(self): mount_dest = '/mnt' mount_origin = '/tmp' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( binds={mount_origin: { "bind": mount_dest, "ro": False }} ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Binds'] = ["/tmp:/mnt:rw"] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_binds_mode(self): mount_dest = '/mnt' mount_origin = '/tmp' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( binds={mount_origin: { "bind": mount_dest, "mode": "z", }} ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Binds'] = ["/tmp:/mnt:z"] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_binds_mode_and_ro_error(self): try: mount_dest = '/mnt' mount_origin = '/tmp' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( binds={mount_origin: { "bind": mount_dest, "mode": "z", "ro": True, }} ) ) except ValueError: return self.fail('Command should raise ValueError') def test_create_container_with_binds_list(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( binds=[ "/tmp:/mnt/1:ro", "/tmp:/mnt/2", ], ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Binds'] = [ "/tmp:/mnt/1:ro", "/tmp:/mnt/2", ] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_port_binds(self): self.maxDiff = None self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( port_bindings={ 1111: None, 2222: 2222, '3333/udp': (3333,), 4444: ('127.0.0.1',), 5555: ('127.0.0.1', 5555), 6666: [('127.0.0.1',), ('192.168.0.1',)] } ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') data = json.loads(args[1]['data']) port_bindings = data['HostConfig']['PortBindings'] self.assertTrue('1111/tcp' in port_bindings) self.assertTrue('2222/tcp' in port_bindings) self.assertTrue('3333/udp' in port_bindings) self.assertTrue('4444/tcp' in port_bindings) self.assertTrue('5555/tcp' in port_bindings) self.assertTrue('6666/tcp' in port_bindings) self.assertEqual( [{"HostPort": "", "HostIp": ""}], port_bindings['1111/tcp'] ) self.assertEqual( [{"HostPort": "2222", "HostIp": ""}], port_bindings['2222/tcp'] ) self.assertEqual( [{"HostPort": "3333", "HostIp": ""}], port_bindings['3333/udp'] ) self.assertEqual( [{"HostPort": "", "HostIp": "127.0.0.1"}], port_bindings['4444/tcp'] ) self.assertEqual( [{"HostPort": "5555", "HostIp": "127.0.0.1"}], port_bindings['5555/tcp'] ) self.assertEqual(len(port_bindings['6666/tcp']), 2) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_mac_address(self): mac_address_expected = "02:42:ac:11:00:0a" container = self.client.create_container( 'busybox', ['sleep', '60'], mac_address=mac_address_expected) res = self.client.inspect_container(container['Id']) self.assertEqual(mac_address_expected, res['NetworkSettings']['MacAddress']) def test_create_container_with_links(self): link_path = 'path' alias = 'alias' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( links={link_path: alias} ) ) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/create' ) expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Links'] = ['path:alias'] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) def test_create_container_with_multiple_links(self): link_path = 'path' alias = 'alias' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( links={ link_path + '1': alias + '1', link_path + '2': alias + '2' } ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Links'] = [ 'path1:alias1', 'path2:alias2' ] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) def test_create_container_with_links_as_list_of_tuples(self): link_path = 'path' alias = 'alias' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( links=[(link_path, alias)] ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Links'] = ['path:alias'] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) def test_create_container_privileged(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config(privileged=True) ) expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Privileged'] = True args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_lxc_conf(self): if six.PY2: try: self.client.start( fake_api.FAKE_CONTAINER_ID, lxc_conf={'lxc.conf.k': 'lxc.conf.value'} ) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start( fake_api.FAKE_CONTAINER_ID, lxc_conf={'lxc.conf.k': 'lxc.conf.value'} ) def test_start_container_with_lxc_conf_compat(self): if six.PY2: try: self.client.start( fake_api.FAKE_CONTAINER_ID, lxc_conf=[{'Key': 'lxc.conf.k', 'Value': 'lxc.conf.value'}] ) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start( fake_api.FAKE_CONTAINER_ID, lxc_conf=[{'Key': 'lxc.conf.k', 'Value': 'lxc.conf.value'}] ) def test_start_container_with_binds_ro(self): mount_dest = '/mnt' mount_origin = '/tmp' if six.PY2: try: self.client.start( fake_api.FAKE_CONTAINER_ID, binds={ mount_origin: { "bind": mount_dest, "ro": True } } ) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start( fake_api.FAKE_CONTAINER_ID, binds={ mount_origin: { "bind": mount_dest, "ro": True } } ) def test_start_container_with_binds_rw(self): mount_dest = '/mnt' mount_origin = '/tmp' if six.PY2: try: self.client.start( fake_api.FAKE_CONTAINER_ID, binds={ mount_origin: {"bind": mount_dest, "ro": False} } ) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start( fake_api.FAKE_CONTAINER_ID, binds={ mount_origin: {"bind": mount_dest, "ro": False} } ) def test_start_container_with_port_binds(self): self.maxDiff = None if six.PY2: try: self.client.start(fake_api.FAKE_CONTAINER_ID, port_bindings={ 1111: None, 2222: 2222, '3333/udp': (3333,), 4444: ('127.0.0.1',), 5555: ('127.0.0.1', 5555), 6666: [('127.0.0.1',), ('192.168.0.1',)] }) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start(fake_api.FAKE_CONTAINER_ID, port_bindings={ 1111: None, 2222: 2222, '3333/udp': (3333,), 4444: ('127.0.0.1',), 5555: ('127.0.0.1', 5555), 6666: [('127.0.0.1',), ('192.168.0.1',)] }) def test_start_container_with_links(self): # one link link_path = 'path' alias = 'alias' if six.PY2: try: self.client.start(fake_api.FAKE_CONTAINER_ID, links={link_path: alias}) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start( fake_api.FAKE_CONTAINER_ID, links={link_path: alias} ) def test_start_container_with_multiple_links(self): link_path = 'path' alias = 'alias' if six.PY2: try: self.client.start( fake_api.FAKE_CONTAINER_ID, links={ link_path + '1': alias + '1', link_path + '2': alias + '2' } ) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start( fake_api.FAKE_CONTAINER_ID, links={ link_path + '1': alias + '1', link_path + '2': alias + '2' } ) def test_start_container_with_links_as_list_of_tuples(self): # one link link_path = 'path' alias = 'alias' if six.PY2: try: self.client.start(fake_api.FAKE_CONTAINER_ID, links=[(link_path, alias)]) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start(fake_api.FAKE_CONTAINER_ID, links=[(link_path, alias)]) def test_start_container_privileged(self): if six.PY2: try: self.client.start(fake_api.FAKE_CONTAINER_ID, privileged=True) except DeprecationWarning: return else: self.fail('Expected a DeprecationWarning') else: with self.assertWarns(DeprecationWarning): self.client.start(fake_api.FAKE_CONTAINER_ID, privileged=True) def test_start_container_with_dict_instead_of_id(self): self.client.start({'Id': fake_api.FAKE_CONTAINER_ID}) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual(json.loads(args[1]['data']), {}) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_restart_policy(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( restart_policy={ "Name": "always", "MaximumRetryCount": 0 } ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['RestartPolicy'] = { "MaximumRetryCount": 0, "Name": "always" } self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_added_capabilities(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config(cap_add=['MKNOD']) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['CapAdd'] = ['MKNOD'] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_dropped_capabilities(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config(cap_drop=['MKNOD']) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['CapDrop'] = ['MKNOD'] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_devices(self): self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( devices=['/dev/sda:/dev/xvda:rwm', '/dev/sdb:/dev/xvdb', '/dev/sdc'] ) ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Devices'] = [ {'CgroupPermissions': 'rwm', 'PathInContainer': '/dev/xvda', 'PathOnHost': '/dev/sda'}, {'CgroupPermissions': 'rwm', 'PathInContainer': '/dev/xvdb', 'PathOnHost': '/dev/sdb'}, {'CgroupPermissions': 'rwm', 'PathInContainer': '/dev/sdc', 'PathOnHost': '/dev/sdc'} ] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_labels_dict(self): labels_dict = { six.text_type('foo'): six.text_type('1'), six.text_type('bar'): six.text_type('2'), } self.client.create_container( 'busybox', 'true', labels=labels_dict, ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data'])['Labels'], labels_dict) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_labels_list(self): labels_list = [ six.text_type('foo'), six.text_type('bar'), ] labels_dict = { six.text_type('foo'): six.text_type(), six.text_type('bar'): six.text_type(), } self.client.create_container( 'busybox', 'true', labels=labels_list, ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') self.assertEqual(json.loads(args[1]['data'])['Labels'], labels_dict) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_create_container_with_named_volume(self): mount_dest = '/mnt' volume_name = 'name' self.client.create_container( 'busybox', 'true', host_config=self.client.create_host_config( binds={volume_name: { "bind": mount_dest, "ro": False }}), volume_driver='foodriver', ) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/create') expected_payload = self.base_create_payload() expected_payload['VolumeDriver'] = 'foodriver' expected_payload['HostConfig'] = self.client.create_host_config() expected_payload['HostConfig']['Binds'] = ["name:/mnt:rw"] self.assertEqual(json.loads(args[1]['data']), expected_payload) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], DEFAULT_TIMEOUT_SECONDS ) def test_resize_container(self): self.client.resize( {'Id': fake_api.FAKE_CONTAINER_ID}, height=15, width=120 ) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/resize', params={'h': 15, 'w': 120}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_rename_container(self): self.client.rename( {'Id': fake_api.FAKE_CONTAINER_ID}, name='foobar' ) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/rename', params={'name': 'foobar'}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_wait(self): self.client.wait(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/wait', timeout=None ) def test_wait_with_dict_instead_of_id(self): self.client.wait({'Id': fake_api.FAKE_CONTAINER_ID}) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/wait', timeout=None ) def _socket_path_for_client_session(self, client): socket_adapter = client.get_adapter('http+docker://') return socket_adapter.socket_path def test_url_compatibility_unix(self): c = docker.Client(base_url="unix://socket") assert self._socket_path_for_client_session(c) == '/socket' def test_url_compatibility_unix_triple_slash(self): c = docker.Client(base_url="unix:///socket") assert self._socket_path_for_client_session(c) == '/socket' def test_url_compatibility_http_unix_triple_slash(self): c = docker.Client(base_url="http+unix:///socket") assert self._socket_path_for_client_session(c) == '/socket' def test_url_compatibility_http(self): c = docker.Client(base_url="http://hostname:1234") assert c.base_url == "http://hostname:1234" def test_url_compatibility_tcp(self): c = docker.Client(base_url="tcp://hostname:1234") assert c.base_url == "http://hostname:1234" def test_logs(self): with mock.patch('docker.Client.inspect_container', fake_inspect_container): logs = self.client.logs(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 0, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=DEFAULT_TIMEOUT_SECONDS, stream=False ) self.assertEqual( logs, 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii') ) def test_logs_with_dict_instead_of_id(self): with mock.patch('docker.Client.inspect_container', fake_inspect_container): logs = self.client.logs({'Id': fake_api.FAKE_CONTAINER_ID}) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 0, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=DEFAULT_TIMEOUT_SECONDS, stream=False ) self.assertEqual( logs, 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii') ) def test_log_streaming(self): with mock.patch('docker.Client.inspect_container', fake_inspect_container): self.client.logs(fake_api.FAKE_CONTAINER_ID, stream=True) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 1, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=DEFAULT_TIMEOUT_SECONDS, stream=True ) def test_log_tail(self): with mock.patch('docker.Client.inspect_container', fake_inspect_container): self.client.logs(fake_api.FAKE_CONTAINER_ID, stream=False, tail=10) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 0, 'stderr': 1, 'stdout': 1, 'tail': 10}, timeout=DEFAULT_TIMEOUT_SECONDS, stream=False ) def test_log_tty(self): m = mock.Mock() with mock.patch('docker.Client.inspect_container', fake_inspect_container_tty): with mock.patch('docker.Client._stream_raw_result', m): self.client.logs(fake_api.FAKE_CONTAINER_ID, stream=True) self.assertTrue(m.called) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 1, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=DEFAULT_TIMEOUT_SECONDS, stream=True ) def test_diff(self): self.client.diff(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/changes', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_diff_with_dict_instead_of_id(self): self.client.diff({'Id': fake_api.FAKE_CONTAINER_ID}) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/changes', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_port(self): self.client.port({'Id': fake_api.FAKE_CONTAINER_ID}, 1111) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/json', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_stop_container(self): timeout = 2 self.client.stop(fake_api.FAKE_CONTAINER_ID, timeout=timeout) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/stop', params={'t': timeout}, timeout=(DEFAULT_TIMEOUT_SECONDS + timeout) ) def test_stop_container_with_dict_instead_of_id(self): timeout = 2 self.client.stop({'Id': fake_api.FAKE_CONTAINER_ID}, timeout=timeout) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/stop', params={'t': timeout}, timeout=(DEFAULT_TIMEOUT_SECONDS + timeout) ) def test_exec_create(self): self.client.exec_create(fake_api.FAKE_CONTAINER_ID, ['ls', '-1']) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/{0}/exec'.format( fake_api.FAKE_CONTAINER_ID ) ) self.assertEqual( json.loads(args[1]['data']), { 'Tty': False, 'AttachStdout': True, 'Container': fake_api.FAKE_CONTAINER_ID, 'Cmd': ['ls', '-1'], 'Privileged': False, 'AttachStdin': False, 'AttachStderr': True, 'User': '' } ) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_exec_start(self): self.client.exec_start(fake_api.FAKE_EXEC_ID) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'exec/{0}/start'.format( fake_api.FAKE_EXEC_ID ) ) self.assertEqual( json.loads(args[1]['data']), { 'Tty': False, 'Detach': False, } ) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_exec_inspect(self): self.client.exec_inspect(fake_api.FAKE_EXEC_ID) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'exec/{0}/json'.format( fake_api.FAKE_EXEC_ID ) ) def test_exec_resize(self): self.client.exec_resize(fake_api.FAKE_EXEC_ID, height=20, width=60) fake_request.assert_called_with( url_prefix + 'exec/{0}/resize'.format(fake_api.FAKE_EXEC_ID), params={'h': 20, 'w': 60}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_pause_container(self): self.client.pause(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/pause', timeout=(DEFAULT_TIMEOUT_SECONDS) ) def test_unpause_container(self): self.client.unpause(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/unpause', timeout=(DEFAULT_TIMEOUT_SECONDS) ) def test_kill_container(self): self.client.kill(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/kill', params={}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_kill_container_with_dict_instead_of_id(self): self.client.kill({'Id': fake_api.FAKE_CONTAINER_ID}) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/kill', params={}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_kill_container_with_signal(self): self.client.kill(fake_api.FAKE_CONTAINER_ID, signal=signal.SIGTERM) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/kill', params={'signal': signal.SIGTERM}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_restart_container(self): self.client.restart(fake_api.FAKE_CONTAINER_ID, timeout=2) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/restart', params={'t': 2}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_restart_container_with_dict_instead_of_id(self): self.client.restart({'Id': fake_api.FAKE_CONTAINER_ID}, timeout=2) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/restart', params={'t': 2}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_remove_container(self): self.client.remove_container(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b', params={'v': False, 'link': False, 'force': False}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_remove_container_with_dict_instead_of_id(self): self.client.remove_container({'Id': fake_api.FAKE_CONTAINER_ID}) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b', params={'v': False, 'link': False, 'force': False}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_remove_link(self): self.client.remove_container(fake_api.FAKE_CONTAINER_ID, link=True) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b', params={'v': False, 'link': True, 'force': False}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_export(self): self.client.export(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/export', stream=True, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_export_with_dict_instead_of_id(self): self.client.export({'Id': fake_api.FAKE_CONTAINER_ID}) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/export', stream=True, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_inspect_container(self): self.client.inspect_container(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/json', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_inspect_container_undefined_id(self): for arg in None, '', {True: True}: try: self.client.inspect_container(arg) except docker.errors.NullResource as e: self.assertEqual( e.args[0], 'image or container param is undefined' ) else: self.fail('Command expected NullResource exception') def test_container_stats(self): self.client.stats(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/stats', timeout=60, stream=True ) ################## # IMAGES TESTS # ################## def test_pull(self): self.client.pull('joffrey/test001') args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'images/create' ) self.assertEqual( args[1]['params'], {'tag': None, 'fromImage': 'joffrey/test001'} ) self.assertFalse(args[1]['stream']) def test_pull_stream(self): self.client.pull('joffrey/test001', stream=True) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'images/create' ) self.assertEqual( args[1]['params'], {'tag': None, 'fromImage': 'joffrey/test001'} ) self.assertTrue(args[1]['stream']) def test_commit(self): self.client.commit(fake_api.FAKE_CONTAINER_ID) fake_request.assert_called_with( url_prefix + 'commit', data='{}', headers={'Content-Type': 'application/json'}, params={ 'repo': None, 'comment': None, 'tag': None, 'container': '3cc2351ab11b', 'author': None }, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_remove_image(self): self.client.remove_image(fake_api.FAKE_IMAGE_ID) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128', params={'force': False, 'noprune': False}, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_image_history(self): self.client.history(fake_api.FAKE_IMAGE_NAME) fake_request.assert_called_with( url_prefix + 'images/test_image/history', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_import_image(self): self.client.import_image( fake_api.FAKE_TARBALL_PATH, repository=fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) fake_request.assert_called_with( url_prefix + 'images/create', params={ 'repo': fake_api.FAKE_REPO_NAME, 'tag': fake_api.FAKE_TAG_NAME, 'fromSrc': fake_api.FAKE_TARBALL_PATH }, data=None, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_import_image_from_bytes(self): stream = (i for i in range(0, 100)) self.client.import_image( stream, repository=fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) fake_request.assert_called_with( url_prefix + 'images/create', params={ 'repo': fake_api.FAKE_REPO_NAME, 'tag': fake_api.FAKE_TAG_NAME, 'fromSrc': '-', }, headers={ 'Content-Type': 'application/tar', }, data=stream, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_import_image_from_image(self): self.client.import_image( image=fake_api.FAKE_IMAGE_NAME, repository=fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) fake_request.assert_called_with( url_prefix + 'images/create', params={ 'repo': fake_api.FAKE_REPO_NAME, 'tag': fake_api.FAKE_TAG_NAME, 'fromImage': fake_api.FAKE_IMAGE_NAME }, data=None, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_inspect_image(self): self.client.inspect_image(fake_api.FAKE_IMAGE_NAME) fake_request.assert_called_with( url_prefix + 'images/test_image/json', timeout=DEFAULT_TIMEOUT_SECONDS ) def test_inspect_image_undefined_id(self): for arg in None, '', {True: True}: try: self.client.inspect_image(arg) except docker.errors.NullResource as e: self.assertEqual( e.args[0], 'image or container param is undefined' ) else: self.fail('Command expected NullResource exception') def test_insert_image(self): try: self.client.insert(fake_api.FAKE_IMAGE_NAME, fake_api.FAKE_URL, fake_api.FAKE_PATH) except docker.errors.DeprecatedMethod: self.assertTrue( docker.utils.compare_version('1.12', self.client._version) >= 0 ) return fake_request.assert_called_with( url_prefix + 'images/test_image/insert', params={ 'url': fake_api.FAKE_URL, 'path': fake_api.FAKE_PATH }, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_push_image(self): with mock.patch('docker.auth.auth.resolve_authconfig', fake_resolve_authconfig): self.client.push(fake_api.FAKE_IMAGE_NAME) fake_request.assert_called_with( url_prefix + 'images/test_image/push', params={ 'tag': None }, data='{}', headers={'Content-Type': 'application/json'}, stream=False, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_push_image_with_tag(self): with mock.patch('docker.auth.auth.resolve_authconfig', fake_resolve_authconfig): self.client.push( fake_api.FAKE_IMAGE_NAME, tag=fake_api.FAKE_TAG_NAME ) fake_request.assert_called_with( url_prefix + 'images/test_image/push', params={ 'tag': fake_api.FAKE_TAG_NAME, }, data='{}', headers={'Content-Type': 'application/json'}, stream=False, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_push_image_stream(self): with mock.patch('docker.auth.auth.resolve_authconfig', fake_resolve_authconfig): self.client.push(fake_api.FAKE_IMAGE_NAME, stream=True) fake_request.assert_called_with( url_prefix + 'images/test_image/push', params={ 'tag': None }, data='{}', headers={'Content-Type': 'application/json'}, stream=True, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_tag_image(self): self.client.tag(fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/tag', params={ 'tag': None, 'repo': 'repo', 'force': 0 }, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_tag_image_tag(self): self.client.tag( fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/tag', params={ 'tag': 'tag', 'repo': 'repo', 'force': 0 }, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_tag_image_force(self): self.client.tag( fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME, force=True) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/tag', params={ 'tag': None, 'repo': 'repo', 'force': 1 }, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_get_image(self): self.client.get_image(fake_api.FAKE_IMAGE_ID) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/get', stream=True, timeout=DEFAULT_TIMEOUT_SECONDS ) def test_load_image(self): self.client.load_image('Byte Stream....') fake_request.assert_called_with( url_prefix + 'images/load', data='Byte Stream....', timeout=DEFAULT_TIMEOUT_SECONDS ) ################# # BUILDER TESTS # ################# def test_build_container(self): script = io.BytesIO('\n'.join([ 'FROM busybox', 'MAINTAINER docker-py', 'RUN mkdir -p /tmp/test', 'EXPOSE 8080', 'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' ' /tmp/silence.tar.gz' ]).encode('ascii')) self.client.build(fileobj=script) def test_build_container_pull(self): script = io.BytesIO('\n'.join([ 'FROM busybox', 'MAINTAINER docker-py', 'RUN mkdir -p /tmp/test', 'EXPOSE 8080', 'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' ' /tmp/silence.tar.gz' ]).encode('ascii')) self.client.build(fileobj=script, pull=True) def test_build_container_stream(self): script = io.BytesIO('\n'.join([ 'FROM busybox', 'MAINTAINER docker-py', 'RUN mkdir -p /tmp/test', 'EXPOSE 8080', 'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' ' /tmp/silence.tar.gz' ]).encode('ascii')) self.client.build(fileobj=script, stream=True) def test_build_container_custom_context(self): script = io.BytesIO('\n'.join([ 'FROM busybox', 'MAINTAINER docker-py', 'RUN mkdir -p /tmp/test', 'EXPOSE 8080', 'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' ' /tmp/silence.tar.gz' ]).encode('ascii')) context = docker.utils.mkbuildcontext(script) self.client.build(fileobj=context, custom_context=True) def test_build_container_custom_context_gzip(self): script = io.BytesIO('\n'.join([ 'FROM busybox', 'MAINTAINER docker-py', 'RUN mkdir -p /tmp/test', 'EXPOSE 8080', 'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' ' /tmp/silence.tar.gz' ]).encode('ascii')) context = docker.utils.mkbuildcontext(script) gz_context = gzip.GzipFile(fileobj=context) self.client.build( fileobj=gz_context, custom_context=True, encoding="gzip" ) def test_build_remote_with_registry_auth(self): self.client._auth_configs = { 'https://example.com': { 'user': 'example', 'password': 'example', 'email': 'example@example.com' } } self.client.build(path='https://github.com/docker-library/mongo') def test_build_container_with_named_dockerfile(self): self.client.build('.', dockerfile='nameddockerfile') def test_build_container_with_container_limits(self): self.client.build('.', container_limits={ 'memory': 1024 * 1024, 'cpusetcpus': 1, 'cpushares': 1000, 'memswap': 1024 * 1024 * 8 }) def test_build_container_invalid_container_limits(self): self.assertRaises( docker.errors.DockerException, lambda: self.client.build('.', container_limits={ 'foo': 'bar' }) ) ####################### # PY SPECIFIC TESTS # ####################### def test_load_config_no_file(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder) cfg = docker.auth.load_config(folder) self.assertTrue(cfg is not None) def test_load_config(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder) dockercfg_path = os.path.join(folder, '.dockercfg') with open(dockercfg_path, 'w') as f: auth_ = base64.b64encode(b'sakuya:izayoi').decode('ascii') f.write('auth = {0}\n'.format(auth_)) f.write('email = sakuya@scarlet.net') cfg = docker.auth.load_config(dockercfg_path) self.assertTrue(docker.auth.INDEX_NAME in cfg) self.assertNotEqual(cfg[docker.auth.INDEX_NAME], None) cfg = cfg[docker.auth.INDEX_NAME] self.assertEqual(cfg['username'], 'sakuya') self.assertEqual(cfg['password'], 'izayoi') self.assertEqual(cfg['email'], 'sakuya@scarlet.net') self.assertEqual(cfg.get('auth'), None) def test_load_config_with_random_name(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder) dockercfg_path = os.path.join(folder, '.{0}.dockercfg'.format( random.randrange(100000))) registry = 'https://your.private.registry.io' auth_ = base64.b64encode(b'sakuya:izayoi').decode('ascii') config = { registry: { 'auth': '{0}'.format(auth_), 'email': 'sakuya@scarlet.net' } } with open(dockercfg_path, 'w') as f: f.write(json.dumps(config)) cfg = docker.auth.load_config(dockercfg_path) self.assertTrue(registry in cfg) self.assertNotEqual(cfg[registry], None) cfg = cfg[registry] self.assertEqual(cfg['username'], 'sakuya') self.assertEqual(cfg['password'], 'izayoi') self.assertEqual(cfg['email'], 'sakuya@scarlet.net') self.assertEqual(cfg.get('auth'), None) def test_tar_with_excludes(self): base = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, base) for d in ['test/foo', 'bar']: os.makedirs(os.path.join(base, d)) for f in ['a.txt', 'b.py', 'other.png']: with open(os.path.join(base, d, f), 'w') as f: f.write("content") for exclude, names in ( (['*.py'], ['bar', 'bar/a.txt', 'bar/other.png', 'test', 'test/foo', 'test/foo/a.txt', 'test/foo/other.png']), (['*.png', 'bar'], ['test', 'test/foo', 'test/foo/a.txt', 'test/foo/b.py']), (['test/foo', 'a.txt'], ['bar', 'bar/a.txt', 'bar/b.py', 'bar/other.png', 'test']), ): with docker.utils.tar(base, exclude=exclude) as archive: tar = tarfile.open(fileobj=archive) self.assertEqual(sorted(tar.getnames()), names) def test_tar_with_empty_directory(self): base = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, base) for d in ['foo', 'bar']: os.makedirs(os.path.join(base, d)) with docker.utils.tar(base) as archive: tar = tarfile.open(fileobj=archive) self.assertEqual(sorted(tar.getnames()), ['bar', 'foo']) def test_tar_with_file_symlinks(self): base = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, base) with open(os.path.join(base, 'foo'), 'w') as f: f.write("content") os.makedirs(os.path.join(base, 'bar')) os.symlink('../foo', os.path.join(base, 'bar/foo')) with docker.utils.tar(base) as archive: tar = tarfile.open(fileobj=archive) self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo']) def test_tar_with_directory_symlinks(self): base = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, base) for d in ['foo', 'bar']: os.makedirs(os.path.join(base, d)) os.symlink('../foo', os.path.join(base, 'bar/foo')) with docker.utils.tar(base) as archive: tar = tarfile.open(fileobj=archive) self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo']) ####################### # HOST CONFIG TESTS # ####################### def test_create_host_config_secopt(self): security_opt = ['apparmor:test_profile'] result = self.client.create_host_config(security_opt=security_opt) self.assertIn('SecurityOpt', result) self.assertEqual(result['SecurityOpt'], security_opt) self.assertRaises( docker.errors.DockerException, self.client.create_host_config, security_opt='wrong' ) class StreamTest(Cleanup, base.BaseTestCase): def setUp(self): socket_dir = tempfile.mkdtemp() self.build_context = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, socket_dir) self.addCleanup(shutil.rmtree, self.build_context) self.socket_file = os.path.join(socket_dir, 'test_sock.sock') self.server_socket = self._setup_socket() self.stop_server = False server_thread = threading.Thread(target=self.run_server) server_thread.setDaemon(True) server_thread.start() self.response = None self.request_handler = None self.addCleanup(server_thread.join) self.addCleanup(self.stop) def stop(self): self.stop_server = True def _setup_socket(self): server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) server_sock.bind(self.socket_file) # Non-blocking mode so that we can shut the test down easily server_sock.setblocking(0) server_sock.listen(5) return server_sock def run_server(self): try: while not self.stop_server: try: connection, client_address = self.server_socket.accept() except socket.error: # Probably no connection to accept yet time.sleep(0.01) continue connection.setblocking(1) try: self.request_handler(connection) finally: connection.close() finally: self.server_socket.close() def early_response_sending_handler(self, connection): data = b'' headers = None connection.sendall(self.response) while not headers: data += connection.recv(2048) parts = data.split(b'\r\n\r\n', 1) if len(parts) == 2: headers, data = parts mo = re.search(r'Content-Length: ([0-9]+)', headers.decode()) assert mo content_length = int(mo.group(1)) while True: if len(data) >= content_length: break data += connection.recv(2048) def test_early_stream_response(self): self.request_handler = self.early_response_sending_handler lines = [] for i in range(0, 50): line = str(i).encode() lines += [('%x' % len(line)).encode(), line] lines.append(b'0') lines.append(b'') self.response = ( b'HTTP/1.1 200 OK\r\n' b'Transfer-Encoding: chunked\r\n' b'\r\n' ) + b'\r\n'.join(lines) with docker.Client(base_url="http+unix://" + self.socket_file) \ as client: for i in range(5): try: stream = client.build( path=self.build_context, stream=True ) break except requests.ConnectionError as e: if i == 4: raise e self.assertEqual(list(stream), [ str(i).encode() for i in range(50)]) if __name__ == '__main__': unittest.main()