docker-compose/tests/unit/config/config_test.py

# encoding: utf-8
from __future__ import print_function

import os
import shutil
import tempfile
from operator import itemgetter

import py
import pytest

from compose.config import config
from compose.config.errors import ConfigurationError
from compose.config.types import VolumeSpec
from compose.const import IS_WINDOWS_PLATFORM
from tests import mock
from tests import unittest


def make_service_dict(name, service_dict, working_dir, filename=None):
    """
    Test helper function to construct a ServiceExtendsResolver
    """
    resolver = config.ServiceExtendsResolver(config.ServiceConfig(
        working_dir=working_dir,
        filename=filename,
        name=name,
        config=service_dict))
    return config.process_service(resolver.run())


def service_sort(services):
    return sorted(services, key=itemgetter('name'))


def build_config_details(contents, working_dir='working_dir', filename='filename.yml'):
    return config.ConfigDetails(
        working_dir,
        [config.ConfigFile(filename, contents)])


class ConfigTest(unittest.TestCase):
    def test_load(self):
        service_dicts = config.load(
            build_config_details(
                {
                    'foo': {'image': 'busybox'},
                    'bar': {'image': 'busybox', 'environment': ['FOO=1']},
                },
                'tests/fixtures/extends',
                'common.yml'
            )
        )

        self.assertEqual(
            service_sort(service_dicts),
            service_sort([
                {
                    'name': 'bar',
                    'image': 'busybox',
                    'environment': {'FOO': '1'},
                },
                {
                    'name': 'foo',
                    'image': 'busybox',
                }
            ])
        )

    def test_load_throws_error_when_not_dict(self):
        with self.assertRaises(ConfigurationError):
            config.load(
                build_config_details(
                    {'web': 'busybox:latest'},
                    'working_dir',
                    'filename.yml'
                )
            )

    def test_config_invalid_service_names(self):
        for invalid_name in ['?not?allowed', ' ', '', '!', '/', '\xe2']:
            with pytest.raises(ConfigurationError) as exc:
                config.load(build_config_details(
                    {invalid_name: {'image': 'busybox'}},
                    'working_dir',
                    'filename.yml'))
            assert 'Invalid service name \'%s\'' % invalid_name in exc.exconly()

    def test_load_with_invalid_field_name(self):
        config_details = build_config_details(
            {'web': {'image': 'busybox', 'name': 'bogus'}},
            'working_dir',
            'filename.yml')
        with pytest.raises(ConfigurationError) as exc:
            config.load(config_details)
        error_msg = "Unsupported config option for 'web' service: 'name'"
        assert error_msg in exc.exconly()
        assert "Validation failed in file 'filename.yml'" in exc.exconly()

    def test_load_invalid_service_definition(self):
        config_details = build_config_details(
            {'web': 'wrong'},
            'working_dir',
            'filename.yml')
        with pytest.raises(ConfigurationError) as exc:
            config.load(config_details)
        error_msg = "service 'web' doesn't have any configuration options"
        assert error_msg in exc.exconly()

    def test_config_integer_service_name_raise_validation_error(self):
        expected_error_msg = ("In file 'filename.yml' service name: 1 needs to "
                              "be a string, eg '1'")
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {1: {'image': 'busybox'}},
                    'working_dir',
                    'filename.yml'
                )
            )

    @pytest.mark.xfail(IS_WINDOWS_PLATFORM, reason='paths use slash')
    def test_load_with_multiple_files(self):
        base_file = config.ConfigFile(
            'base.yaml',
            {
                'web': {
                    'image': 'example/web',
                    'links': ['db'],
                },
                'db': {
                    'image': 'example/db',
                },
            })
        override_file = config.ConfigFile(
            'override.yaml',
            {
                'web': {
                    'build': '/',
                    'volumes': ['/home/user/project:/code'],
                },
            })
        details = config.ConfigDetails('.', [base_file, override_file])

        service_dicts = config.load(details)
        expected = [
            {
                'name': 'web',
                'build': '/',
                'links': ['db'],
                'volumes': [VolumeSpec.parse('/home/user/project:/code')],
            },
            {
                'name': 'db',
                'image': 'example/db',
            },
        ]
        self.assertEqual(service_sort(service_dicts), service_sort(expected))

    def test_load_with_multiple_files_and_empty_override(self):
        base_file = config.ConfigFile(
            'base.yml',
            {'web': {'image': 'example/web'}})
        override_file = config.ConfigFile('override.yml', None)
        details = config.ConfigDetails('.', [base_file, override_file])

        with pytest.raises(ConfigurationError) as exc:
            config.load(details)
        error_msg = "Top level object in 'override.yml' needs to be an object"
        assert error_msg in exc.exconly()

    def test_load_with_multiple_files_and_empty_base(self):
        base_file = config.ConfigFile('base.yml', None)
        override_file = config.ConfigFile(
            'override.yml',
            {'web': {'image': 'example/web'}})
        details = config.ConfigDetails('.', [base_file, override_file])

        with pytest.raises(ConfigurationError) as exc:
            config.load(details)
        assert "Top level object in 'base.yml' needs to be an object" in exc.exconly()

    def test_load_with_multiple_files_and_extends_in_override_file(self):
        base_file = config.ConfigFile(
            'base.yaml',
            {
                'web': {'image': 'example/web'},
            })
        override_file = config.ConfigFile(
            'override.yaml',
            {
                'web': {
                    'extends': {
                        'file': 'common.yml',
                        'service': 'base',
                    },
                    'volumes': ['/home/user/project:/code'],
                },
            })
        details = config.ConfigDetails('.', [base_file, override_file])

        tmpdir = py.test.ensuretemp('config_test')
        self.addCleanup(tmpdir.remove)
        tmpdir.join('common.yml').write("""
            base:
              labels: ['label=one']
        """)
        with tmpdir.as_cwd():
            service_dicts = config.load(details)

        expected = [
            {
                'name': 'web',
                'image': 'example/web',
                'volumes': [VolumeSpec.parse('/home/user/project:/code')],
                'labels': {'label': 'one'},
            },
        ]
        self.assertEqual(service_sort(service_dicts), service_sort(expected))

    def test_load_with_multiple_files_and_invalid_override(self):
        base_file = config.ConfigFile(
            'base.yaml',
            {'web': {'image': 'example/web'}})
        override_file = config.ConfigFile(
            'override.yaml',
            {'bogus': 'thing'})
        details = config.ConfigDetails('.', [base_file, override_file])

        with pytest.raises(ConfigurationError) as exc:
            config.load(details)
        assert "service 'bogus' doesn't have any configuration" in exc.exconly()
        assert "In file 'override.yaml'" in exc.exconly()

    def test_config_valid_service_names(self):
        for valid_name in ['_', '-', '.__.', '_what-up.', 'what_.up----', 'whatup']:
            services = config.load(
                build_config_details(
                    {valid_name: {'image': 'busybox'}},
                    'tests/fixtures/extends',
                    'common.yml'))
            assert services[0]['name'] == valid_name

    def test_config_invalid_ports_format_validation(self):
        expected_error_msg = "Service 'web' configuration key 'ports' contains an invalid type"
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            for invalid_ports in [{"1": "8000"}, False, 0, "8000", 8000, ["8000", "8000"]]:
                config.load(
                    build_config_details(
                        {'web': {'image': 'busybox', 'ports': invalid_ports}},
                        'working_dir',
                        'filename.yml'
                    )
                )

    def test_config_valid_ports_format_validation(self):
        valid_ports = [["8000", "9000"], ["8000/8050"], ["8000"], [8000], ["49153-49154:3002-3003"]]
        for ports in valid_ports:
            config.load(
                build_config_details(
                    {'web': {'image': 'busybox', 'ports': ports}},
                    'working_dir',
                    'filename.yml'
                )
            )

    def test_config_hint(self):
        expected_error_msg = "(did you mean 'privileged'?)"
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'foo': {'image': 'busybox', 'privilige': 'something'},
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_invalid_config_build_and_image_specified(self):
        expected_error_msg = "Service 'foo' has both an image and build path specified."
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'foo': {'image': 'busybox', 'build': '.'},
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_invalid_config_type_should_be_an_array(self):
        expected_error_msg = "Service 'foo' configuration key 'links' contains an invalid type, it should be an array"
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'foo': {'image': 'busybox', 'links': 'an_link'},
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_invalid_config_not_a_dictionary(self):
        expected_error_msg = ("Top level object in 'filename.yml' needs to be "
                              "an object.")
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    ['foo', 'lol'],
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_invalid_config_not_unique_items(self):
        expected_error_msg = "has non-unique elements"
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'web': {'build': '.', 'devices': ['/dev/foo:/dev/foo', '/dev/foo:/dev/foo']}
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_invalid_list_of_strings_format(self):
        expected_error_msg = "Service 'web' configuration key 'command' contains 1"
        expected_error_msg += ", which is an invalid type, it should be a string"
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'web': {'build': '.', 'command': [1]}
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_config_image_and_dockerfile_raise_validation_error(self):
        expected_error_msg = "Service 'web' has both an image and alternate Dockerfile."
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {'web': {'image': 'busybox', 'dockerfile': 'Dockerfile.alt'}},
                    'working_dir',
                    'filename.yml'
                )
            )

    def test_config_extra_hosts_string_raises_validation_error(self):
        expected_error_msg = "Service 'web' configuration key 'extra_hosts' contains an invalid type"

        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {'web': {
                        'image': 'busybox',
                        'extra_hosts': 'somehost:162.242.195.82'
                    }},
                    'working_dir',
                    'filename.yml'
                )
            )

    def test_config_extra_hosts_list_of_dicts_validation_error(self):
        expected_error_msg = "key 'extra_hosts' contains {'somehost': '162.242.195.82'}, which is an invalid type, it should be a string"

        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {'web': {
                        'image': 'busybox',
                        'extra_hosts': [
                            {'somehost': '162.242.195.82'},
                            {'otherhost': '50.31.209.229'}
                        ]
                    }},
                    'working_dir',
                    'filename.yml'
                )
            )

    def test_config_ulimits_invalid_keys_validation_error(self):
        expected = ("Service 'web' configuration key 'ulimits' 'nofile' contains "
                    "unsupported option: 'not_soft_or_hard'")

        with pytest.raises(ConfigurationError) as exc:
            config.load(build_config_details(
                {
                    'web': {
                        'image': 'busybox',
                        'ulimits': {
                            'nofile': {
                                "not_soft_or_hard": 100,
                                "soft": 10000,
                                "hard": 20000,
                            }
                        }
                    }
                },
                'working_dir',
                'filename.yml'))
        assert expected in exc.exconly()

    def test_config_ulimits_required_keys_validation_error(self):

        with pytest.raises(ConfigurationError) as exc:
            config.load(build_config_details(
                {
                    'web': {
                        'image': 'busybox',
                        'ulimits': {'nofile': {"soft": 10000}}
                    }
                },
                'working_dir',
                'filename.yml'))
        assert "Service 'web' configuration key 'ulimits' 'nofile'" in exc.exconly()
        assert "'hard' is a required property" in exc.exconly()

    def test_config_ulimits_soft_greater_than_hard_error(self):
        expected = "cannot contain a 'soft' value higher than 'hard' value"

        with pytest.raises(ConfigurationError) as exc:
            config.load(build_config_details(
                {
                    'web': {
                        'image': 'busybox',
                        'ulimits': {
                            'nofile': {"soft": 10000, "hard": 1000}
                        }
                    }
                },
                'working_dir',
                'filename.yml'))
        assert expected in exc.exconly()

    def test_valid_config_which_allows_two_type_definitions(self):
        expose_values = [["8000"], [8000]]
        for expose in expose_values:
            service = config.load(
                build_config_details(
                    {'web': {
                        'image': 'busybox',
                        'expose': expose
                    }},
                    'working_dir',
                    'filename.yml'
                )
            )
            self.assertEqual(service[0]['expose'], expose)

    def test_valid_config_oneof_string_or_list(self):
        entrypoint_values = [["sh"], "sh"]
        for entrypoint in entrypoint_values:
            service = config.load(
                build_config_details(
                    {'web': {
                        'image': 'busybox',
                        'entrypoint': entrypoint
                    }},
                    'working_dir',
                    'filename.yml'
                )
            )
            self.assertEqual(service[0]['entrypoint'], entrypoint)

    @mock.patch('compose.config.validation.log')
    def test_logs_warning_for_boolean_in_environment(self, mock_logging):
        expected_warning_msg = "There is a boolean value in the 'environment' key."
        config.load(
            build_config_details(
                {'web': {
                    'image': 'busybox',
                    'environment': {'SHOW_STUFF': True}
                }},
                'working_dir',
                'filename.yml'
            )
        )

        self.assertTrue(mock_logging.warn.called)
        self.assertTrue(expected_warning_msg in mock_logging.warn.call_args[0][0])

    def test_config_valid_environment_dict_key_contains_dashes(self):
        services = config.load(
            build_config_details(
                {'web': {
                    'image': 'busybox',
                    'environment': {'SPRING_JPA_HIBERNATE_DDL-AUTO': 'none'}
                }},
                'working_dir',
                'filename.yml'
            )
        )
        self.assertEqual(services[0]['environment']['SPRING_JPA_HIBERNATE_DDL-AUTO'], 'none')

    def test_load_yaml_with_yaml_error(self):
        tmpdir = py.test.ensuretemp('invalid_yaml_test')
        self.addCleanup(tmpdir.remove)
        invalid_yaml_file = tmpdir.join('docker-compose.yml')
        invalid_yaml_file.write("""
            web:
              this is bogus: ok: what
        """)
        with pytest.raises(ConfigurationError) as exc:
            config.load_yaml(str(invalid_yaml_file))

        assert 'line 3, column 32' in exc.exconly()

    def test_validate_extra_hosts_invalid(self):
        with pytest.raises(ConfigurationError) as exc:
            config.load(build_config_details({
                'web': {
                    'image': 'alpine',
                    'extra_hosts': "www.example.com: 192.168.0.17",
                }
            }))
        assert "'extra_hosts' contains an invalid type" in exc.exconly()

    def test_validate_extra_hosts_invalid_list(self):
        with pytest.raises(ConfigurationError) as exc:
            config.load(build_config_details({
                'web': {
                    'image': 'alpine',
                    'extra_hosts': [
                        {'www.example.com': '192.168.0.17'},
                        {'api.example.com': '192.168.0.18'}
                    ],
                }
            }))
        assert "which is an invalid type" in exc.exconly()


class InterpolationTest(unittest.TestCase):
    @mock.patch.dict(os.environ)
    def test_config_file_with_environment_variable(self):
        os.environ.update(
            IMAGE="busybox",
            HOST_PORT="80",
            LABEL_VALUE="myvalue",
        )

        service_dicts = config.load(
            config.find('tests/fixtures/environment-interpolation', None),
        )

        self.assertEqual(service_dicts, [
            {
                'name': 'web',
                'image': 'busybox',
                'ports': ['80:8000'],
                'labels': {'mylabel': 'myvalue'},
                'hostname': 'host-',
                'command': '${ESCAPED}',
            }
        ])

    @mock.patch.dict(os.environ)
    def test_unset_variable_produces_warning(self):
        os.environ.pop('FOO', None)
        os.environ.pop('BAR', None)
        config_details = build_config_details(
            {
                'web': {
                    'image': '${FOO}',
                    'command': '${BAR}',
                    'container_name': '${BAR}',
                },
            },
            '.',
            None,
        )

        with mock.patch('compose.config.interpolation.log') as log:
            config.load(config_details)

            self.assertEqual(2, log.warn.call_count)
            warnings = sorted(args[0][0] for args in log.warn.call_args_list)
            self.assertIn('BAR', warnings[0])
            self.assertIn('FOO', warnings[1])

    @mock.patch.dict(os.environ)
    def test_invalid_interpolation(self):
        with self.assertRaises(config.ConfigurationError) as cm:
            config.load(
                build_config_details(
                    {'web': {'image': '${'}},
                    'working_dir',
                    'filename.yml'
                )
            )

        self.assertIn('Invalid', cm.exception.msg)
        self.assertIn('for "image" option', cm.exception.msg)
        self.assertIn('in service "web"', cm.exception.msg)
        self.assertIn('"${"', cm.exception.msg)

    def test_empty_environment_key_allowed(self):
        service_dict = config.load(
            build_config_details(
                {
                    'web': {
                        'build': '.',
                        'environment': {
                            'POSTGRES_PASSWORD': ''
                        },
                    },
                },
                '.',
                None,
            )
        )[0]
        self.assertEquals(service_dict['environment']['POSTGRES_PASSWORD'], '')


class VolumeConfigTest(unittest.TestCase):
    def test_no_binding(self):
        d = make_service_dict('foo', {'build': '.', 'volumes': ['/data']}, working_dir='.')
        self.assertEqual(d['volumes'], ['/data'])

    @mock.patch.dict(os.environ)
    def test_volume_binding_with_environment_variable(self):
        os.environ['VOLUME_PATH'] = '/host/path'
        d = config.load(build_config_details(
            {'foo': {'build': '.', 'volumes': ['${VOLUME_PATH}:/container/path']}},
            '.',
        ))[0]
        self.assertEqual(d['volumes'], [VolumeSpec.parse('/host/path:/container/path')])

    @pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='posix paths')
    @mock.patch.dict(os.environ)
    def test_volume_binding_with_home(self):
        os.environ['HOME'] = '/home/user'
        d = make_service_dict('foo', {'build': '.', 'volumes': ['~:/container/path']}, working_dir='.')
        self.assertEqual(d['volumes'], ['/home/user:/container/path'])

    def test_name_does_not_expand(self):
        d = make_service_dict('foo', {'build': '.', 'volumes': ['mydatavolume:/data']}, working_dir='.')
        self.assertEqual(d['volumes'], ['mydatavolume:/data'])

    def test_absolute_posix_path_does_not_expand(self):
        d = make_service_dict('foo', {'build': '.', 'volumes': ['/var/lib/data:/data']}, working_dir='.')
        self.assertEqual(d['volumes'], ['/var/lib/data:/data'])

    def test_absolute_windows_path_does_not_expand(self):
        d = make_service_dict('foo', {'build': '.', 'volumes': ['c:\\data:/data']}, working_dir='.')
        self.assertEqual(d['volumes'], ['c:\\data:/data'])

    @pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='posix paths')
    def test_relative_path_does_expand_posix(self):
        d = make_service_dict('foo', {'build': '.', 'volumes': ['./data:/data']}, working_dir='/home/me/myproject')
        self.assertEqual(d['volumes'], ['/home/me/myproject/data:/data'])

        d = make_service_dict('foo', {'build': '.', 'volumes': ['.:/data']}, working_dir='/home/me/myproject')
        self.assertEqual(d['volumes'], ['/home/me/myproject:/data'])

        d = make_service_dict('foo', {'build': '.', 'volumes': ['../otherproject:/data']}, working_dir='/home/me/myproject')
        self.assertEqual(d['volumes'], ['/home/me/otherproject:/data'])

    @pytest.mark.skipif(not IS_WINDOWS_PLATFORM, reason='windows paths')
    def test_relative_path_does_expand_windows(self):
        d = make_service_dict('foo', {'build': '.', 'volumes': ['./data:/data']}, working_dir='c:\\Users\\me\\myproject')
        self.assertEqual(d['volumes'], ['c:\\Users\\me\\myproject\\data:/data'])

        d = make_service_dict('foo', {'build': '.', 'volumes': ['.:/data']}, working_dir='c:\\Users\\me\\myproject')
        self.assertEqual(d['volumes'], ['c:\\Users\\me\\myproject:/data'])

        d = make_service_dict('foo', {'build': '.', 'volumes': ['../otherproject:/data']}, working_dir='c:\\Users\\me\\myproject')
        self.assertEqual(d['volumes'], ['c:\\Users\\me\\otherproject:/data'])

    @mock.patch.dict(os.environ)
    def test_home_directory_with_driver_does_not_expand(self):
        os.environ['NAME'] = 'surprise!'
        d = make_service_dict('foo', {
            'build': '.',
            'volumes': ['~:/data'],
            'volume_driver': 'foodriver',
        }, working_dir='.')
        self.assertEqual(d['volumes'], ['~:/data'])

    def test_volume_path_with_non_ascii_directory(self):
        volume = u'/Füü/data:/data'
        container_path = config.resolve_volume_path(".", volume)
        self.assertEqual(container_path, volume)


class MergePathMappingTest(object):
    def config_name(self):
        return ""

    def test_empty(self):
        service_dict = config.merge_service_dicts({}, {})
        self.assertNotIn(self.config_name(), service_dict)

    def test_no_override(self):
        service_dict = config.merge_service_dicts(
            {self.config_name(): ['/foo:/code', '/data']},
            {},
        )
        self.assertEqual(set(service_dict[self.config_name()]), set(['/foo:/code', '/data']))

    def test_no_base(self):
        service_dict = config.merge_service_dicts(
            {},
            {self.config_name(): ['/bar:/code']},
        )
        self.assertEqual(set(service_dict[self.config_name()]), set(['/bar:/code']))

    def test_override_explicit_path(self):
        service_dict = config.merge_service_dicts(
            {self.config_name(): ['/foo:/code', '/data']},
            {self.config_name(): ['/bar:/code']},
        )
        self.assertEqual(set(service_dict[self.config_name()]), set(['/bar:/code', '/data']))

    def test_add_explicit_path(self):
        service_dict = config.merge_service_dicts(
            {self.config_name(): ['/foo:/code', '/data']},
            {self.config_name(): ['/bar:/code', '/quux:/data']},
        )
        self.assertEqual(set(service_dict[self.config_name()]), set(['/bar:/code', '/quux:/data']))

    def test_remove_explicit_path(self):
        service_dict = config.merge_service_dicts(
            {self.config_name(): ['/foo:/code', '/quux:/data']},
            {self.config_name(): ['/bar:/code', '/data']},
        )
        self.assertEqual(set(service_dict[self.config_name()]), set(['/bar:/code', '/data']))


class MergeVolumesTest(unittest.TestCase, MergePathMappingTest):
    def config_name(self):
        return 'volumes'


class MergeDevicesTest(unittest.TestCase, MergePathMappingTest):
    def config_name(self):
        return 'devices'


class BuildOrImageMergeTest(unittest.TestCase):
    def test_merge_build_or_image_no_override(self):
        self.assertEqual(
            config.merge_service_dicts({'build': '.'}, {}),
            {'build': '.'},
        )

        self.assertEqual(
            config.merge_service_dicts({'image': 'redis'}, {}),
            {'image': 'redis'},
        )

    def test_merge_build_or_image_override_with_same(self):
        self.assertEqual(
            config.merge_service_dicts({'build': '.'}, {'build': './web'}),
            {'build': './web'},
        )

        self.assertEqual(
            config.merge_service_dicts({'image': 'redis'}, {'image': 'postgres'}),
            {'image': 'postgres'},
        )

    def test_merge_build_or_image_override_with_other(self):
        self.assertEqual(
            config.merge_service_dicts({'build': '.'}, {'image': 'redis'}),
            {'image': 'redis'}
        )

        self.assertEqual(
            config.merge_service_dicts({'image': 'redis'}, {'build': '.'}),
            {'build': '.'}
        )


class MergeListsTest(unittest.TestCase):
    def test_empty(self):
        service_dict = config.merge_service_dicts({}, {})
        self.assertNotIn('ports', service_dict)

    def test_no_override(self):
        service_dict = config.merge_service_dicts(
            {'ports': ['10:8000', '9000']},
            {},
        )
        self.assertEqual(set(service_dict['ports']), set(['10:8000', '9000']))

    def test_no_base(self):
        service_dict = config.merge_service_dicts(
            {},
            {'ports': ['10:8000', '9000']},
        )
        self.assertEqual(set(service_dict['ports']), set(['10:8000', '9000']))

    def test_add_item(self):
        service_dict = config.merge_service_dicts(
            {'ports': ['10:8000', '9000']},
            {'ports': ['20:8000']},
        )
        self.assertEqual(set(service_dict['ports']), set(['10:8000', '9000', '20:8000']))


class MergeStringsOrListsTest(unittest.TestCase):
    def test_no_override(self):
        service_dict = config.merge_service_dicts(
            {'dns': '8.8.8.8'},
            {},
        )
        self.assertEqual(set(service_dict['dns']), set(['8.8.8.8']))

    def test_no_base(self):
        service_dict = config.merge_service_dicts(
            {},
            {'dns': '8.8.8.8'},
        )
        self.assertEqual(set(service_dict['dns']), set(['8.8.8.8']))

    def test_add_string(self):
        service_dict = config.merge_service_dicts(
            {'dns': ['8.8.8.8']},
            {'dns': '9.9.9.9'},
        )
        self.assertEqual(set(service_dict['dns']), set(['8.8.8.8', '9.9.9.9']))

    def test_add_list(self):
        service_dict = config.merge_service_dicts(
            {'dns': '8.8.8.8'},
            {'dns': ['9.9.9.9']},
        )
        self.assertEqual(set(service_dict['dns']), set(['8.8.8.8', '9.9.9.9']))


class MergeLabelsTest(unittest.TestCase):
    def test_empty(self):
        service_dict = config.merge_service_dicts({}, {})
        self.assertNotIn('labels', service_dict)

    def test_no_override(self):
        service_dict = config.merge_service_dicts(
            make_service_dict('foo', {'build': '.', 'labels': ['foo=1', 'bar']}, 'tests/'),
            make_service_dict('foo', {'build': '.'}, 'tests/'),
        )
        self.assertEqual(service_dict['labels'], {'foo': '1', 'bar': ''})

    def test_no_base(self):
        service_dict = config.merge_service_dicts(
            make_service_dict('foo', {'build': '.'}, 'tests/'),
            make_service_dict('foo', {'build': '.', 'labels': ['foo=2']}, 'tests/'),
        )
        self.assertEqual(service_dict['labels'], {'foo': '2'})

    def test_override_explicit_value(self):
        service_dict = config.merge_service_dicts(
            make_service_dict('foo', {'build': '.', 'labels': ['foo=1', 'bar']}, 'tests/'),
            make_service_dict('foo', {'build': '.', 'labels': ['foo=2']}, 'tests/'),
        )
        self.assertEqual(service_dict['labels'], {'foo': '2', 'bar': ''})

    def test_add_explicit_value(self):
        service_dict = config.merge_service_dicts(
            make_service_dict('foo', {'build': '.', 'labels': ['foo=1', 'bar']}, 'tests/'),
            make_service_dict('foo', {'build': '.', 'labels': ['bar=2']}, 'tests/'),
        )
        self.assertEqual(service_dict['labels'], {'foo': '1', 'bar': '2'})

    def test_remove_explicit_value(self):
        service_dict = config.merge_service_dicts(
            make_service_dict('foo', {'build': '.', 'labels': ['foo=1', 'bar=2']}, 'tests/'),
            make_service_dict('foo', {'build': '.', 'labels': ['bar']}, 'tests/'),
        )
        self.assertEqual(service_dict['labels'], {'foo': '1', 'bar': ''})


class MemoryOptionsTest(unittest.TestCase):
    def test_validation_fails_with_just_memswap_limit(self):
        """
        When you set a 'memswap_limit' it is invalid config unless you also set
        a mem_limit
        """
        expected_error_msg = (
            "Service 'foo' configuration key 'memswap_limit' is invalid: when "
            "defining 'memswap_limit' you must set 'mem_limit' as well"
        )
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'foo': {'image': 'busybox', 'memswap_limit': 2000000},
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_validation_with_correct_memswap_values(self):
        service_dict = config.load(
            build_config_details(
                {'foo': {'image': 'busybox', 'mem_limit': 1000000, 'memswap_limit': 2000000}},
                'tests/fixtures/extends',
                'common.yml'
            )
        )
        self.assertEqual(service_dict[0]['memswap_limit'], 2000000)

    def test_memswap_can_be_a_string(self):
        service_dict = config.load(
            build_config_details(
                {'foo': {'image': 'busybox', 'mem_limit': "1G", 'memswap_limit': "512M"}},
                'tests/fixtures/extends',
                'common.yml'
            )
        )
        self.assertEqual(service_dict[0]['memswap_limit'], "512M")


class EnvTest(unittest.TestCase):
    def test_parse_environment_as_list(self):
        environment = [
            'NORMAL=F1',
            'CONTAINS_EQUALS=F=2',
            'TRAILING_EQUALS=',
        ]
        self.assertEqual(
            config.parse_environment(environment),
            {'NORMAL': 'F1', 'CONTAINS_EQUALS': 'F=2', 'TRAILING_EQUALS': ''},
        )

    def test_parse_environment_as_dict(self):
        environment = {
            'NORMAL': 'F1',
            'CONTAINS_EQUALS': 'F=2',
            'TRAILING_EQUALS': None,
        }
        self.assertEqual(config.parse_environment(environment), environment)

    def test_parse_environment_invalid(self):
        with self.assertRaises(ConfigurationError):
            config.parse_environment('a=b')

    def test_parse_environment_empty(self):
        self.assertEqual(config.parse_environment(None), {})

    @mock.patch.dict(os.environ)
    def test_resolve_environment(self):
        os.environ['FILE_DEF'] = 'E1'
        os.environ['FILE_DEF_EMPTY'] = 'E2'
        os.environ['ENV_DEF'] = 'E3'

        service_dict = make_service_dict(
            'foo', {
                'build': '.',
                'environment': {
                    'FILE_DEF': 'F1',
                    'FILE_DEF_EMPTY': '',
                    'ENV_DEF': None,
                    'NO_DEF': None
                },
            },
            'tests/'
        )

        self.assertEqual(
            service_dict['environment'],
            {'FILE_DEF': 'F1', 'FILE_DEF_EMPTY': '', 'ENV_DEF': 'E3', 'NO_DEF': ''},
        )

    def test_env_from_file(self):
        service_dict = make_service_dict(
            'foo',
            {'build': '.', 'env_file': 'one.env'},
            'tests/fixtures/env',
        )
        self.assertEqual(
            service_dict['environment'],
            {'ONE': '2', 'TWO': '1', 'THREE': '3', 'FOO': 'bar'},
        )

    def test_env_from_multiple_files(self):
        service_dict = make_service_dict(
            'foo',
            {'build': '.', 'env_file': ['one.env', 'two.env']},
            'tests/fixtures/env',
        )
        self.assertEqual(
            service_dict['environment'],
            {'ONE': '2', 'TWO': '1', 'THREE': '3', 'FOO': 'baz', 'DOO': 'dah'},
        )

    def test_env_nonexistent_file(self):
        options = {'env_file': 'nonexistent.env'}
        self.assertRaises(
            ConfigurationError,
            lambda: make_service_dict('foo', options, 'tests/fixtures/env'),
        )

    @mock.patch.dict(os.environ)
    def test_resolve_environment_from_file(self):
        os.environ['FILE_DEF'] = 'E1'
        os.environ['FILE_DEF_EMPTY'] = 'E2'
        os.environ['ENV_DEF'] = 'E3'
        service_dict = make_service_dict(
            'foo',
            {'build': '.', 'env_file': 'resolve.env'},
            'tests/fixtures/env',
        )
        self.assertEqual(
            service_dict['environment'],
            {
                'FILE_DEF': u'bär',
                'FILE_DEF_EMPTY': '',
                'ENV_DEF': 'E3',
                'NO_DEF': ''
            },
        )

    @pytest.mark.xfail(IS_WINDOWS_PLATFORM, reason='paths use slash')
    @mock.patch.dict(os.environ)
    def test_resolve_path(self):
        os.environ['HOSTENV'] = '/tmp'
        os.environ['CONTAINERENV'] = '/host/tmp'

        service_dict = config.load(
            build_config_details(
                {'foo': {'build': '.', 'volumes': ['$HOSTENV:$CONTAINERENV']}},
                "tests/fixtures/env",
            )
        )[0]
        self.assertEqual(
            set(service_dict['volumes']),
            set([VolumeSpec.parse('/tmp:/host/tmp')]))

        service_dict = config.load(
            build_config_details(
                {'foo': {'build': '.', 'volumes': ['/opt${HOSTENV}:/opt${CONTAINERENV}']}},
                "tests/fixtures/env",
            )
        )[0]
        self.assertEqual(
            set(service_dict['volumes']),
            set([VolumeSpec.parse('/opt/tmp:/opt/host/tmp')]))


def load_from_filename(filename):
    return config.load(config.find('.', [filename]))


class ExtendsTest(unittest.TestCase):
    def test_extends(self):
        service_dicts = load_from_filename('tests/fixtures/extends/docker-compose.yml')

        self.assertEqual(service_sort(service_dicts), service_sort([
            {
                'name': 'mydb',
                'image': 'busybox',
                'command': 'top',
            },
            {
                'name': 'myweb',
                'image': 'busybox',
                'command': 'top',
                'links': ['mydb:db'],
                'environment': {
                    "FOO": "1",
                    "BAR": "2",
                    "BAZ": "2",
                },
            }
        ]))

    def test_nested(self):
        service_dicts = load_from_filename('tests/fixtures/extends/nested.yml')

        self.assertEqual(service_dicts, [
            {
                'name': 'myweb',
                'image': 'busybox',
                'command': '/bin/true',
                'environment': {
                    "FOO": "2",
                    "BAR": "2",
                },
            },
        ])

    def test_self_referencing_file(self):
        """
        We specify a 'file' key that is the filename we're already in.
        """
        service_dicts = load_from_filename('tests/fixtures/extends/specify-file-as-self.yml')
        self.assertEqual(service_sort(service_dicts), service_sort([
            {
                'environment':
                {
                    'YEP': '1', 'BAR': '1', 'BAZ': '3'
                },
                'image': 'busybox',
                'name': 'myweb'
            },
            {
                'environment':
                {'YEP': '1'},
                'image': 'busybox',
                'name': 'otherweb'
            },
            {
                'environment':
                {'YEP': '1', 'BAZ': '3'},
                'image': 'busybox',
                'name': 'web'
            }
        ]))

    def test_circular(self):
        with pytest.raises(config.CircularReference) as exc:
            load_from_filename('tests/fixtures/extends/circle-1.yml')

        path = [
            (os.path.basename(filename), service_name)
            for (filename, service_name) in exc.value.trail
        ]
        expected = [
            ('circle-1.yml', 'web'),
            ('circle-2.yml', 'other'),
            ('circle-1.yml', 'web'),
        ]
        self.assertEqual(path, expected)

    def test_extends_validation_empty_dictionary(self):
        with self.assertRaisesRegexp(ConfigurationError, 'service'):
            config.load(
                build_config_details(
                    {
                        'web': {'image': 'busybox', 'extends': {}},
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_extends_validation_missing_service_key(self):
        with self.assertRaisesRegexp(ConfigurationError, "'service' is a required property"):
            config.load(
                build_config_details(
                    {
                        'web': {'image': 'busybox', 'extends': {'file': 'common.yml'}},
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_extends_validation_invalid_key(self):
        expected_error_msg = (
            "Service 'web' configuration key 'extends' "
            "contains unsupported option: 'rogue_key'"
        )
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'web': {
                            'image': 'busybox',
                            'extends': {
                                'file': 'common.yml',
                                'service': 'web',
                                'rogue_key': 'is not allowed'
                            }
                        },
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_extends_validation_sub_property_key(self):
        expected_error_msg = (
            "Service 'web' configuration key 'extends' 'file' contains 1, "
            "which is an invalid type, it should be a string"
        )
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            config.load(
                build_config_details(
                    {
                        'web': {
                            'image': 'busybox',
                            'extends': {
                                'file': 1,
                                'service': 'web',
                            }
                        },
                    },
                    'tests/fixtures/extends',
                    'filename.yml'
                )
            )

    def test_extends_validation_no_file_key_no_filename_set(self):
        dictionary = {'extends': {'service': 'web'}}

        def load_config():
            return make_service_dict('myweb', dictionary, working_dir='tests/fixtures/extends')

        self.assertRaisesRegexp(ConfigurationError, 'file', load_config)

    def test_extends_validation_valid_config(self):
        service = config.load(
            build_config_details(
                {
                    'web': {'image': 'busybox', 'extends': {'service': 'web', 'file': 'common.yml'}},
                },
                'tests/fixtures/extends',
                'common.yml'
            )
        )

        self.assertEquals(len(service), 1)
        self.assertIsInstance(service[0], dict)
        self.assertEquals(service[0]['command'], "/bin/true")

    def test_extended_service_with_invalid_config(self):
        expected_error_msg = "Service 'myweb' has neither an image nor a build path specified"

        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            load_from_filename('tests/fixtures/extends/service-with-invalid-schema.yml')

    def test_extended_service_with_valid_config(self):
        service = load_from_filename('tests/fixtures/extends/service-with-valid-composite-extends.yml')
        self.assertEquals(service[0]['command'], "top")

    def test_extends_file_defaults_to_self(self):
        """
        Test not specifying a file in our extends options that the
        config is valid and correctly extends from itself.
        """
        service_dicts = load_from_filename('tests/fixtures/extends/no-file-specified.yml')
        self.assertEqual(service_sort(service_dicts), service_sort([
            {
                'name': 'myweb',
                'image': 'busybox',
                'environment': {
                    "BAR": "1",
                    "BAZ": "3",
                }
            },
            {
                'name': 'web',
                'image': 'busybox',
                'environment': {
                    "BAZ": "3",
                }
            }
        ]))

    def test_invalid_links_in_extended_service(self):
        expected_error_msg = "services with 'links' cannot be extended"
        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            load_from_filename('tests/fixtures/extends/invalid-links.yml')

    def test_invalid_volumes_from_in_extended_service(self):
        expected_error_msg = "services with 'volumes_from' cannot be extended"

        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            load_from_filename('tests/fixtures/extends/invalid-volumes.yml')

    def test_invalid_net_in_extended_service(self):
        expected_error_msg = "services with 'net: container' cannot be extended"

        with self.assertRaisesRegexp(ConfigurationError, expected_error_msg):
            load_from_filename('tests/fixtures/extends/invalid-net.yml')

    @mock.patch.dict(os.environ)
    def test_valid_interpolation_in_extended_service(self):
        os.environ.update(
            HOSTNAME_VALUE="penguin",
        )
        expected_interpolated_value = "host-penguin"

        service_dicts = load_from_filename('tests/fixtures/extends/valid-interpolation.yml')
        for service in service_dicts:
            self.assertTrue(service['hostname'], expected_interpolated_value)

    @pytest.mark.xfail(IS_WINDOWS_PLATFORM, reason='paths use slash')
    def test_volume_path(self):
        dicts = load_from_filename('tests/fixtures/volume-path/docker-compose.yml')

        paths = [
            VolumeSpec(
                os.path.abspath('tests/fixtures/volume-path/common/foo'),
                '/foo',
                'rw'),
            VolumeSpec(
                os.path.abspath('tests/fixtures/volume-path/bar'),
                '/bar',
                'rw')
        ]

        self.assertEqual(set(dicts[0]['volumes']), set(paths))

    def test_parent_build_path_dne(self):
        child = load_from_filename('tests/fixtures/extends/nonexistent-path-child.yml')

        self.assertEqual(child, [
            {
                'name': 'dnechild',
                'image': 'busybox',
                'command': '/bin/true',
                'environment': {
                    "FOO": "1",
                    "BAR": "2",
                },
            },
        ])

    def test_load_throws_error_when_base_service_does_not_exist(self):
        err_msg = r'''Cannot extend service 'foo' in .*: Service not found'''
        with self.assertRaisesRegexp(ConfigurationError, err_msg):
            load_from_filename('tests/fixtures/extends/nonexistent-service.yml')

    def test_partial_service_config_in_extends_is_still_valid(self):
        dicts = load_from_filename('tests/fixtures/extends/valid-common-config.yml')
        self.assertEqual(dicts[0]['environment'], {'FOO': '1'})

    def test_extended_service_with_verbose_and_shorthand_way(self):
        services = load_from_filename('tests/fixtures/extends/verbose-and-shorthand.yml')
        self.assertEqual(service_sort(services), service_sort([
            {
                'name': 'base',
                'image': 'busybox',
                'environment': {'BAR': '1'},
            },
            {
                'name': 'verbose',
                'image': 'busybox',
                'environment': {'BAR': '1', 'FOO': '1'},
            },
            {
                'name': 'shorthand',
                'image': 'busybox',
                'environment': {'BAR': '1', 'FOO': '2'},
            },
        ]))

    def test_extends_with_environment_and_env_files(self):
        tmpdir = py.test.ensuretemp('test_extends_with_environment')
        self.addCleanup(tmpdir.remove)
        commondir = tmpdir.mkdir('common')
        commondir.join('base.yml').write("""
            app:
                image: 'example/app'
                env_file:
                    - 'envs'
                environment:
                    - SECRET
        """)
        tmpdir.join('docker-compose.yml').write("""
            ext:
                extends:
                    file: common/base.yml
                    service: app
                env_file:
                    - 'envs'
                environment:
                    - THING
        """)
        commondir.join('envs').write("""
            COMMON_ENV_FILE=1
        """)
        tmpdir.join('envs').write("""
            FROM_ENV_FILE=1
        """)

        expected = [
            {
                'name': 'ext',
                'image': 'example/app',
                'environment': {
                    'SECRET': 'secret',
                    'FROM_ENV_FILE': '1',
                    'COMMON_ENV_FILE': '1',
                    'THING': 'thing',
                },
            },
        ]
        with mock.patch.dict(os.environ):
            os.environ['SECRET'] = 'secret'
            os.environ['THING'] = 'thing'
            config = load_from_filename(str(tmpdir.join('docker-compose.yml')))

        assert config == expected


@pytest.mark.xfail(IS_WINDOWS_PLATFORM, reason='paths use slash')
class ExpandPathTest(unittest.TestCase):
    working_dir = '/home/user/somedir'

    def test_expand_path_normal(self):
        result = config.expand_path(self.working_dir, 'myfile')
        self.assertEqual(result, self.working_dir + '/' + 'myfile')

    def test_expand_path_absolute(self):
        abs_path = '/home/user/otherdir/somefile'
        result = config.expand_path(self.working_dir, abs_path)
        self.assertEqual(result, abs_path)

    def test_expand_path_with_tilde(self):
        test_path = '~/otherdir/somefile'
        with mock.patch.dict(os.environ):
            os.environ['HOME'] = user_path = '/home/user/'
            result = config.expand_path(self.working_dir, test_path)

        self.assertEqual(result, user_path + 'otherdir/somefile')


class VolumePathTest(unittest.TestCase):

    @pytest.mark.xfail((not IS_WINDOWS_PLATFORM), reason='does not have a drive')
    def test_split_path_mapping_with_windows_path(self):
        windows_volume_path = "c:\\Users\\msamblanet\\Documents\\anvil\\connect\\config:/opt/connect/config:ro"
        expected_mapping = (
            "/opt/connect/config:ro",
            "c:\\Users\\msamblanet\\Documents\\anvil\\connect\\config"
        )

        mapping = config.split_path_mapping(windows_volume_path)

        self.assertEqual(mapping, expected_mapping)


@pytest.mark.xfail(IS_WINDOWS_PLATFORM, reason='paths use slash')
class BuildPathTest(unittest.TestCase):
    def setUp(self):
        self.abs_context_path = os.path.join(os.getcwd(), 'tests/fixtures/build-ctx')

    def test_nonexistent_path(self):
        with self.assertRaises(ConfigurationError):
            config.load(
                build_config_details(
                    {
                        'foo': {'build': 'nonexistent.path'},
                    },
                    'working_dir',
                    'filename.yml'
                )
            )

    def test_relative_path(self):
        relative_build_path = '../build-ctx/'
        service_dict = make_service_dict(
            'relpath',
            {'build': relative_build_path},
            working_dir='tests/fixtures/build-path'
        )
        self.assertEquals(service_dict['build'], self.abs_context_path)

    def test_absolute_path(self):
        service_dict = make_service_dict(
            'abspath',
            {'build': self.abs_context_path},
            working_dir='tests/fixtures/build-path'
        )
        self.assertEquals(service_dict['build'], self.abs_context_path)

    def test_from_file(self):
        service_dict = load_from_filename('tests/fixtures/build-path/docker-compose.yml')
        self.assertEquals(service_dict, [{'name': 'foo', 'build': self.abs_context_path}])


class GetDefaultConfigFilesTestCase(unittest.TestCase):

    files = [
        'docker-compose.yml',
        'docker-compose.yaml',
        'fig.yml',
        'fig.yaml',
    ]

    def test_get_config_path_default_file_in_basedir(self):
        for index, filename in enumerate(self.files):
            self.assertEqual(
                filename,
                get_config_filename_for_files(self.files[index:]))
        with self.assertRaises(config.ComposeFileNotFound):
            get_config_filename_for_files([])

    def test_get_config_path_default_file_in_parent_dir(self):
        """Test with files placed in the subdir"""

        def get_config_in_subdir(files):
            return get_config_filename_for_files(files, subdir=True)

        for index, filename in enumerate(self.files):
            self.assertEqual(filename, get_config_in_subdir(self.files[index:]))
        with self.assertRaises(config.ComposeFileNotFound):
            get_config_in_subdir([])


def get_config_filename_for_files(filenames, subdir=None):
    def make_files(dirname, filenames):
        for fname in filenames:
            with open(os.path.join(dirname, fname), 'w') as f:
                f.write('')

    project_dir = tempfile.mkdtemp()
    try:
        make_files(project_dir, filenames)
        if subdir:
            base_dir = tempfile.mkdtemp(dir=project_dir)
        else:
            base_dir = project_dir
        filename, = config.get_default_config_files(base_dir)
        return os.path.basename(filename)
    finally:
        shutil.rmtree(project_dir)