summaryrefslogtreecommitdiff
path: root/tests/test.py
diff options
context:
space:
mode:
authorSVN-Git Migration <python-modules-team@lists.alioth.debian.org>2015-10-08 11:55:19 -0700
committerSVN-Git Migration <python-modules-team@lists.alioth.debian.org>2015-10-08 11:55:19 -0700
commit7cd7ca7e903ee2dfb1a91ec118f8653b2310ec7d (patch)
tree421a6ed086ffa058221bb07911193b7c52295c68 /tests/test.py
parentdbb06b3ad559090aee9d0de2bd08355ac090fe4a (diff)
Imported Upstream version 0.5.3
Diffstat (limited to 'tests/test.py')
-rw-r--r--tests/test.py1630
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()