# encoding: utf-8 from __future__ import absolute_import from __future__ import unicode_literals import datetime import docker import pytest from docker.errors import NotFound from .. import mock from .. import unittest from compose.config.config import Config from compose.config.types import VolumeFromSpec from compose.const import COMPOSEFILE_V1 as V1 from compose.const import COMPOSEFILE_V2_0 as V2_0 from compose.const import COMPOSEFILE_V2_4 as V2_4 from compose.const import LABEL_SERVICE from compose.container import Container from compose.errors import OperationFailedError from compose.project import NoSuchService from compose.project import Project from compose.project import ProjectError from compose.service import ImageType from compose.service import Service class ProjectTest(unittest.TestCase): def setUp(self): self.mock_client = mock.create_autospec(docker.APIClient) self.mock_client._general_configs = {} self.mock_client.api_version = docker.constants.DEFAULT_DOCKER_API_VERSION def test_from_config_v1(self): config = Config( version=V1, services=[ { 'name': 'web', 'image': 'busybox:latest', }, { 'name': 'db', 'image': 'busybox:latest', }, ], networks=None, volumes=None, secrets=None, configs=None, ) project = Project.from_config( name='composetest', config_data=config, client=None, ) assert len(project.services) == 2 assert project.get_service('web').name == 'web' assert project.get_service('web').options['image'] == 'busybox:latest' assert project.get_service('db').name == 'db' assert project.get_service('db').options['image'] == 'busybox:latest' assert not project.networks.use_networking @mock.patch('compose.network.Network.true_name', lambda n: n.full_name) def test_from_config_v2(self): config = Config( version=V2_0, services=[ { 'name': 'web', 'image': 'busybox:latest', }, { 'name': 'db', 'image': 'busybox:latest', }, ], networks=None, volumes=None, secrets=None, configs=None, ) project = Project.from_config('composetest', config, None) assert len(project.services) == 2 assert project.networks.use_networking def test_get_service(self): web = Service( project='composetest', name='web', client=None, image="busybox:latest", ) project = Project('test', [web], None) assert project.get_service('web') == web def test_get_services_returns_all_services_without_args(self): web = Service( project='composetest', name='web', image='foo', ) console = Service( project='composetest', name='console', image='foo', ) project = Project('test', [web, console], None) assert project.get_services() == [web, console] def test_get_services_returns_listed_services_with_args(self): web = Service( project='composetest', name='web', image='foo', ) console = Service( project='composetest', name='console', image='foo', ) project = Project('test', [web, console], None) assert project.get_services(['console']) == [console] def test_get_services_with_include_links(self): db = Service( project='composetest', name='db', image='foo', ) web = Service( project='composetest', name='web', image='foo', links=[(db, 'database')] ) cache = Service( project='composetest', name='cache', image='foo' ) console = Service( project='composetest', name='console', image='foo', links=[(web, 'web')] ) project = Project('test', [web, db, cache, console], None) assert project.get_services(['console'], include_deps=True) == [db, web, console] def test_get_services_removes_duplicates_following_links(self): db = Service( project='composetest', name='db', image='foo', ) web = Service( project='composetest', name='web', image='foo', links=[(db, 'database')] ) project = Project('test', [web, db], None) assert project.get_services(['web', 'db'], include_deps=True) == [db, web] def test_use_volumes_from_container(self): container_id = 'aabbccddee' container_dict = dict(Name='aaa', Id=container_id) self.mock_client.inspect_container.return_value = container_dict project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[{ 'name': 'test', 'image': 'busybox:latest', 'volumes_from': [VolumeFromSpec('aaa', 'rw', 'container')] }], networks=None, volumes=None, secrets=None, configs=None, ), ) assert project.get_service('test')._get_volumes_from() == [container_id + ":rw"] def test_use_volumes_from_service_no_container(self): container_name = 'test_vol_1' self.mock_client.containers.return_value = [ { "Name": container_name, "Names": [container_name], "Id": container_name, "Image": 'busybox:latest' } ] project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[ { 'name': 'vol', 'image': 'busybox:latest' }, { 'name': 'test', 'image': 'busybox:latest', 'volumes_from': [VolumeFromSpec('vol', 'rw', 'service')] } ], networks=None, volumes=None, secrets=None, configs=None, ), ) assert project.get_service('test')._get_volumes_from() == [container_name + ":rw"] @mock.patch('compose.network.Network.true_name', lambda n: n.full_name) def test_use_volumes_from_service_container(self): container_ids = ['aabbccddee', '12345'] project = Project.from_config( name='test', client=None, config_data=Config( version=V2_0, services=[ { 'name': 'vol', 'image': 'busybox:latest' }, { 'name': 'test', 'image': 'busybox:latest', 'volumes_from': [VolumeFromSpec('vol', 'rw', 'service')] } ], networks=None, volumes=None, secrets=None, configs=None, ), ) with mock.patch.object(Service, 'containers') as mock_return: mock_return.return_value = [ mock.Mock(id=container_id, spec=Container) for container_id in container_ids] assert ( project.get_service('test')._get_volumes_from() == [container_ids[0] + ':rw'] ) def test_events(self): services = [Service(name='web'), Service(name='db')] project = Project('test', services, self.mock_client) self.mock_client.events.return_value = iter([ { 'status': 'create', 'from': 'example/image', 'id': 'abcde', 'time': 1420092061, 'timeNano': 14200920610000002000, }, { 'status': 'attach', 'from': 'example/image', 'id': 'abcde', 'time': 1420092061, 'timeNano': 14200920610000003000, }, { 'status': 'create', 'from': 'example/other', 'id': 'bdbdbd', 'time': 1420092061, 'timeNano': 14200920610000005000, }, { 'status': 'create', 'from': 'example/db', 'id': 'ababa', 'time': 1420092061, 'timeNano': 14200920610000004000, }, { 'status': 'destroy', 'from': 'example/db', 'id': 'eeeee', 'time': 1420092061, 'timeNano': 14200920610000004000, }, ]) def dt_with_microseconds(dt, us): return datetime.datetime.fromtimestamp(dt).replace(microsecond=us) def get_container(cid): if cid == 'eeeee': raise NotFound(None, None, "oops") if cid == 'abcde': name = 'web' labels = {LABEL_SERVICE: name} elif cid == 'ababa': name = 'db' labels = {LABEL_SERVICE: name} else: labels = {} name = '' return { 'Id': cid, 'Config': {'Labels': labels}, 'Name': '/project_%s_1' % name, } self.mock_client.inspect_container.side_effect = get_container events = project.events() events_list = list(events) # Assert the return value is a generator assert not list(events) assert events_list == [ { 'type': 'container', 'service': 'web', 'action': 'create', 'id': 'abcde', 'attributes': { 'name': 'project_web_1', 'image': 'example/image', }, 'time': dt_with_microseconds(1420092061, 2), 'container': Container(None, {'Id': 'abcde'}), }, { 'type': 'container', 'service': 'web', 'action': 'attach', 'id': 'abcde', 'attributes': { 'name': 'project_web_1', 'image': 'example/image', }, 'time': dt_with_microseconds(1420092061, 3), 'container': Container(None, {'Id': 'abcde'}), }, { 'type': 'container', 'service': 'db', 'action': 'create', 'id': 'ababa', 'attributes': { 'name': 'project_db_1', 'image': 'example/db', }, 'time': dt_with_microseconds(1420092061, 4), 'container': Container(None, {'Id': 'ababa'}), }, ] def test_net_unset(self): project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V1, services=[ { 'name': 'test', 'image': 'busybox:latest', } ], networks=None, volumes=None, secrets=None, configs=None, ), ) service = project.get_service('test') assert service.network_mode.id is None assert 'NetworkMode' not in service._get_container_host_config({}) def test_use_net_from_container(self): container_id = 'aabbccddee' container_dict = dict(Name='aaa', Id=container_id) self.mock_client.inspect_container.return_value = container_dict project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[ { 'name': 'test', 'image': 'busybox:latest', 'network_mode': 'container:aaa' }, ], networks=None, volumes=None, secrets=None, configs=None, ), ) service = project.get_service('test') assert service.network_mode.mode == 'container:' + container_id def test_use_net_from_service(self): container_name = 'test_aaa_1' self.mock_client.containers.return_value = [ { "Name": container_name, "Names": [container_name], "Id": container_name, "Image": 'busybox:latest' } ] project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[ { 'name': 'aaa', 'image': 'busybox:latest' }, { 'name': 'test', 'image': 'busybox:latest', 'network_mode': 'service:aaa' }, ], networks=None, volumes=None, secrets=None, configs=None, ), ) service = project.get_service('test') assert service.network_mode.mode == 'container:' + container_name def test_uses_default_network_true(self): project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[ { 'name': 'foo', 'image': 'busybox:latest' }, ], networks=None, volumes=None, secrets=None, configs=None, ), ) assert 'default' in project.networks.networks def test_uses_default_network_false(self): project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[ { 'name': 'foo', 'image': 'busybox:latest', 'networks': {'custom': None} }, ], networks={'custom': {}}, volumes=None, secrets=None, configs=None, ), ) assert 'default' not in project.networks.networks def test_container_without_name(self): self.mock_client.containers.return_value = [ {'Image': 'busybox:latest', 'Id': '1', 'Name': '1'}, {'Image': 'busybox:latest', 'Id': '2', 'Name': None}, {'Image': 'busybox:latest', 'Id': '3'}, ] self.mock_client.inspect_container.return_value = { 'Id': '1', 'Config': { 'Labels': { LABEL_SERVICE: 'web', }, }, } project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[{ 'name': 'web', 'image': 'busybox:latest', }], networks=None, volumes=None, secrets=None, configs=None, ), ) assert [c.id for c in project.containers()] == ['1'] def test_down_with_no_resources(self): project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[{ 'name': 'web', 'image': 'busybox:latest', }], networks={'default': {}}, volumes={'data': {}}, secrets=None, configs=None, ), ) self.mock_client.remove_network.side_effect = NotFound(None, None, 'oops') self.mock_client.remove_volume.side_effect = NotFound(None, None, 'oops') project.down(ImageType.all, True) self.mock_client.remove_image.assert_called_once_with("busybox:latest") def test_no_warning_on_stop(self): self.mock_client.info.return_value = {'Swarm': {'LocalNodeState': 'active'}} project = Project('composetest', [], self.mock_client) with mock.patch('compose.project.log') as fake_log: project.stop() assert fake_log.warn.call_count == 0 def test_no_warning_in_normal_mode(self): self.mock_client.info.return_value = {'Swarm': {'LocalNodeState': 'inactive'}} project = Project('composetest', [], self.mock_client) with mock.patch('compose.project.log') as fake_log: project.up() assert fake_log.warn.call_count == 0 def test_no_warning_with_no_swarm_info(self): self.mock_client.info.return_value = {} project = Project('composetest', [], self.mock_client) with mock.patch('compose.project.log') as fake_log: project.up() assert fake_log.warn.call_count == 0 def test_no_such_service_unicode(self): assert NoSuchService('十六夜　咲夜'.encode('utf-8')).msg == 'No such service: 十六夜　咲夜' assert NoSuchService('十六夜　咲夜').msg == 'No such service: 十六夜　咲夜' def test_project_platform_value(self): service_config = { 'name': 'web', 'image': 'busybox:latest', } config_data = Config( version=V2_4, services=[service_config], networks={}, volumes={}, secrets=None, configs=None ) project = Project.from_config(name='test', client=self.mock_client, config_data=config_data) assert project.get_service('web').platform is None project = Project.from_config( name='test', client=self.mock_client, config_data=config_data, default_platform='windows' ) assert project.get_service('web').platform == 'windows' service_config['platform'] = 'linux/s390x' project = Project.from_config(name='test', client=self.mock_client, config_data=config_data) assert project.get_service('web').platform == 'linux/s390x' project = Project.from_config( name='test', client=self.mock_client, config_data=config_data, default_platform='windows' ) assert project.get_service('web').platform == 'linux/s390x' @mock.patch('compose.parallel.ParallelStreamWriter._write_noansi') def test_error_parallel_pull(self, mock_write): project = Project.from_config( name='test', client=self.mock_client, config_data=Config( version=V2_0, services=[{ 'name': 'web', 'image': 'busybox:latest', }], networks=None, volumes=None, secrets=None, configs=None, ), ) self.mock_client.pull.side_effect = OperationFailedError('pull error') with pytest.raises(ProjectError): project.pull(parallel_pull=True) self.mock_client.pull.side_effect = OperationFailedError(b'pull error') with pytest.raises(ProjectError): project.pull(parallel_pull=True)