# 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 io import json import os import shutil import signal import sys import tarfile import tempfile import unittest import gzip import docker import requests import six import fake_api try: from unittest import mock except ImportError: import mock 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_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+unix://var/run/docker.sock/v{0}/'.format( docker.client.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, unittest.TestCase): 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 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( e.message, 'version parameter must be a string. Found float' ) ######################### # INFORMATION TESTS # ######################### def test_version(self): try: self.client.version() except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'version', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_info(self): try: self.client.info() except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'info', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_search(self): try: self.client.search('busybox') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/search', params={'term': 'busybox'}, timeout=docker.client.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 ################### # LISTING TESTS # ################### def test_images(self): try: self.client.images(all=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 0, 'all': 1}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_images_quiet(self): try: self.client.images(all=True, quiet=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 1, 'all': 1}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_image_ids(self): try: self.client.images(quiet=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 1, 'all': 0}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_images_filters(self): try: self.client.images(filters={'dangling': True}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/json', params={'filter': None, 'only_ids': 0, 'all': 0, 'filters': '{"dangling": ["true"]}'}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_list_containers(self): try: self.client.containers(all=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/json', params={ 'all': 1, 'since': None, 'size': 0, 'limit': -1, 'trunc_cmd': 1, 'before': None }, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) ##################### # CONTAINER TESTS # ##################### def test_create_container(self): try: self.client.create_container('busybox', 'true') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "StdinOnce": false, "OpenStdin": false, "NetworkDisabled": false, "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_binds(self): mount_dest = '/mnt' try: self.client.create_container('busybox', ['ls', mount_dest], volumes=[mount_dest]) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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": {}}, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_volume_string(self): mount_dest = '/mnt' try: self.client.create_container('busybox', ['ls', mount_dest], volumes=mount_dest) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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": {}}, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_ports(self): try: self.client.create_container('busybox', 'ls', ports=[1111, (2222, 'udp'), (3333,)]) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "ExposedPorts": { "1111/tcp": {}, "2222/udp": {}, "3333/tcp": {} }, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_entrypoint(self): try: self.client.create_container('busybox', 'hello', entrypoint='cowsay') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "Entrypoint": "cowsay", "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_cpu_shares(self): try: self.client.create_container('busybox', 'ls', cpu_shares=5) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "CpuShares": 5, "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_cpuset(self): try: self.client.create_container('busybox', 'ls', cpuset='0,1') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "Cpuset": "0,1", "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_working_dir(self): try: self.client.create_container('busybox', 'ls', working_dir='/root') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "OpenStdin": false, "StdinOnce": false, "NetworkDisabled": false, "WorkingDir": "/root", "MemorySwap": 0}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_create_container_with_stdin_open(self): try: self.client.create_container('busybox', 'true', stdin_open=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "StdinOnce": true, "OpenStdin": true, "NetworkDisabled": false, "MemorySwap": 0}''')) 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 as e: self.assertTrue( docker.utils.compare_version('1.10', self.client._version) >= 0 ) return except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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): try: self.client.create_container('busybox', 'true', volumes_from=[]) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertTrue('VolumesFrom' not in data) def test_create_named_container(self): try: self.client.create_container('busybox', 'true', name='marisa-kirisame') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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, "Memory": 0, "AttachStderr": true, "AttachStdout": true, "StdinOnce": false, "OpenStdin": false, "NetworkDisabled": false, "MemorySwap": 0}''')) 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): try: self.client.create_container('busybox', 'true', mem_limit=128.0) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['Memory'], 128.0) def test_create_container_with_mem_limit_as_string(self): try: self.client.create_container('busybox', 'true', mem_limit='128') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['Memory'], 128.0) def test_create_container_with_mem_limit_as_string_with_k_unit(self): try: self.client.create_container('busybox', 'true', mem_limit='128k') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['Memory'], 128.0 * 1024) def test_create_container_with_mem_limit_as_string_with_m_unit(self): try: self.client.create_container('busybox', 'true', mem_limit='128m') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['Memory'], 128.0 * 1024 * 1024) def test_create_container_with_mem_limit_as_string_with_g_unit(self): try: self.client.create_container('busybox', 'true', mem_limit='128g') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args data = json.loads(args[1]['data']) self.assertEqual(data['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_container, 'busybox', 'true', mem_limit='128p') self.assertRaises(docker.errors.DockerException, self.client.create_container, 'busybox', 'true', mem_limit='1f28') def test_start_container(self): try: self.client.start(fake_api.FAKE_CONTAINER_ID) except Exception as e: raise e self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_lxc_conf(self): try: self.client.start( fake_api.FAKE_CONTAINER_ID, lxc_conf={'lxc.conf.k': 'lxc.conf.value'} ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"LxcConf": [{"Value": "lxc.conf.value", "Key": "lxc.conf.k"}], "PublishAllPorts": False, "Privileged": False} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_lxc_conf_compat(self): try: self.client.start( fake_api.FAKE_CONTAINER_ID, lxc_conf=[{'Key': 'lxc.conf.k', 'Value': 'lxc.conf.value'}] ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/3cc2351ab11b/start') self.assertEqual( json.loads(args[1]['data']), { "LxcConf": [{"Key": "lxc.conf.k", "Value": "lxc.conf.value"}], "PublishAllPorts": False, "Privileged": False, } ) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_binds_ro(self): try: mount_dest = '/mnt' mount_origin = '/tmp' self.client.start(fake_api.FAKE_CONTAINER_ID, binds={mount_origin: { "bind": mount_dest, "ro": True }}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/3cc2351ab11b/start') self.assertEqual(json.loads(args[1]['data']), {"Binds": ["/tmp:/mnt:ro"], "PublishAllPorts": False, "Privileged": False}) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS) def test_start_container_with_binds_rw(self): try: mount_dest = '/mnt' mount_origin = '/tmp' self.client.start(fake_api.FAKE_CONTAINER_ID, binds={mount_origin: { "bind": mount_dest, "ro": False}}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/3cc2351ab11b/start') self.assertEqual(json.loads(args[1]['data']), {"Binds": ["/tmp:/mnt:rw"], "PublishAllPorts": False, "Privileged": False}) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_port_binds(self): self.maxDiff = None 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 Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'containers/3cc2351ab11b/start') data = json.loads(args[1]['data']) self.assertEqual(data['PublishAllPorts'], False) self.assertTrue('1111/tcp' in data['PortBindings']) self.assertTrue('2222/tcp' in data['PortBindings']) self.assertTrue('3333/udp' in data['PortBindings']) self.assertTrue('4444/tcp' in data['PortBindings']) self.assertTrue('5555/tcp' in data['PortBindings']) self.assertTrue('6666/tcp' in data['PortBindings']) self.assertEqual( [{"HostPort": "", "HostIp": ""}], data['PortBindings']['1111/tcp'] ) self.assertEqual( [{"HostPort": "2222", "HostIp": ""}], data['PortBindings']['2222/tcp'] ) self.assertEqual( [{"HostPort": "3333", "HostIp": ""}], data['PortBindings']['3333/udp'] ) self.assertEqual( [{"HostPort": "", "HostIp": "127.0.0.1"}], data['PortBindings']['4444/tcp'] ) self.assertEqual( [{"HostPort": "5555", "HostIp": "127.0.0.1"}], data['PortBindings']['5555/tcp'] ) self.assertEqual(len(data['PortBindings']['6666/tcp']), 2) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_links(self): # one link try: link_path = 'path' alias = 'alias' self.client.start(fake_api.FAKE_CONTAINER_ID, links={link_path: alias}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False, "Links": ["path:alias"]} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) def test_start_container_with_multiple_links(self): try: link_path = 'path' alias = 'alias' self.client.start( fake_api.FAKE_CONTAINER_ID, links={ link_path + '1': alias + '1', link_path + '2': alias + '2' } ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), { "PublishAllPorts": False, "Privileged": False, "Links": ["path1:alias1", "path2:alias2"] } ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) def test_start_container_with_links_as_list_of_tuples(self): # one link try: link_path = 'path' alias = 'alias' self.client.start(fake_api.FAKE_CONTAINER_ID, links=[(link_path, alias)]) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False, "Links": ["path:alias"]} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) def test_start_container_privileged(self): try: self.client.start(fake_api.FAKE_CONTAINER_ID, privileged=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual(json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": True}) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_dict_instead_of_id(self): try: self.client.start({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_restart_policy(self): try: self.client.start(fake_api.FAKE_CONTAINER_ID, restart_policy={ "Name": "always", "MaximumRetryCount": 0 }) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False, "RestartPolicy": {"MaximumRetryCount": 0, "Name": "always"}} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_added_capabilities(self): try: self.client.start(fake_api.FAKE_CONTAINER_ID, cap_add=['MKNOD']) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False, "CapAdd": ["MKNOD"]} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_dropped_capabilities(self): try: self.client.start(fake_api.FAKE_CONTAINER_ID, cap_drop=['MKNOD']) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False, "CapDrop": ["MKNOD"]} ) self.assertEqual( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_start_container_with_devices(self): try: self.client.start(fake_api.FAKE_CONTAINER_ID, devices=['/dev/sda:/dev/xvda:rwm', '/dev/sdb:/dev/xvdb', '/dev/sdc']) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual( args[0][0], url_prefix + 'containers/3cc2351ab11b/start' ) self.assertEqual( json.loads(args[1]['data']), {"PublishAllPorts": False, "Privileged": False, "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( args[1]['headers'], {'Content-Type': 'application/json'} ) self.assertEqual( args[1]['timeout'], docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_resize_container(self): try: self.client.resize( {'Id': fake_api.FAKE_CONTAINER_ID}, height=15, width=120 ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/resize', params={'h': 15, 'w': 120}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_wait(self): try: self.client.wait(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/wait', timeout=None ) def test_wait_with_dict_instead_of_id(self): try: self.client.wait({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: raise e self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/wait', timeout=None ) def test_url_compatibility_unix(self): c = docker.Client(base_url="unix://socket") assert c.base_url == "http+unix://socket" def test_url_compatibility_unix_triple_slash(self): c = docker.Client(base_url="unix:///socket") assert c.base_url == "http+unix://socket" def test_url_compatibility_http_unix_triple_slash(self): c = docker.Client(base_url="http+unix:///socket") assert c.base_url == "http+unix://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): try: logs = self.client.logs(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 0, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=docker.client.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): try: logs = self.client.logs({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 0, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS, stream=False ) self.assertEqual( logs, 'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii') ) def test_log_streaming(self): try: self.client.logs(fake_api.FAKE_CONTAINER_ID, stream=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 1, 'stderr': 1, 'stdout': 1, 'tail': 'all'}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS, stream=True ) def test_log_tail(self): try: self.client.logs(fake_api.FAKE_CONTAINER_ID, stream=False, tail=10) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/logs', params={'timestamps': 0, 'follow': 0, 'stderr': 1, 'stdout': 1, 'tail': 10}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS, stream=False ) def test_diff(self): try: self.client.diff(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/changes', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_diff_with_dict_instead_of_id(self): try: self.client.diff({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/changes', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_port(self): try: self.client.port({'Id': fake_api.FAKE_CONTAINER_ID}, 1111) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/json', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_stop_container(self): timeout = 2 try: self.client.stop(fake_api.FAKE_CONTAINER_ID, timeout=timeout) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/stop', params={'t': timeout}, timeout=(docker.client.DEFAULT_TIMEOUT_SECONDS + timeout) ) def test_stop_container_with_dict_instead_of_id(self): timeout = 2 try: self.client.stop({'Id': fake_api.FAKE_CONTAINER_ID}, timeout=timeout) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/stop', params={'t': timeout}, timeout=(docker.client.DEFAULT_TIMEOUT_SECONDS + timeout) ) def test_execute_command(self): try: self.client.execute(fake_api.FAKE_CONTAINER_ID, ['ls', '-1']) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) args = fake_request.call_args self.assertEqual(args[0][0], url_prefix + 'exec/3cc2351ab11b/start') self.assertEqual(json.loads(args[1]['data']), json.loads('''{ "Tty": false, "AttachStderr": true, "Container": "3cc2351ab11b", "Cmd": ["ls", "-1"], "AttachStdin": false, "User": "", "Detach": false, "Privileged": false, "AttachStdout": true}''')) self.assertEqual(args[1]['headers'], {'Content-Type': 'application/json'}) def test_pause_container(self): try: self.client.pause(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/pause', timeout=(docker.client.DEFAULT_TIMEOUT_SECONDS) ) def test_unpause_container(self): try: self.client.unpause(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/unpause', timeout=(docker.client.DEFAULT_TIMEOUT_SECONDS) ) def test_kill_container(self): try: self.client.kill(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/kill', params={}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_kill_container_with_dict_instead_of_id(self): try: self.client.kill({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/kill', params={}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_kill_container_with_signal(self): try: self.client.kill(fake_api.FAKE_CONTAINER_ID, signal=signal.SIGTERM) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/kill', params={'signal': signal.SIGTERM}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_restart_container(self): try: self.client.restart(fake_api.FAKE_CONTAINER_ID, timeout=2) except Exception as e: self.fail('Command should not raise exception : {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/restart', params={'t': 2}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_restart_container_with_dict_instead_of_id(self): try: self.client.restart({'Id': fake_api.FAKE_CONTAINER_ID}, timeout=2) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/restart', params={'t': 2}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_remove_container(self): try: self.client.remove_container(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b', params={'v': False, 'link': False, 'force': False}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_remove_container_with_dict_instead_of_id(self): try: self.client.remove_container({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b', params={'v': False, 'link': False, 'force': False}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_remove_link(self): try: self.client.remove_container(fake_api.FAKE_CONTAINER_ID, link=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b', params={'v': False, 'link': True, 'force': False}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_export(self): try: self.client.export(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/export', stream=True, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_export_with_dict_instead_of_id(self): try: self.client.export({'Id': fake_api.FAKE_CONTAINER_ID}) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/export', stream=True, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_inspect_container(self): try: self.client.inspect_container(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'containers/3cc2351ab11b/json', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) ################## # IMAGES TESTS # ################## def test_pull(self): try: self.client.pull('joffrey/test001') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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): try: self.client.pull('joffrey/test001', stream=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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): try: self.client.commit(fake_api.FAKE_CONTAINER_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_remove_image(self): try: self.client.remove_image(fake_api.FAKE_IMAGE_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128', params={'force': False, 'noprune': False}, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_image_history(self): try: self.client.history(fake_api.FAKE_IMAGE_NAME) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/test_image/history', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_import_image(self): try: self.client.import_image( fake_api.FAKE_TARBALL_PATH, repository=fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_import_image_from_file(self): buf = tempfile.NamedTemporaryFile(delete=False) try: # pretent the buffer is a file self.client.import_image( buf.name, repository=fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/create', params={ 'repo': fake_api.FAKE_REPO_NAME, 'tag': fake_api.FAKE_TAG_NAME, 'fromSrc': '-' }, data='', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) buf.close() os.remove(buf.name) def test_import_image_from_image(self): try: self.client.import_image( image=fake_api.FAKE_IMAGE_NAME, repository=fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_inspect_image(self): try: self.client.inspect_image(fake_api.FAKE_IMAGE_NAME) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/test_image/json', timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) 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 as e: self.assertTrue( docker.utils.compare_version('1.12', self.client._version) >= 0 ) return except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/test_image/insert', params={ 'url': fake_api.FAKE_URL, 'path': fake_api.FAKE_PATH }, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_push_image(self): try: with mock.patch('docker.auth.auth.resolve_authconfig', fake_resolve_authconfig): self.client.push(fake_api.FAKE_IMAGE_NAME) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/test_image/push', params={ 'tag': None }, data='{}', headers={'Content-Type': 'application/json'}, stream=False, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_push_image_with_tag(self): try: 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 ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_push_image_stream(self): try: with mock.patch('docker.auth.auth.resolve_authconfig', fake_resolve_authconfig): self.client.push(fake_api.FAKE_IMAGE_NAME, stream=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/test_image/push', params={ 'tag': None }, data='{}', headers={'Content-Type': 'application/json'}, stream=True, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_tag_image(self): try: self.client.tag(fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/tag', params={ 'tag': None, 'repo': 'repo', 'force': 0 }, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_tag_image_tag(self): try: self.client.tag( fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME, tag=fake_api.FAKE_TAG_NAME ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/tag', params={ 'tag': 'tag', 'repo': 'repo', 'force': 0 }, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_tag_image_force(self): try: self.client.tag( fake_api.FAKE_IMAGE_ID, fake_api.FAKE_REPO_NAME, force=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/tag', params={ 'tag': None, 'repo': 'repo', 'force': 1 }, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_get_image(self): try: self.client.get_image(fake_api.FAKE_IMAGE_ID) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/e9aa60c60128/get', stream=True, timeout=docker.client.DEFAULT_TIMEOUT_SECONDS ) def test_load_image(self): try: self.client.load_image('Byte Stream....') except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) fake_request.assert_called_with( url_prefix + 'images/load', data='Byte Stream....', timeout=docker.client.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')) try: self.client.build(fileobj=script) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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')) try: self.client.build(fileobj=script, pull=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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')) try: self.client.build(fileobj=script, stream=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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) try: self.client.build(fileobj=context, custom_context=True) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) 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) try: self.client.build( fileobj=gz_context, custom_context=True, encoding="gzip" ) except Exception as e: self.fail('Command should not raise exception: {0}'.format(e)) ####################### # 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) f = open(os.path.join(folder, '.dockercfg'), 'w') auth_ = base64.b64encode(b'sakuya:izayoi').decode('ascii') f.write('auth = {0}\n'.format(auth_)) f.write('email = sakuya@scarlet.net') f.close() cfg = docker.auth.load_config(folder) self.assertTrue(docker.auth.INDEX_URL in cfg) self.assertNotEqual(cfg[docker.auth.INDEX_URL], None) cfg = cfg[docker.auth.INDEX_URL] 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']), ): archive = docker.utils.tar(base, exclude=exclude) 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)) archive = docker.utils.tar(base) 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')) archive = docker.utils.tar(base) 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')) archive = docker.utils.tar(base) tar = tarfile.open(fileobj=archive) self.assertEqual(sorted(tar.getnames()), ['bar', 'bar/foo', 'foo']) if __name__ == '__main__': unittest.main()