mirror of https://github.com/docker/docs.git
1520 lines
54 KiB
Python
1520 lines
54 KiB
Python
# 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)
|