from __future__ import absolute_import from __future__ import unicode_literals import docker import pytest from docker.errors import APIError from .. import mock from .. import unittest from compose.config.types import VolumeFromSpec from compose.config.types import VolumeSpec from compose.const import LABEL_CONFIG_HASH from compose.const import LABEL_ONE_OFF from compose.const import LABEL_PROJECT from compose.const import LABEL_SERVICE from compose.container import Container from compose.project import OneOffFilter from compose.service import build_ulimits from compose.service import build_volume_binding from compose.service import BuildAction from compose.service import ContainerNetworkMode from compose.service import get_container_data_volumes from compose.service import ImageType from compose.service import merge_volume_bindings from compose.service import NeedsBuildError from compose.service import NetworkMode from compose.service import NoSuchImageError from compose.service import parse_repository_tag from compose.service import Service from compose.service import ServiceNetworkMode from compose.service import warn_on_masked_volume class ServiceTest(unittest.TestCase): def setUp(self): self.mock_client = mock.create_autospec(docker.Client) def test_containers(self): service = Service('db', self.mock_client, 'myproject', image='foo') self.mock_client.containers.return_value = [] self.assertEqual(list(service.containers()), []) def test_containers_with_containers(self): self.mock_client.containers.return_value = [ dict(Name=str(i), Image='foo', Id=i) for i in range(3) ] service = Service('db', self.mock_client, 'myproject', image='foo') self.assertEqual([c.id for c in service.containers()], list(range(3))) expected_labels = [ '{0}=myproject'.format(LABEL_PROJECT), '{0}=db'.format(LABEL_SERVICE), '{0}=False'.format(LABEL_ONE_OFF), ] self.mock_client.containers.assert_called_once_with( all=False, filters={'label': expected_labels}) def test_container_without_name(self): self.mock_client.containers.return_value = [ {'Image': 'foo', 'Id': '1', 'Name': '1'}, {'Image': 'foo', 'Id': '2', 'Name': None}, {'Image': 'foo', 'Id': '3'}, ] service = Service('db', self.mock_client, 'myproject', image='foo') self.assertEqual([c.id for c in service.containers()], ['1']) self.assertEqual(service._next_container_number(), 2) self.assertEqual(service.get_container(1).id, '1') def test_get_volumes_from_container(self): container_id = 'aabbccddee' service = Service( 'test', image='foo', volumes_from=[ VolumeFromSpec( mock.Mock(id=container_id, spec=Container), 'rw', 'container')]) self.assertEqual(service._get_volumes_from(), [container_id + ':rw']) def test_get_volumes_from_container_read_only(self): container_id = 'aabbccddee' service = Service( 'test', image='foo', volumes_from=[ VolumeFromSpec( mock.Mock(id=container_id, spec=Container), 'ro', 'container')]) self.assertEqual(service._get_volumes_from(), [container_id + ':ro']) def test_get_volumes_from_service_container_exists(self): container_ids = ['aabbccddee', '12345'] from_service = mock.create_autospec(Service) from_service.containers.return_value = [ mock.Mock(id=container_id, spec=Container) for container_id in container_ids ] service = Service( 'test', volumes_from=[VolumeFromSpec(from_service, 'rw', 'service')], image='foo') self.assertEqual(service._get_volumes_from(), [container_ids[0] + ":rw"]) def test_get_volumes_from_service_container_exists_with_flags(self): for mode in ['ro', 'rw', 'z', 'rw,z', 'z,rw']: container_ids = ['aabbccddee:' + mode, '12345:' + mode] from_service = mock.create_autospec(Service) from_service.containers.return_value = [ mock.Mock(id=container_id.split(':')[0], spec=Container) for container_id in container_ids ] service = Service( 'test', volumes_from=[VolumeFromSpec(from_service, mode, 'service')], image='foo') self.assertEqual(service._get_volumes_from(), [container_ids[0]]) def test_get_volumes_from_service_no_container(self): container_id = 'abababab' from_service = mock.create_autospec(Service) from_service.containers.return_value = [] from_service.create_container.return_value = mock.Mock( id=container_id, spec=Container) service = Service( 'test', image='foo', volumes_from=[VolumeFromSpec(from_service, 'rw', 'service')]) self.assertEqual(service._get_volumes_from(), [container_id + ':rw']) from_service.create_container.assert_called_once_with() def test_split_domainname_none(self): service = Service('foo', image='foo', hostname='name', client=self.mock_client) opts = service._get_container_create_options({'image': 'foo'}, 1) self.assertEqual(opts['hostname'], 'name', 'hostname') self.assertFalse('domainname' in opts, 'domainname') def test_memory_swap_limit(self): self.mock_client.create_host_config.return_value = {} service = Service( name='foo', image='foo', hostname='name', client=self.mock_client, mem_limit=1000000000, memswap_limit=2000000000) service._get_container_create_options({'some': 'overrides'}, 1) self.assertTrue(self.mock_client.create_host_config.called) self.assertEqual( self.mock_client.create_host_config.call_args[1]['mem_limit'], 1000000000 ) self.assertEqual( self.mock_client.create_host_config.call_args[1]['memswap_limit'], 2000000000 ) def test_cgroup_parent(self): self.mock_client.create_host_config.return_value = {} service = Service( name='foo', image='foo', hostname='name', client=self.mock_client, cgroup_parent='test') service._get_container_create_options({'some': 'overrides'}, 1) self.assertTrue(self.mock_client.create_host_config.called) self.assertEqual( self.mock_client.create_host_config.call_args[1]['cgroup_parent'], 'test' ) def test_log_opt(self): self.mock_client.create_host_config.return_value = {} log_opt = {'syslog-address': 'tcp://192.168.0.42:123'} logging = {'driver': 'syslog', 'options': log_opt} service = Service( name='foo', image='foo', hostname='name', client=self.mock_client, log_driver='syslog', logging=logging) service._get_container_create_options({'some': 'overrides'}, 1) self.assertTrue(self.mock_client.create_host_config.called) self.assertEqual( self.mock_client.create_host_config.call_args[1]['log_config'], {'Type': 'syslog', 'Config': {'syslog-address': 'tcp://192.168.0.42:123'}} ) def test_split_domainname_fqdn(self): service = Service( 'foo', hostname='name.domain.tld', image='foo', client=self.mock_client) opts = service._get_container_create_options({'image': 'foo'}, 1) self.assertEqual(opts['hostname'], 'name', 'hostname') self.assertEqual(opts['domainname'], 'domain.tld', 'domainname') def test_split_domainname_both(self): service = Service( 'foo', hostname='name', image='foo', domainname='domain.tld', client=self.mock_client) opts = service._get_container_create_options({'image': 'foo'}, 1) self.assertEqual(opts['hostname'], 'name', 'hostname') self.assertEqual(opts['domainname'], 'domain.tld', 'domainname') def test_split_domainname_weird(self): service = Service( 'foo', hostname='name.sub', domainname='domain.tld', image='foo', client=self.mock_client) opts = service._get_container_create_options({'image': 'foo'}, 1) self.assertEqual(opts['hostname'], 'name.sub', 'hostname') self.assertEqual(opts['domainname'], 'domain.tld', 'domainname') def test_no_default_hostname_when_not_using_networking(self): service = Service( 'foo', image='foo', use_networking=False, client=self.mock_client, ) opts = service._get_container_create_options({'image': 'foo'}, 1) self.assertIsNone(opts.get('hostname')) def test_get_container_create_options_with_name_option(self): service = Service( 'foo', image='foo', client=self.mock_client, container_name='foo1') name = 'the_new_name' opts = service._get_container_create_options( {'name': name}, 1, one_off=OneOffFilter.only) self.assertEqual(opts['name'], name) def test_get_container_create_options_does_not_mutate_options(self): labels = {'thing': 'real'} environment = {'also': 'real'} service = Service( 'foo', image='foo', labels=dict(labels), client=self.mock_client, environment=dict(environment), ) self.mock_client.inspect_image.return_value = {'Id': 'abcd'} prev_container = mock.Mock( id='ababab', image_config={'ContainerConfig': {}}) prev_container.get.return_value = None opts = service._get_container_create_options( {}, 1, previous_container=prev_container) self.assertEqual(service.options['labels'], labels) self.assertEqual(service.options['environment'], environment) self.assertEqual( opts['labels'][LABEL_CONFIG_HASH], '2524a06fcb3d781aa2c981fc40bcfa08013bb318e4273bfa388df22023e6f2aa') assert opts['environment'] == ['also=real'] def test_get_container_create_options_sets_affinity_with_binds(self): service = Service( 'foo', image='foo', client=self.mock_client, ) self.mock_client.inspect_image.return_value = {'Id': 'abcd'} prev_container = mock.Mock( id='ababab', image_config={'ContainerConfig': {'Volumes': ['/data']}}) def container_get(key): return { 'Mounts': [ { 'Destination': '/data', 'Source': '/some/path', 'Name': 'abab1234', }, ] }.get(key, None) prev_container.get.side_effect = container_get opts = service._get_container_create_options( {}, 1, previous_container=prev_container) assert opts['environment'] == ['affinity:container==ababab'] def test_get_container_create_options_no_affinity_without_binds(self): service = Service('foo', image='foo', client=self.mock_client) self.mock_client.inspect_image.return_value = {'Id': 'abcd'} prev_container = mock.Mock( id='ababab', image_config={'ContainerConfig': {}}) prev_container.get.return_value = None opts = service._get_container_create_options( {}, 1, previous_container=prev_container) assert opts['environment'] == [] def test_get_container_not_found(self): self.mock_client.containers.return_value = [] service = Service('foo', client=self.mock_client, image='foo') self.assertRaises(ValueError, service.get_container) @mock.patch('compose.service.Container', autospec=True) def test_get_container(self, mock_container_class): container_dict = dict(Name='default_foo_2') self.mock_client.containers.return_value = [container_dict] service = Service('foo', image='foo', client=self.mock_client) container = service.get_container(number=2) self.assertEqual(container, mock_container_class.from_ps.return_value) mock_container_class.from_ps.assert_called_once_with( self.mock_client, container_dict) @mock.patch('compose.service.log', autospec=True) def test_pull_image(self, mock_log): service = Service('foo', client=self.mock_client, image='someimage:sometag') service.pull() self.mock_client.pull.assert_called_once_with( 'someimage', tag='sometag', stream=True) mock_log.info.assert_called_once_with('Pulling foo (someimage:sometag)...') def test_pull_image_no_tag(self): service = Service('foo', client=self.mock_client, image='ababab') service.pull() self.mock_client.pull.assert_called_once_with( 'ababab', tag='latest', stream=True) @mock.patch('compose.service.log', autospec=True) def test_pull_image_digest(self, mock_log): service = Service('foo', client=self.mock_client, image='someimage@sha256:1234') service.pull() self.mock_client.pull.assert_called_once_with( 'someimage', tag='sha256:1234', stream=True) mock_log.info.assert_called_once_with('Pulling foo (someimage@sha256:1234)...') @mock.patch('compose.service.Container', autospec=True) def test_recreate_container(self, _): mock_container = mock.create_autospec(Container) service = Service('foo', client=self.mock_client, image='someimage') service.image = lambda: {'Id': 'abc123'} new_container = service.recreate_container(mock_container) mock_container.stop.assert_called_once_with(timeout=10) mock_container.rename_to_tmp_name.assert_called_once_with() new_container.start.assert_called_once_with() mock_container.remove.assert_called_once_with() @mock.patch('compose.service.Container', autospec=True) def test_recreate_container_with_timeout(self, _): mock_container = mock.create_autospec(Container) self.mock_client.inspect_image.return_value = {'Id': 'abc123'} service = Service('foo', client=self.mock_client, image='someimage') service.recreate_container(mock_container, timeout=1) mock_container.stop.assert_called_once_with(timeout=1) def test_parse_repository_tag(self): self.assertEqual(parse_repository_tag("root"), ("root", "", ":")) self.assertEqual(parse_repository_tag("root:tag"), ("root", "tag", ":")) self.assertEqual(parse_repository_tag("user/repo"), ("user/repo", "", ":")) self.assertEqual(parse_repository_tag("user/repo:tag"), ("user/repo", "tag", ":")) self.assertEqual(parse_repository_tag("url:5000/repo"), ("url:5000/repo", "", ":")) self.assertEqual( parse_repository_tag("url:5000/repo:tag"), ("url:5000/repo", "tag", ":")) self.assertEqual( parse_repository_tag("root@sha256:digest"), ("root", "sha256:digest", "@")) self.assertEqual( parse_repository_tag("user/repo@sha256:digest"), ("user/repo", "sha256:digest", "@")) self.assertEqual( parse_repository_tag("url:5000/repo@sha256:digest"), ("url:5000/repo", "sha256:digest", "@")) def test_create_container(self): service = Service('foo', client=self.mock_client, build={'context': '.'}) self.mock_client.inspect_image.side_effect = [ NoSuchImageError, {'Id': 'abc123'}, ] self.mock_client.build.return_value = [ '{"stream": "Successfully built abcd"}', ] with mock.patch('compose.service.log', autospec=True) as mock_log: service.create_container() assert mock_log.warn.called _, args, _ = mock_log.warn.mock_calls[0] assert 'was built because it did not already exist' in args[0] self.mock_client.build.assert_called_once_with( tag='default_foo', dockerfile=None, stream=True, path='.', pull=False, forcerm=False, nocache=False, rm=True, buildargs=None, ) def test_ensure_image_exists_no_build(self): service = Service('foo', client=self.mock_client, build={'context': '.'}) self.mock_client.inspect_image.return_value = {'Id': 'abc123'} service.ensure_image_exists(do_build=BuildAction.skip) assert not self.mock_client.build.called def test_ensure_image_exists_no_build_but_needs_build(self): service = Service('foo', client=self.mock_client, build={'context': '.'}) self.mock_client.inspect_image.side_effect = NoSuchImageError with pytest.raises(NeedsBuildError): service.ensure_image_exists(do_build=BuildAction.skip) def test_ensure_image_exists_force_build(self): service = Service('foo', client=self.mock_client, build={'context': '.'}) self.mock_client.inspect_image.return_value = {'Id': 'abc123'} self.mock_client.build.return_value = [ '{"stream": "Successfully built abcd"}', ] with mock.patch('compose.service.log', autospec=True) as mock_log: service.ensure_image_exists(do_build=BuildAction.force) assert not mock_log.warn.called self.mock_client.build.assert_called_once_with( tag='default_foo', dockerfile=None, stream=True, path='.', pull=False, forcerm=False, nocache=False, rm=True, buildargs=None, ) def test_build_does_not_pull(self): self.mock_client.build.return_value = [ b'{"stream": "Successfully built 12345"}', ] service = Service('foo', client=self.mock_client, build={'context': '.'}) service.build() self.assertEqual(self.mock_client.build.call_count, 1) self.assertFalse(self.mock_client.build.call_args[1]['pull']) def test_config_dict(self): self.mock_client.inspect_image.return_value = {'Id': 'abcd'} service = Service( 'foo', image='example.com/foo', client=self.mock_client, network_mode=ServiceNetworkMode(Service('other')), networks={'default': None}, links=[(Service('one'), 'one')], volumes_from=[VolumeFromSpec(Service('two'), 'rw', 'service')]) config_dict = service.config_dict() expected = { 'image_id': 'abcd', 'options': {'image': 'example.com/foo'}, 'links': [('one', 'one')], 'net': 'other', 'networks': {'default': None}, 'volumes_from': [('two', 'rw')], } assert config_dict == expected def test_config_dict_with_network_mode_from_container(self): self.mock_client.inspect_image.return_value = {'Id': 'abcd'} container = Container( self.mock_client, {'Id': 'aaabbb', 'Name': '/foo_1'}) service = Service( 'foo', image='example.com/foo', client=self.mock_client, network_mode=ContainerNetworkMode(container)) config_dict = service.config_dict() expected = { 'image_id': 'abcd', 'options': {'image': 'example.com/foo'}, 'links': [], 'networks': {}, 'net': 'aaabbb', 'volumes_from': [], } assert config_dict == expected def test_remove_image_none(self): web = Service('web', image='example', client=self.mock_client) assert not web.remove_image(ImageType.none) assert not self.mock_client.remove_image.called def test_remove_image_local_with_image_name_doesnt_remove(self): web = Service('web', image='example', client=self.mock_client) assert not web.remove_image(ImageType.local) assert not self.mock_client.remove_image.called def test_remove_image_local_without_image_name_does_remove(self): web = Service('web', build='.', client=self.mock_client) assert web.remove_image(ImageType.local) self.mock_client.remove_image.assert_called_once_with(web.image_name) def test_remove_image_all_does_remove(self): web = Service('web', image='example', client=self.mock_client) assert web.remove_image(ImageType.all) self.mock_client.remove_image.assert_called_once_with(web.image_name) def test_remove_image_with_error(self): self.mock_client.remove_image.side_effect = error = APIError( message="testing", response={}, explanation="Boom") web = Service('web', image='example', client=self.mock_client) with mock.patch('compose.service.log', autospec=True) as mock_log: assert not web.remove_image(ImageType.all) mock_log.error.assert_called_once_with( "Failed to remove image for service %s: %s", web.name, error) def test_specifies_host_port_with_no_ports(self): service = Service( 'foo', image='foo') self.assertEqual(service.specifies_host_port(), False) def test_specifies_host_port_with_container_port(self): service = Service( 'foo', image='foo', ports=["2000"]) self.assertEqual(service.specifies_host_port(), False) def test_specifies_host_port_with_host_port(self): service = Service( 'foo', image='foo', ports=["1000:2000"]) self.assertEqual(service.specifies_host_port(), True) def test_specifies_host_port_with_host_ip_no_port(self): service = Service( 'foo', image='foo', ports=["127.0.0.1::2000"]) self.assertEqual(service.specifies_host_port(), False) def test_specifies_host_port_with_host_ip_and_port(self): service = Service( 'foo', image='foo', ports=["127.0.0.1:1000:2000"]) self.assertEqual(service.specifies_host_port(), True) def test_specifies_host_port_with_container_port_range(self): service = Service( 'foo', image='foo', ports=["2000-3000"]) self.assertEqual(service.specifies_host_port(), False) def test_specifies_host_port_with_host_port_range(self): service = Service( 'foo', image='foo', ports=["1000-2000:2000-3000"]) self.assertEqual(service.specifies_host_port(), True) def test_specifies_host_port_with_host_ip_no_port_range(self): service = Service( 'foo', image='foo', ports=["127.0.0.1::2000-3000"]) self.assertEqual(service.specifies_host_port(), False) def test_specifies_host_port_with_host_ip_and_port_range(self): service = Service( 'foo', image='foo', ports=["127.0.0.1:1000-2000:2000-3000"]) self.assertEqual(service.specifies_host_port(), True) def test_image_name_from_config(self): image_name = 'example/web:latest' service = Service('foo', image=image_name) assert service.image_name == image_name def test_image_name_default(self): service = Service('foo', project='testing') assert service.image_name == 'testing_foo' @mock.patch('compose.service.log', autospec=True) def test_only_log_warning_when_host_ports_clash(self, mock_log): self.mock_client.inspect_image.return_value = {'Id': 'abcd'} name = 'foo' service = Service( name, client=self.mock_client, ports=["8080:80"]) service.scale(0) self.assertFalse(mock_log.warn.called) service.scale(1) self.assertFalse(mock_log.warn.called) service.scale(2) mock_log.warn.assert_called_once_with( 'The "{}" service specifies a port on the host. If multiple containers ' 'for this service are created on a single host, the port will clash.'.format(name)) class TestServiceNetwork(object): def test_connect_container_to_networks_short_aliase_exists(self): mock_client = mock.create_autospec(docker.Client) service = Service( 'db', mock_client, 'myproject', image='foo', networks={'project_default': {}}) container = Container( None, { 'Id': 'abcdef', 'NetworkSettings': { 'Networks': { 'project_default': { 'Aliases': ['analias', 'abcdef'], }, }, }, }, True) service.connect_container_to_networks(container) assert not mock_client.disconnect_container_from_network.call_count assert not mock_client.connect_container_to_network.call_count def sort_by_name(dictionary_list): return sorted(dictionary_list, key=lambda k: k['name']) class BuildUlimitsTestCase(unittest.TestCase): def test_build_ulimits_with_dict(self): ulimits = build_ulimits( { 'nofile': {'soft': 10000, 'hard': 20000}, 'nproc': {'soft': 65535, 'hard': 65535} } ) expected = [ {'name': 'nofile', 'soft': 10000, 'hard': 20000}, {'name': 'nproc', 'soft': 65535, 'hard': 65535} ] assert sort_by_name(ulimits) == sort_by_name(expected) def test_build_ulimits_with_ints(self): ulimits = build_ulimits({'nofile': 20000, 'nproc': 65535}) expected = [ {'name': 'nofile', 'soft': 20000, 'hard': 20000}, {'name': 'nproc', 'soft': 65535, 'hard': 65535} ] assert sort_by_name(ulimits) == sort_by_name(expected) def test_build_ulimits_with_integers_and_dicts(self): ulimits = build_ulimits( { 'nproc': 65535, 'nofile': {'soft': 10000, 'hard': 20000} } ) expected = [ {'name': 'nofile', 'soft': 10000, 'hard': 20000}, {'name': 'nproc', 'soft': 65535, 'hard': 65535} ] assert sort_by_name(ulimits) == sort_by_name(expected) class NetTestCase(unittest.TestCase): def test_network_mode(self): network_mode = NetworkMode('host') self.assertEqual(network_mode.id, 'host') self.assertEqual(network_mode.mode, 'host') self.assertEqual(network_mode.service_name, None) def test_network_mode_container(self): container_id = 'abcd' network_mode = ContainerNetworkMode(Container(None, {'Id': container_id})) self.assertEqual(network_mode.id, container_id) self.assertEqual(network_mode.mode, 'container:' + container_id) self.assertEqual(network_mode.service_name, None) def test_network_mode_service(self): container_id = 'bbbb' service_name = 'web' mock_client = mock.create_autospec(docker.Client) mock_client.containers.return_value = [ {'Id': container_id, 'Name': container_id, 'Image': 'abcd'}, ] service = Service(name=service_name, client=mock_client) network_mode = ServiceNetworkMode(service) self.assertEqual(network_mode.id, service_name) self.assertEqual(network_mode.mode, 'container:' + container_id) self.assertEqual(network_mode.service_name, service_name) def test_network_mode_service_no_containers(self): service_name = 'web' mock_client = mock.create_autospec(docker.Client) mock_client.containers.return_value = [] service = Service(name=service_name, client=mock_client) network_mode = ServiceNetworkMode(service) self.assertEqual(network_mode.id, service_name) self.assertEqual(network_mode.mode, None) self.assertEqual(network_mode.service_name, service_name) def build_mount(destination, source, mode='rw'): return {'Source': source, 'Destination': destination, 'Mode': mode} class ServiceVolumesTest(unittest.TestCase): def setUp(self): self.mock_client = mock.create_autospec(docker.Client) def test_build_volume_binding(self): binding = build_volume_binding(VolumeSpec.parse('/outside:/inside')) assert binding == ('/inside', '/outside:/inside:rw') def test_get_container_data_volumes(self): options = [VolumeSpec.parse(v) for v in [ '/host/volume:/host/volume:ro', '/new/volume', '/existing/volume', 'named:/named/vol', ]] self.mock_client.inspect_image.return_value = { 'ContainerConfig': { 'Volumes': { '/mnt/image/data': {}, } } } container = Container(self.mock_client, { 'Image': 'ababab', 'Mounts': [ { 'Source': '/host/volume', 'Destination': '/host/volume', 'Mode': '', 'RW': True, 'Name': 'hostvolume', }, { 'Source': '/var/lib/docker/aaaaaaaa', 'Destination': '/existing/volume', 'Mode': '', 'RW': True, 'Name': 'existingvolume', }, { 'Source': '/var/lib/docker/bbbbbbbb', 'Destination': '/removed/volume', 'Mode': '', 'RW': True, 'Name': 'removedvolume', }, { 'Source': '/var/lib/docker/cccccccc', 'Destination': '/mnt/image/data', 'Mode': '', 'RW': True, 'Name': 'imagedata', }, ] }, has_been_inspected=True) expected = [ VolumeSpec.parse('existingvolume:/existing/volume:rw'), VolumeSpec.parse('imagedata:/mnt/image/data:rw'), ] volumes = get_container_data_volumes(container, options) assert sorted(volumes) == sorted(expected) def test_merge_volume_bindings(self): options = [ VolumeSpec.parse('/host/volume:/host/volume:ro'), VolumeSpec.parse('/host/rw/volume:/host/rw/volume'), VolumeSpec.parse('/new/volume'), VolumeSpec.parse('/existing/volume'), ] self.mock_client.inspect_image.return_value = { 'ContainerConfig': {'Volumes': {}} } previous_container = Container(self.mock_client, { 'Id': 'cdefab', 'Image': 'ababab', 'Mounts': [{ 'Source': '/var/lib/docker/aaaaaaaa', 'Destination': '/existing/volume', 'Mode': '', 'RW': True, 'Name': 'existingvolume', }], }, has_been_inspected=True) expected = [ '/host/volume:/host/volume:ro', '/host/rw/volume:/host/rw/volume:rw', 'existingvolume:/existing/volume:rw', ] binds, affinity = merge_volume_bindings(options, previous_container) assert sorted(binds) == sorted(expected) assert affinity == {'affinity:container': '=cdefab'} def test_mount_same_host_path_to_two_volumes(self): service = Service( 'web', image='busybox', volumes=[ VolumeSpec.parse('/host/path:/data1'), VolumeSpec.parse('/host/path:/data2'), ], client=self.mock_client, ) self.mock_client.inspect_image.return_value = { 'Id': 'ababab', 'ContainerConfig': { 'Volumes': {} } } service._get_container_create_options( override_options={}, number=1, ) self.assertEqual( set(self.mock_client.create_host_config.call_args[1]['binds']), set([ '/host/path:/data1:rw', '/host/path:/data2:rw', ]), ) def test_get_container_create_options_with_different_host_path_in_container_json(self): service = Service( 'web', image='busybox', volumes=[VolumeSpec.parse('/host/path:/data')], client=self.mock_client, ) volume_name = 'abcdefff1234' self.mock_client.inspect_image.return_value = { 'Id': 'ababab', 'ContainerConfig': { 'Volumes': { '/data': {}, } } } self.mock_client.inspect_container.return_value = { 'Id': '123123123', 'Image': 'ababab', 'Mounts': [ { 'Destination': '/data', 'Source': '/mnt/sda1/host/path', 'Mode': '', 'RW': True, 'Driver': 'local', 'Name': volume_name, }, ] } service._get_container_create_options( override_options={}, number=1, previous_container=Container(self.mock_client, {'Id': '123123123'}), ) assert ( self.mock_client.create_host_config.call_args[1]['binds'] == ['{}:/data:rw'.format(volume_name)] ) def test_warn_on_masked_volume_no_warning_when_no_container_volumes(self): volumes_option = [VolumeSpec('/home/user', '/path', 'rw')] container_volumes = [] service = 'service_name' with mock.patch('compose.service.log', autospec=True) as mock_log: warn_on_masked_volume(volumes_option, container_volumes, service) assert not mock_log.warn.called def test_warn_on_masked_volume_when_masked(self): volumes_option = [VolumeSpec('/home/user', '/path', 'rw')] container_volumes = [ VolumeSpec('/var/lib/docker/path', '/path', 'rw'), VolumeSpec('/var/lib/docker/path', '/other', 'rw'), ] service = 'service_name' with mock.patch('compose.service.log', autospec=True) as mock_log: warn_on_masked_volume(volumes_option, container_volumes, service) mock_log.warn.assert_called_once_with(mock.ANY) def test_warn_on_masked_no_warning_with_same_path(self): volumes_option = [VolumeSpec('/home/user', '/path', 'rw')] container_volumes = [VolumeSpec('/home/user', '/path', 'rw')] service = 'service_name' with mock.patch('compose.service.log', autospec=True) as mock_log: warn_on_masked_volume(volumes_option, container_volumes, service) assert not mock_log.warn.called def test_warn_on_masked_no_warning_with_container_only_option(self): volumes_option = [VolumeSpec(None, '/path', 'rw')] container_volumes = [ VolumeSpec('/var/lib/docker/volume/path', '/path', 'rw') ] service = 'service_name' with mock.patch('compose.service.log', autospec=True) as mock_log: warn_on_masked_volume(volumes_option, container_volumes, service) assert not mock_log.warn.called def test_create_with_special_volume_mode(self): self.mock_client.inspect_image.return_value = {'Id': 'imageid'} self.mock_client.create_container.return_value = {'Id': 'containerid'} volume = '/tmp:/foo:z' Service( 'web', client=self.mock_client, image='busybox', volumes=[VolumeSpec.parse(volume)], ).create_container() assert self.mock_client.create_container.call_count == 1 self.assertEqual( self.mock_client.create_host_config.call_args[1]['binds'], [volume])