path: root/tests/test.py

# 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)


def fake_get_raw_response_socket(self, response):
    class FakeSocket(object):
        def __init__(self, content):
            self.content = content
            self.idx = 0

        def __iter__(self):
            for line in self.content:
                yield line

        def settimeout(self, noop):
            return

        def setblocking(self, noop):
            return

        def makefile(self):
            file_contents = '{0}\r\n{1}'.format(
                len(self.content), self.content.decode('ascii')
            )
            if six.PY3:
                return io.StringIO(file_contents)
            return io.StringIO(unicode(file_contents))

        def recv(self, size):
            self.idx += size
            return self.content[self.idx - size:self.idx]

    return FakeSocket(response.content)


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)
@mock.patch('docker.Client._get_raw_response_socket',
            fake_get_raw_response_socket)
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_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))

        result = bytes() if six.PY3 else str()
        for line in logs:
            result += line

        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=True
        )

        self.assertEqual(
            result,
            '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))

        result = bytes() if six.PY3 else str()
        for line in logs:
            result += line

        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=True
        )

        self.assertEqual(
            result,
            'Flowering Nights\n(Sakuya Iyazoi)\n'.encode('ascii')
        )

    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:
            next(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.assertTrue(args[1]['stream'])

    def test_pull_tag(self):
        try:
            next(self.client.pull('joffrey/test001:latest'))
        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': 'latest', '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):
                next(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',
            data='{}',
            headers={'Content-Type': 'application/json'},
            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_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()