from __future__ import unicode_literals from __future__ import absolute_import from .. import unittest import mock import docker from docker.utils import LogConfig from .. import mock from .. import unittest 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.service import ConfigError from compose.service import ContainerNet from compose.service import NeedsBuildError from compose.service import Net from compose.service import NoSuchImageError from compose.service import Service from compose.service import ServiceNet from compose.service import build_port_bindings from compose.service import build_volume_binding from compose.service import get_container_data_volumes from compose.service import merge_volume_bindings from compose.service import parse_repository_tag from compose.service import parse_volume_spec from compose.service import split_port class ServiceTest(unittest.TestCase): def setUp(self): self.mock_client = mock.create_autospec(docker.Client) def test_name_validations(self): self.assertRaises(ConfigError, lambda: Service(name='', image='foo')) self.assertRaises(ConfigError, lambda: Service(name=' ', image='foo')) self.assertRaises(ConfigError, lambda: Service(name='/', image='foo')) self.assertRaises(ConfigError, lambda: Service(name='!', image='foo')) self.assertRaises(ConfigError, lambda: Service(name='\xe2', image='foo')) Service('a', image='foo') Service('foo', image='foo') Service('foo-bar', image='foo') Service('foo.bar', image='foo') Service('foo_bar', image='foo') Service('_', image='foo') Service('___', image='foo') Service('-', image='foo') Service('--', image='foo') Service('.__.', image='foo') def test_project_validation(self): self.assertRaises(ConfigError, lambda: Service('bar')) self.assertRaises(ConfigError, lambda: Service(name='foo', project='>', image='foo')) Service(name='foo', project='bar.bar__', image='foo') def test_containers(self): service = Service('db', self.mock_client, 'myproject', image='foo') self.mock_client.containers.return_value = [] self.assertEqual(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()], 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=[mock.Mock(id=container_id, spec=Container)]) self.assertEqual(service._get_volumes_from(), [container_id]) 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=[from_service], image='foo') self.assertEqual(service._get_volumes_from(), container_ids) 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=[from_service]) self.assertEqual(service._get_volumes_from(), [container_id]) from_service.create_container.assert_called_once_with() def test_split_port_with_host_ip(self): internal_port, external_port = split_port("127.0.0.1:1000:2000") self.assertEqual(internal_port, "2000") self.assertEqual(external_port, ("127.0.0.1", "1000")) def test_split_port_with_protocol(self): internal_port, external_port = split_port("127.0.0.1:1000:2000/udp") self.assertEqual(internal_port, "2000/udp") self.assertEqual(external_port, ("127.0.0.1", "1000")) def test_split_port_with_host_ip_no_port(self): internal_port, external_port = split_port("127.0.0.1::2000") self.assertEqual(internal_port, "2000") self.assertEqual(external_port, ("127.0.0.1", None)) def test_split_port_with_host_port(self): internal_port, external_port = split_port("1000:2000") self.assertEqual(internal_port, "2000") self.assertEqual(external_port, "1000") def test_split_port_no_host_port(self): internal_port, external_port = split_port("2000") self.assertEqual(internal_port, "2000") self.assertEqual(external_port, None) def test_split_port_invalid(self): with self.assertRaises(ConfigError): split_port("0.0.0.0:1000:2000:tcp") def test_build_port_bindings_with_one_port(self): port_bindings = build_port_bindings(["127.0.0.1:1000:1000"]) self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")]) def test_build_port_bindings_with_matching_internal_ports(self): port_bindings = build_port_bindings(["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"]) self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000"), ("127.0.0.1", "2000")]) def test_build_port_bindings_with_nonmatching_internal_ports(self): port_bindings = build_port_bindings(["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"]) self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")]) self.assertEqual(port_bindings["2000"], [("127.0.0.1", "2000")]) def test_split_domainname_none(self): service = Service('foo', image='foo', hostname='name', client=self.mock_client) self.mock_client.containers.return_value = [] 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): service = Service(name='foo', image='foo', hostname='name', client=self.mock_client, mem_limit=1000000000, memswap_limit=2000000000) self.mock_client.containers.return_value = [] opts = service._get_container_create_options({'some': 'overrides'}, 1) self.assertEqual(opts['host_config']['MemorySwap'], 2000000000) self.assertEqual(opts['host_config']['Memory'], 1000000000) def test_log_opt(self): log_opt = {'address': 'tcp://192.168.0.42:123'} service = Service(name='foo', image='foo', hostname='name', client=self.mock_client, log_driver='syslog', log_opt=log_opt) self.mock_client.containers.return_value = [] opts = service._get_container_create_options({'some': 'overrides'}, 1) self.assertIsInstance(opts['host_config']['LogConfig'], LogConfig) self.assertEqual(opts['host_config']['LogConfig'].type, 'syslog') self.assertEqual(opts['host_config']['LogConfig'].config, log_opt) def test_split_domainname_fqdn(self): service = Service( 'foo', hostname='name.domain.tld', image='foo', client=self.mock_client) self.mock_client.containers.return_value = [] 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) self.mock_client.containers.return_value = [] 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) self.mock_client.containers.return_value = [] 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_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=True) 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': {}}) 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], '3c85881a8903b9d73a06c41860c8be08acce1494ab4cf8408375966dccd714de') self.assertEqual( opts['environment'], { 'affinity:container': '=ababab', 'also': 'real', } ) 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.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) self.mock_client.rename.assert_called_once_with( mock_container.id, '%s_%s' % (mock_container.short_id, mock_container.name)) 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")) @mock.patch('compose.service.Container', autospec=True) def test_create_container_latest_is_used_when_no_tag_specified(self, mock_container): service = Service('foo', client=self.mock_client, image='someimage') images = [] def pull(repo, tag=None, **kwargs): self.assertEqual('someimage', repo) self.assertEqual('latest', tag) images.append({'Id': 'abc123'}) return [] service.image = lambda *args, **kwargs: mock_get_image(images) self.mock_client.pull = pull service.create_container() self.assertEqual(1, len(images)) def test_create_container_with_build(self): service = Service('foo', client=self.mock_client, build='.') images = [] service.image = lambda *args, **kwargs: mock_get_image(images) service.build = lambda: images.append({'Id': 'abc123'}) service.create_container(do_build=True) self.assertEqual(1, len(images)) def test_create_container_no_build(self): service = Service('foo', client=self.mock_client, build='.') service.image = lambda: {'Id': 'abc123'} service.create_container(do_build=False) self.assertFalse(self.mock_client.build.called) def test_create_container_no_build_but_needs_build(self): service = Service('foo', client=self.mock_client, build='.') service.image = lambda *args, **kwargs: mock_get_image([]) with self.assertRaises(NeedsBuildError): service.create_container(do_build=False) def test_build_does_not_pull(self): self.mock_client.build.return_value = [ '{"stream": "Successfully built 12345"}', ] service = Service('foo', client=self.mock_client, build='.') 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, net=ServiceNet(Service('other')), links=[(Service('one'), 'one')], volumes_from=[Service('two')]) config_dict = service.config_dict() expected = { 'image_id': 'abcd', 'options': {'image': 'example.com/foo'}, 'links': [('one', 'one')], 'net': 'other', 'volumes_from': ['two'], } self.assertEqual(config_dict, expected) def test_config_dict_with_net_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, net=container) config_dict = service.config_dict() expected = { 'image_id': 'abcd', 'options': {'image': 'example.com/foo'}, 'links': [], 'net': 'aaabbb', 'volumes_from': [], } self.assertEqual(config_dict, expected) class NetTestCase(unittest.TestCase): def test_net(self): net = Net('host') self.assertEqual(net.id, 'host') self.assertEqual(net.mode, 'host') self.assertEqual(net.service_name, None) def test_net_container(self): container_id = 'abcd' net = ContainerNet(Container(None, {'Id': container_id})) self.assertEqual(net.id, container_id) self.assertEqual(net.mode, 'container:' + container_id) self.assertEqual(net.service_name, None) def test_net_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) net = ServiceNet(service) self.assertEqual(net.id, service_name) self.assertEqual(net.mode, 'container:' + container_id) self.assertEqual(net.service_name, service_name) def test_net_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) net = ServiceNet(service) self.assertEqual(net.id, service_name) self.assertEqual(net.mode, None) self.assertEqual(net.service_name, service_name) def mock_get_image(images): if images: return images[0] else: raise NoSuchImageError() class ServiceVolumesTest(unittest.TestCase): def setUp(self): self.mock_client = mock.create_autospec(docker.Client) def test_parse_volume_spec_only_one_path(self): spec = parse_volume_spec('/the/volume') self.assertEqual(spec, (None, '/the/volume', 'rw')) def test_parse_volume_spec_internal_and_external(self): spec = parse_volume_spec('external:interval') self.assertEqual(spec, ('external', 'interval', 'rw')) def test_parse_volume_spec_with_mode(self): spec = parse_volume_spec('external:interval:ro') self.assertEqual(spec, ('external', 'interval', 'ro')) spec = parse_volume_spec('external:interval:z') self.assertEqual(spec, ('external', 'interval', 'z')) def test_parse_volume_spec_too_many_parts(self): with self.assertRaises(ConfigError): parse_volume_spec('one:two:three:four') def test_build_volume_binding(self): binding = build_volume_binding(parse_volume_spec('/outside:/inside')) self.assertEqual(binding, ('/inside', '/outside:/inside:rw')) def test_get_container_data_volumes(self): options = [ '/host/volume:/host/volume:ro', '/new/volume', '/existing/volume', ] self.mock_client.inspect_image.return_value = { 'ContainerConfig': { 'Volumes': { '/mnt/image/data': {}, } } } container = Container(self.mock_client, { 'Image': 'ababab', 'Volumes': { '/host/volume': '/host/volume', '/existing/volume': '/var/lib/docker/aaaaaaaa', '/removed/volume': '/var/lib/docker/bbbbbbbb', '/mnt/image/data': '/var/lib/docker/cccccccc', }, }, has_been_inspected=True) expected = { '/existing/volume': '/var/lib/docker/aaaaaaaa:/existing/volume:rw', '/mnt/image/data': '/var/lib/docker/cccccccc:/mnt/image/data:rw', } binds = get_container_data_volumes(container, options) self.assertEqual(binds, expected) def test_merge_volume_bindings(self): options = [ '/host/volume:/host/volume:ro', '/host/rw/volume:/host/rw/volume', '/new/volume', '/existing/volume', ] self.mock_client.inspect_image.return_value = { 'ContainerConfig': {'Volumes': {}} } intermediate_container = Container(self.mock_client, { 'Image': 'ababab', 'Volumes': {'/existing/volume': '/var/lib/docker/aaaaaaaa'}, }, has_been_inspected=True) expected = [ '/host/volume:/host/volume:ro', '/host/rw/volume:/host/rw/volume:rw', '/var/lib/docker/aaaaaaaa:/existing/volume:rw', ] binds = merge_volume_bindings(options, intermediate_container) self.assertEqual(set(binds), set(expected)) def test_mount_same_host_path_to_two_volumes(self): service = Service( 'web', image='busybox', volumes=[ '/host/path:/data1', '/host/path:/data2', ], client=self.mock_client, ) self.mock_client.inspect_image.return_value = { 'Id': 'ababab', 'ContainerConfig': { 'Volumes': {} } } create_options = service._get_container_create_options( override_options={}, number=1, ) self.assertEqual( set(create_options['host_config']['Binds']), set([ '/host/path:/data1:rw', '/host/path:/data2:rw', ]), ) def test_different_host_path_in_container_json(self): service = Service( 'web', image='busybox', volumes=['/host/path:/data'], client=self.mock_client, ) self.mock_client.inspect_image.return_value = { 'Id': 'ababab', 'ContainerConfig': { 'Volumes': { '/data': {}, } } } self.mock_client.inspect_container.return_value = { 'Id': '123123123', 'Image': 'ababab', 'Volumes': { '/data': '/mnt/sda1/host/path', }, } create_options = service._get_container_create_options( override_options={}, number=1, previous_container=Container(self.mock_client, {'Id': '123123123'}), ) self.assertEqual( create_options['host_config']['Binds'], ['/mnt/sda1/host/path:/data:rw'], ) def test_create_with_special_volume_mode(self): self.mock_client.inspect_image.return_value = {'Id': 'imageid'} create_calls = [] def create_container(*args, **kwargs): create_calls.append((args, kwargs)) return {'Id': 'containerid'} self.mock_client.create_container = create_container volumes = ['/tmp:/foo:z'] Service( 'web', client=self.mock_client, image='busybox', volumes=volumes, ).create_container() self.assertEqual(len(create_calls), 1) self.assertEqual(create_calls[0][1]['host_config']['Binds'], volumes)