diff options
author | SVN-Git Migration <python-modules-team@lists.alioth.debian.org> | 2015-10-08 11:55:19 -0700 |
---|---|---|
committer | SVN-Git Migration <python-modules-team@lists.alioth.debian.org> | 2015-10-08 11:55:19 -0700 |
commit | 7cd7ca7e903ee2dfb1a91ec118f8653b2310ec7d (patch) | |
tree | 421a6ed086ffa058221bb07911193b7c52295c68 /tests/test.py | |
parent | dbb06b3ad559090aee9d0de2bd08355ac090fe4a (diff) |
Imported Upstream version 0.5.3
Diffstat (limited to 'tests/test.py')
-rw-r--r-- | tests/test.py | 1630 |
1 files changed, 0 insertions, 1630 deletions
diff --git a/tests/test.py b/tests/test.py deleted file mode 100644 index 984e350..0000000 --- a/tests/test.py +++ /dev/null @@ -1,1630 +0,0 @@ -# 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': {}} - - ######################### - # 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_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_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_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}, - 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}, - 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}, - timeout=docker.client.DEFAULT_TIMEOUT_SECONDS, - stream=True - ) - - 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_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_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/a.txt', 'bar/other.png', - 'test/foo/a.txt', 'test/foo/other.png']), - (['*.png', 'bar'], ['test/foo/a.txt', 'test/foo/b.py']), - (['test/foo', 'a.txt'], ['bar/a.txt', 'bar/b.py', - 'bar/other.png']), - ): - archive = docker.utils.tar(base, exclude=exclude) - tar = tarfile.open(fileobj=archive) - self.assertEqual(sorted(tar.getnames()), names) - - -if __name__ == '__main__': - unittest.main() |