docker-compose/tests/unit/cli/docker_client_test.py

from __future__ import absolute_import
from __future__ import unicode_literals

import os
import platform

import docker
import pytest

import compose
from compose.cli import errors
from compose.cli.docker_client import docker_client
from compose.cli.docker_client import tls_config_from_options
from tests import mock
from tests import unittest


class DockerClientTestCase(unittest.TestCase):

    def test_docker_client_no_home(self):
        with mock.patch.dict(os.environ):
            del os.environ['HOME']
            docker_client(os.environ)

    @mock.patch.dict(os.environ)
    def test_docker_client_with_custom_timeout(self):
        os.environ['COMPOSE_HTTP_TIMEOUT'] = '123'
        client = docker_client(os.environ)
        assert client.timeout == 123

    @mock.patch.dict(os.environ)
    def test_custom_timeout_error(self):
        os.environ['COMPOSE_HTTP_TIMEOUT'] = '123'
        client = docker_client(os.environ)

        with mock.patch('compose.cli.errors.log') as fake_log:
            with pytest.raises(errors.ConnectionError):
                with errors.handle_connection_errors(client):
                    raise errors.RequestsConnectionError(
                        errors.ReadTimeoutError(None, None, None))

        assert fake_log.error.call_count == 1
        assert '123' in fake_log.error.call_args[0][0]

        with mock.patch('compose.cli.errors.log') as fake_log:
            with pytest.raises(errors.ConnectionError):
                with errors.handle_connection_errors(client):
                    raise errors.ReadTimeout()

        assert fake_log.error.call_count == 1
        assert '123' in fake_log.error.call_args[0][0]

    def test_user_agent(self):
        client = docker_client(os.environ)
        expected = "docker-compose/{0} docker-py/{1} {2}/{3}".format(
            compose.__version__,
            docker.__version__,
            platform.system(),
            platform.release()
        )
        self.assertEqual(client.headers['User-Agent'], expected)

    @mock.patch.dict(os.environ)
    def test_docker_client_default_windows_host(self):
        with mock.patch('compose.cli.docker_client.IS_WINDOWS_PLATFORM', True):
            if 'DOCKER_HOST' in os.environ:
                del os.environ['DOCKER_HOST']
            client = docker_client(os.environ)
            assert client.base_url == 'http://127.0.0.1:2375'


class TLSConfigTestCase(unittest.TestCase):
    ca_cert = 'tests/fixtures/tls/ca.pem'
    client_cert = 'tests/fixtures/tls/cert.pem'
    key = 'tests/fixtures/tls/key.key'

    def test_simple_tls(self):
        options = {'--tls': True}
        result = tls_config_from_options(options)
        assert result is True

    def test_tls_ca_cert(self):
        options = {
            '--tlscacert': self.ca_cert, '--tlsverify': True
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.ca_cert == options['--tlscacert']
        assert result.verify is True

    def test_tls_ca_cert_explicit(self):
        options = {
            '--tlscacert': self.ca_cert, '--tls': True,
            '--tlsverify': True
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.ca_cert == options['--tlscacert']
        assert result.verify is True

    def test_tls_client_cert(self):
        options = {
            '--tlscert': self.client_cert, '--tlskey': self.key
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.cert == (options['--tlscert'], options['--tlskey'])

    def test_tls_client_cert_explicit(self):
        options = {
            '--tlscert': self.client_cert, '--tlskey': self.key,
            '--tls': True
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.cert == (options['--tlscert'], options['--tlskey'])

    def test_tls_client_and_ca(self):
        options = {
            '--tlscert': self.client_cert, '--tlskey': self.key,
            '--tlsverify': True, '--tlscacert': self.ca_cert
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.cert == (options['--tlscert'], options['--tlskey'])
        assert result.ca_cert == options['--tlscacert']
        assert result.verify is True

    def test_tls_client_and_ca_explicit(self):
        options = {
            '--tlscert': self.client_cert, '--tlskey': self.key,
            '--tlsverify': True, '--tlscacert': self.ca_cert,
            '--tls': True
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.cert == (options['--tlscert'], options['--tlskey'])
        assert result.ca_cert == options['--tlscacert']
        assert result.verify is True

    def test_tls_client_missing_key(self):
        options = {'--tlscert': self.client_cert}
        with pytest.raises(docker.errors.TLSParameterError):
            tls_config_from_options(options)

        options = {'--tlskey': self.key}
        with pytest.raises(docker.errors.TLSParameterError):
            tls_config_from_options(options)

    def test_assert_hostname_explicit_skip(self):
        options = {'--tlscacert': self.ca_cert, '--skip-hostname-check': True}
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.assert_hostname is False

    def test_tls_client_and_ca_quoted_paths(self):
        options = {
            '--tlscacert': '"{0}"'.format(self.ca_cert),
            '--tlscert': '"{0}"'.format(self.client_cert),
            '--tlskey': '"{0}"'.format(self.key),
            '--tlsverify': True
        }
        result = tls_config_from_options(options)
        assert isinstance(result, docker.tls.TLSConfig)
        assert result.cert == (self.client_cert, self.key)
        assert result.ca_cert == self.ca_cert
        assert result.verify is True