docker-compose/tests/unit/service_test.py

from __future__ import absolute_import
from __future__ import unicode_literals

import docker
import pytest
from docker.constants import DEFAULT_DOCKER_API_VERSION
from docker.errors import APIError
from docker.errors import NotFound

from .. import mock
from .. import unittest
from compose.config.errors import DependencyError
from compose.config.types import MountSpec
from compose.config.types import ServicePort
from compose.config.types import ServiceSecret
from compose.config.types import VolumeFromSpec
from compose.config.types import VolumeSpec
from compose.const import API_VERSIONS
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.const import SECRETS_PATH
from compose.const import WINDOWS_LONGPATH_PREFIX
from compose.container import Container
from compose.errors import OperationFailedError
from compose.parallel import ParallelStreamWriter
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 format_environment
from compose.service import formatted_ports
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 rewrite_build_path
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.APIClient)
        self.mock_client.api_version = DEFAULT_DOCKER_API_VERSION
        self.mock_client._general_configs = {}

    def test_containers(self):
        service = Service('db', self.mock_client, 'myproject', image='foo')
        self.mock_client.containers.return_value = []
        assert 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')
        assert [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')

        assert [c.id for c in service.containers()] == ['1']
        assert service._next_container_number() == 2
        assert 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')])

        assert 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')])

        assert 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')

        assert 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')

            assert 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')])

        assert service._get_volumes_from() == [container_id + ':rw']
        from_service.create_container.assert_called_once_with()

    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)

        assert self.mock_client.create_host_config.called
        assert self.mock_client.create_host_config.call_args[1]['mem_limit'] == 1000000000
        assert self.mock_client.create_host_config.call_args[1]['memswap_limit'] == 2000000000

    def test_self_reference_external_link(self):
        service = Service(
            name='foo',
            external_links=['default_foo_1']
        )
        with pytest.raises(DependencyError):
            service.get_container_name('foo', 1)

    def test_mem_reservation(self):
        self.mock_client.create_host_config.return_value = {}

        service = Service(
            name='foo',
            image='foo',
            hostname='name',
            client=self.mock_client,
            mem_reservation='512m'
        )
        service._get_container_create_options({'some': 'overrides'}, 1)
        assert self.mock_client.create_host_config.called is True
        assert self.mock_client.create_host_config.call_args[1]['mem_reservation'] == '512m'

    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)

        assert self.mock_client.create_host_config.called
        assert 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)

        assert self.mock_client.create_host_config.called
        assert self.mock_client.create_host_config.call_args[1]['log_config'] == {
            'Type': 'syslog', 'Config': {'syslog-address': 'tcp://192.168.0.42:123'}
        }

    def test_stop_grace_period(self):
        self.mock_client.api_version = '1.25'
        self.mock_client.create_host_config.return_value = {}
        service = Service(
            'foo',
            image='foo',
            client=self.mock_client,
            stop_grace_period="1m35s")
        opts = service._get_container_create_options({'image': 'foo'}, 1)
        assert opts['stop_timeout'] == 95

    def test_split_domainname_none(self):
        service = Service(
            'foo',
            image='foo',
            hostname='name.domain.tld',
            client=self.mock_client)
        opts = service._get_container_create_options({'image': 'foo'}, 1)
        assert opts['hostname'] == 'name.domain.tld', 'hostname'
        assert not ('domainname' in opts), 'domainname'

    def test_split_domainname_fqdn(self):
        self.mock_client.api_version = '1.22'
        service = Service(
            'foo',
            hostname='name.domain.tld',
            image='foo',
            client=self.mock_client)
        opts = service._get_container_create_options({'image': 'foo'}, 1)
        assert opts['hostname'] == 'name', 'hostname'
        assert opts['domainname'] == 'domain.tld', 'domainname'

    def test_split_domainname_both(self):
        self.mock_client.api_version = '1.22'
        service = Service(
            'foo',
            hostname='name',
            image='foo',
            domainname='domain.tld',
            client=self.mock_client)
        opts = service._get_container_create_options({'image': 'foo'}, 1)
        assert opts['hostname'] == 'name', 'hostname'
        assert opts['domainname'] == 'domain.tld', 'domainname'

    def test_split_domainname_weird(self):
        self.mock_client.api_version = '1.22'
        service = Service(
            'foo',
            hostname='name.sub',
            domainname='domain.tld',
            image='foo',
            client=self.mock_client)
        opts = service._get_container_create_options({'image': 'foo'}, 1)
        assert opts['hostname'] == 'name.sub', 'hostname'
        assert 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)
        assert opts.get('hostname') is None

    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)
        assert 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.full_slug = 'abcdefff1234'
        prev_container.get.return_value = None

        opts = service._get_container_create_options(
            {}, 1, previous_container=prev_container
        )

        assert service.options['labels'] == labels
        assert service.options['environment'] == environment

        assert 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
        prev_container.full_slug = 'abcdefff1234'

        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
        prev_container.full_slug = 'abcdefff1234'

        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')

        with pytest.raises(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_bdfa3ed91e2c')
        self.mock_client.containers.return_value = [container_dict]
        service = Service('foo', image='foo', client=self.mock_client)

        container = service.get_container(number=2)
        assert 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,
            platform=None)
        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,
            platform=None)

    @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,
            platform=None)
        mock_log.info.assert_called_once_with('Pulling foo (someimage@sha256:1234)...')

    @mock.patch('compose.service.log', autospec=True)
    def test_pull_image_with_platform(self, mock_log):
        self.mock_client.api_version = '1.35'
        service = Service(
            'foo', client=self.mock_client, image='someimage:sometag', platform='windows/x86_64'
        )
        service.pull()
        assert self.mock_client.pull.call_count == 1
        call_args = self.mock_client.pull.call_args
        assert call_args[1]['platform'] == 'windows/x86_64'

    @mock.patch('compose.service.log', autospec=True)
    def test_pull_image_with_platform_unsupported_api(self, mock_log):
        self.mock_client.api_version = '1.33'
        service = Service(
            'foo', client=self.mock_client, image='someimage:sometag', platform='linux/arm'
        )
        with pytest.raises(OperationFailedError):
            service.pull()

    def test_pull_image_with_default_platform(self):
        self.mock_client.api_version = '1.35'

        service = Service(
            'foo', client=self.mock_client, image='someimage:sometag',
            default_platform='linux'
        )
        assert service.platform == 'linux'
        service.pull()

        assert self.mock_client.pull.call_count == 1
        call_args = self.mock_client.pull.call_args
        assert call_args[1]['platform'] == 'linux'

    @mock.patch('compose.service.Container', autospec=True)
    def test_recreate_container(self, _):
        mock_container = mock.create_autospec(Container)
        mock_container.full_slug = 'abcdefff1234'
        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)
        mock_container.full_slug = 'abcdefff1234'
        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):
        assert parse_repository_tag("root") == ("root", "", ":")
        assert parse_repository_tag("root:tag") == ("root", "tag", ":")
        assert parse_repository_tag("user/repo") == ("user/repo", "", ":")
        assert parse_repository_tag("user/repo:tag") == ("user/repo", "tag", ":")
        assert parse_repository_tag("url:5000/repo") == ("url:5000/repo", "", ":")
        assert parse_repository_tag("url:5000/repo:tag") == ("url:5000/repo", "tag", ":")
        assert parse_repository_tag("root@sha256:digest") == ("root", "sha256:digest", "@")
        assert parse_repository_tag("user/repo@sha256:digest") == ("user/repo", "sha256:digest", "@")
        assert 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]

        assert self.mock_client.build.call_count == 1
        self.mock_client.build.call_args[1]['tag'] == 'default_foo'

    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
        assert self.mock_client.build.call_count == 1
        self.mock_client.build.call_args[1]['tag'] == 'default_foo'

    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()

        assert self.mock_client.build.call_count == 1
        assert not self.mock_client.build.call_args[1]['pull']

    def test_build_with_platform(self):
        self.mock_client.api_version = '1.35'
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        service = Service('foo', client=self.mock_client, build={'context': '.'}, platform='linux')
        service.build()

        assert self.mock_client.build.call_count == 1
        call_args = self.mock_client.build.call_args
        assert call_args[1]['platform'] == 'linux'

    def test_build_with_default_platform(self):
        self.mock_client.api_version = '1.35'
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        service = Service(
            'foo', client=self.mock_client, build={'context': '.'},
            default_platform='linux'
        )
        assert service.platform == 'linux'
        service.build()

        assert self.mock_client.build.call_count == 1
        call_args = self.mock_client.build.call_args
        assert call_args[1]['platform'] == 'linux'

    def test_service_platform_precedence(self):
        self.mock_client.api_version = '1.35'

        service = Service(
            'foo', client=self.mock_client, platform='linux/arm',
            default_platform='osx'
        )
        assert service.platform == 'linux/arm'

    def test_service_ignore_default_platform_with_unsupported_api(self):
        self.mock_client.api_version = '1.32'
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        service = Service(
            'foo', client=self.mock_client, default_platform='windows', build={'context': '.'}
        )
        assert service.platform is None
        service.build()
        assert self.mock_client.build.call_count == 1
        call_args = self.mock_client.build.call_args
        assert call_args[1]['platform'] is None

    def test_build_with_override_build_args(self):
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        build_args = {
            'arg1': 'arg1_new_value',
        }
        service = Service('foo', client=self.mock_client,
                          build={'context': '.', 'args': {'arg1': 'arg1', 'arg2': 'arg2'}})
        service.build(build_args_override=build_args)

        called_build_args = self.mock_client.build.call_args[1]['buildargs']

        assert called_build_args['arg1'] == build_args['arg1']
        assert called_build_args['arg2'] == 'arg2'

    def test_build_with_isolation_from_service_config(self):
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        service = Service('foo', client=self.mock_client, build={'context': '.'}, isolation='hyperv')
        service.build()

        assert self.mock_client.build.call_count == 1
        called_build_args = self.mock_client.build.call_args[1]
        assert called_build_args['isolation'] == 'hyperv'

    def test_build_isolation_from_build_override_service_config(self):
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        service = Service(
            'foo', client=self.mock_client, build={'context': '.', 'isolation': 'default'},
            isolation='hyperv'
        )
        service.build()

        assert self.mock_client.build.call_count == 1
        called_build_args = self.mock_client.build.call_args[1]
        assert called_build_args['isolation'] == 'default'

    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_config_hash_matches_label(self):
        self.mock_client.inspect_image.return_value = {'Id': 'abcd'}
        service = Service(
            'foo',
            image='example.com/foo',
            client=self.mock_client,
            network_mode=NetworkMode('bridge'),
            networks={'bridge': {}, 'net2': {}},
            links=[(Service('one', client=self.mock_client), 'one')],
            volumes_from=[VolumeFromSpec(Service('two', client=self.mock_client), 'rw', 'service')],
            volumes=[VolumeSpec('/ext', '/int', 'ro')],
            build={'context': 'some/random/path'},
        )
        config_hash = service.config_hash

        for api_version in set(API_VERSIONS.values()):
            self.mock_client.api_version = api_version
            assert service._get_container_create_options(
                {}, 1
            )['labels'][LABEL_CONFIG_HASH] == config_hash

    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')
        assert not service.specifies_host_port()

    def test_specifies_host_port_with_container_port(self):
        service = Service(
            'foo',
            image='foo',
            ports=["2000"])
        assert not service.specifies_host_port()

    def test_specifies_host_port_with_host_port(self):
        service = Service(
            'foo',
            image='foo',
            ports=["1000:2000"])
        assert service.specifies_host_port()

    def test_specifies_host_port_with_host_ip_no_port(self):
        service = Service(
            'foo',
            image='foo',
            ports=["127.0.0.1::2000"])
        assert not service.specifies_host_port()

    def test_specifies_host_port_with_host_ip_and_port(self):
        service = Service(
            'foo',
            image='foo',
            ports=["127.0.0.1:1000:2000"])
        assert service.specifies_host_port()

    def test_specifies_host_port_with_container_port_range(self):
        service = Service(
            'foo',
            image='foo',
            ports=["2000-3000"])
        assert not service.specifies_host_port()

    def test_specifies_host_port_with_host_port_range(self):
        service = Service(
            'foo',
            image='foo',
            ports=["1000-2000:2000-3000"])
        assert service.specifies_host_port()

    def test_specifies_host_port_with_host_ip_no_port_range(self):
        service = Service(
            'foo',
            image='foo',
            ports=["127.0.0.1::2000-3000"])
        assert not service.specifies_host_port()

    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"])
        assert service.specifies_host_port()

    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'}
        ParallelStreamWriter.instance = None
        name = 'foo'
        service = Service(
            name,
            client=self.mock_client,
            ports=["8080:80"])

        service.scale(0)
        assert not mock_log.warn.called

        service.scale(1)
        assert not 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))

    def test_parse_proxy_config(self):
        default_proxy_config = {
            'httpProxy': 'http://proxy.mycorp.com:3128',
            'httpsProxy': 'https://user:[email protected]:3129',
            'ftpProxy': 'http://ftpproxy.mycorp.com:21',
            'noProxy': '*.intra.mycorp.com',
        }

        self.mock_client.base_url = 'http+docker://localunixsocket'
        self.mock_client._general_configs = {
            'proxies': {
                'default': default_proxy_config,
            }
        }

        service = Service('foo', client=self.mock_client)

        assert service._parse_proxy_config() == {
            'HTTP_PROXY': default_proxy_config['httpProxy'],
            'http_proxy': default_proxy_config['httpProxy'],
            'HTTPS_PROXY': default_proxy_config['httpsProxy'],
            'https_proxy': default_proxy_config['httpsProxy'],
            'FTP_PROXY': default_proxy_config['ftpProxy'],
            'ftp_proxy': default_proxy_config['ftpProxy'],
            'NO_PROXY': default_proxy_config['noProxy'],
            'no_proxy': default_proxy_config['noProxy'],
        }

    def test_parse_proxy_config_per_host(self):
        default_proxy_config = {
            'httpProxy': 'http://proxy.mycorp.com:3128',
            'httpsProxy': 'https://user:[email protected]:3129',
            'ftpProxy': 'http://ftpproxy.mycorp.com:21',
            'noProxy': '*.intra.mycorp.com',
        }
        host_specific_proxy_config = {
            'httpProxy': 'http://proxy.example.com:3128',
            'httpsProxy': 'https://user:[email protected]:3129',
            'ftpProxy': 'http://ftpproxy.example.com:21',
            'noProxy': '*.intra.example.com'
        }

        self.mock_client.base_url = 'http+docker://localunixsocket'
        self.mock_client._general_configs = {
            'proxies': {
                'default': default_proxy_config,
                'tcp://example.docker.com:2376': host_specific_proxy_config,
            }
        }

        service = Service('foo', client=self.mock_client)

        assert service._parse_proxy_config() == {
            'HTTP_PROXY': default_proxy_config['httpProxy'],
            'http_proxy': default_proxy_config['httpProxy'],
            'HTTPS_PROXY': default_proxy_config['httpsProxy'],
            'https_proxy': default_proxy_config['httpsProxy'],
            'FTP_PROXY': default_proxy_config['ftpProxy'],
            'ftp_proxy': default_proxy_config['ftpProxy'],
            'NO_PROXY': default_proxy_config['noProxy'],
            'no_proxy': default_proxy_config['noProxy'],
        }

        self.mock_client._original_base_url = 'tcp://example.docker.com:2376'

        assert service._parse_proxy_config() == {
            'HTTP_PROXY': host_specific_proxy_config['httpProxy'],
            'http_proxy': host_specific_proxy_config['httpProxy'],
            'HTTPS_PROXY': host_specific_proxy_config['httpsProxy'],
            'https_proxy': host_specific_proxy_config['httpsProxy'],
            'FTP_PROXY': host_specific_proxy_config['ftpProxy'],
            'ftp_proxy': host_specific_proxy_config['ftpProxy'],
            'NO_PROXY': host_specific_proxy_config['noProxy'],
            'no_proxy': host_specific_proxy_config['noProxy'],
        }

    def test_build_service_with_proxy_config(self):
        default_proxy_config = {
            'httpProxy': 'http://proxy.mycorp.com:3128',
            'httpsProxy': 'https://user:[email protected]:3129',
        }
        buildargs = {
            'HTTPS_PROXY': 'https://rdcf.th08.jp:8911',
            'https_proxy': 'https://rdcf.th08.jp:8911',
        }
        self.mock_client._general_configs = {
            'proxies': {
                'default': default_proxy_config,
            }
        }
        self.mock_client.base_url = 'http+docker://localunixsocket'
        self.mock_client.build.return_value = [
            b'{"stream": "Successfully built 12345"}',
        ]

        service = Service('foo', client=self.mock_client, build={'context': '.', 'args': buildargs})
        service.build()

        assert self.mock_client.build.call_count == 1
        assert self.mock_client.build.call_args[1]['buildargs'] == {
            'HTTP_PROXY': default_proxy_config['httpProxy'],
            'http_proxy': default_proxy_config['httpProxy'],
            'HTTPS_PROXY': buildargs['HTTPS_PROXY'],
            'https_proxy': buildargs['HTTPS_PROXY'],
        }

    def test_get_create_options_with_proxy_config(self):
        default_proxy_config = {
            'httpProxy': 'http://proxy.mycorp.com:3128',
            'httpsProxy': 'https://user:[email protected]:3129',
            'ftpProxy': 'http://ftpproxy.mycorp.com:21',
        }
        self.mock_client._general_configs = {
            'proxies': {
                'default': default_proxy_config,
            }
        }
        self.mock_client.base_url = 'http+docker://localunixsocket'

        override_options = {
            'environment': {
                'FTP_PROXY': 'ftp://xdge.exo.au:21',
                'ftp_proxy': 'ftp://xdge.exo.au:21',
            }
        }
        environment = {
            'HTTPS_PROXY': 'https://rdcf.th08.jp:8911',
            'https_proxy': 'https://rdcf.th08.jp:8911',
        }

        service = Service('foo', client=self.mock_client, environment=environment)

        create_opts = service._get_container_create_options(override_options, 1)
        assert set(create_opts['environment']) == set(format_environment({
            'HTTP_PROXY': default_proxy_config['httpProxy'],
            'http_proxy': default_proxy_config['httpProxy'],
            'HTTPS_PROXY': environment['HTTPS_PROXY'],
            'https_proxy': environment['HTTPS_PROXY'],
            'FTP_PROXY': override_options['environment']['FTP_PROXY'],
            'ftp_proxy': override_options['environment']['FTP_PROXY'],
        }))

    def test_create_when_removed_containers_are_listed(self):
        # This is aimed at simulating a race between the API call to list the
        # containers, and the ones to inspect each of the listed containers.
        # It can happen that a container has been removed after we listed it.

        # containers() returns a container that is about to be removed
        self.mock_client.containers.return_value = [
            {'Id': 'rm_cont_id', 'Name': 'rm_cont', 'Image': 'img_id'},
        ]

        # inspect_container() will raise a NotFound when trying to inspect
        # rm_cont_id, which at this point has been removed
        def inspect(name):
            if name == 'rm_cont_id':
                raise NotFound(message='Not Found')

            if name == 'new_cont_id':
                return {'Id': 'new_cont_id'}

            raise NotImplementedError("incomplete mock")

        self.mock_client.inspect_container.side_effect = inspect

        self.mock_client.inspect_image.return_value = {'Id': 'imageid'}

        self.mock_client.create_container.return_value = {'Id': 'new_cont_id'}

        # We should nonetheless be able to create a new container
        service = Service('foo', client=self.mock_client)

        assert service.create_container().id == 'new_cont_id'

    def test_build_volume_options_duplicate_binds(self):
        self.mock_client.api_version = '1.29'  # Trigger 3.2 format workaround
        service = Service('foo', client=self.mock_client)
        ctnr_opts, override_opts = service._build_container_volume_options(
            previous_container=None,
            container_options={
                'volumes': [
                    MountSpec.parse({'source': 'vol', 'target': '/data', 'type': 'volume'}),
                    VolumeSpec.parse('vol:/data:rw'),
                ],
                'environment': {},
            },
            override_options={},
        )
        assert 'binds' in override_opts
        assert len(override_opts['binds']) == 1
        assert override_opts['binds'][0] == 'vol:/data:rw'

    def test_volumes_order_is_preserved(self):
        service = Service('foo', client=self.mock_client)
        volumes = [
            VolumeSpec.parse(cfg) for cfg in [
                '/v{0}:/v{0}:rw'.format(i) for i in range(6)
            ]
        ]
        ctnr_opts, override_opts = service._build_container_volume_options(
            previous_container=None,
            container_options={
                'volumes': volumes,
                'environment': {},
            },
            override_options={},
        )
        assert override_opts['binds'] == [vol.repr() for vol in volumes]


class TestServiceNetwork(unittest.TestCase):
    def setUp(self):
        self.mock_client = mock.create_autospec(docker.APIClient)
        self.mock_client.api_version = DEFAULT_DOCKER_API_VERSION
        self.mock_client._general_configs = {}

    def test_connect_container_to_networks_short_aliase_exists(self):
        service = Service(
            'db',
            self.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 self.mock_client.disconnect_container_from_network.call_count
        assert not self.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 setUp(self):
        self.mock_client = mock.create_autospec(docker.APIClient)
        self.mock_client.api_version = DEFAULT_DOCKER_API_VERSION
        self.mock_client._general_configs = {}

    def test_network_mode(self):
        network_mode = NetworkMode('host')
        assert network_mode.id == 'host'
        assert network_mode.mode == 'host'
        assert network_mode.service_name is None

    def test_network_mode_container(self):
        container_id = 'abcd'
        network_mode = ContainerNetworkMode(Container(None, {'Id': container_id}))
        assert network_mode.id == container_id
        assert network_mode.mode == 'container:' + container_id
        assert network_mode.service_name is None

    def test_network_mode_service(self):
        container_id = 'bbbb'
        service_name = 'web'
        self.mock_client.containers.return_value = [
            {'Id': container_id, 'Name': container_id, 'Image': 'abcd'},
        ]

        service = Service(name=service_name, client=self.mock_client)
        network_mode = ServiceNetworkMode(service)

        assert network_mode.id == service_name
        assert network_mode.mode == 'container:' + container_id
        assert network_mode.service_name == service_name

    def test_network_mode_service_no_containers(self):
        service_name = 'web'
        self.mock_client.containers.return_value = []

        service = Service(name=service_name, client=self.mock_client)
        network_mode = ServiceNetworkMode(service)

        assert network_mode.id == service_name
        assert network_mode.mode is None
        assert network_mode.service_name == service_name


class ServicePortsTest(unittest.TestCase):
    def test_formatted_ports(self):
        ports = [
            '3000',
            '0.0.0.0:4025-4030:23000-23005',
            ServicePort(6000, None, None, None, None),
            ServicePort(8080, 8080, None, None, None),
            ServicePort('20000', '20000', 'udp', 'ingress', None),
            ServicePort(30000, '30000', 'tcp', None, '127.0.0.1'),
        ]
        formatted = formatted_ports(ports)
        assert ports[0] in formatted
        assert ports[1] in formatted
        assert '6000/tcp' in formatted
        assert '8080:8080/tcp' in formatted
        assert '20000:20000/udp' in formatted
        assert '127.0.0.1:30000:30000/tcp' in formatted


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.APIClient)
        self.mock_client.api_version = DEFAULT_DOCKER_API_VERSION
        self.mock_client._general_configs = {}

    def test_build_volume_binding(self):
        binding = build_volume_binding(VolumeSpec.parse('/outside:/inside', True))
        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',
            '/dev/tmpfs'
        ]]

        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, ['/dev/tmpfs'], [])
        assert sorted(volumes) == sorted(expected)

    def test_merge_volume_bindings(self):
        options = [
            VolumeSpec.parse(v, True) for v in [
                '/host/volume:/host/volume:ro',
                '/host/rw/volume:/host/rw/volume',
                '/new/volume',
                '/existing/volume',
                '/dev/tmpfs'
            ]
        ]

        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, ['/dev/tmpfs'], 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', True),
                VolumeSpec.parse('/host/path:/data2', True),
            ],
            client=self.mock_client,
        )

        self.mock_client.inspect_image.return_value = {
            'Id': 'ababab',
            'ContainerConfig': {
                'Volumes': {}
            }
        }

        service._get_container_create_options(
            override_options={},
            number=1,
        )

        assert 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, True)],
        ).create_container()

        assert self.mock_client.create_container.call_count == 1
        assert self.mock_client.create_host_config.call_args[1]['binds'] == [volume]


class ServiceSecretTest(unittest.TestCase):
    def setUp(self):
        self.mock_client = mock.create_autospec(docker.APIClient)
        self.mock_client.api_version = DEFAULT_DOCKER_API_VERSION
        self.mock_client._general_configs = {}

    def test_get_secret_volumes(self):
        secret1 = {
            'secret': ServiceSecret.parse({'source': 'secret1', 'target': 'b.txt'}),
            'file': 'a.txt'
        }
        service = Service(
            'web',
            client=self.mock_client,
            image='busybox',
            secrets=[secret1]
        )
        volumes = service.get_secret_volumes()

        assert volumes[0].source == secret1['file']
        assert volumes[0].target == '{}/{}'.format(SECRETS_PATH, secret1['secret'].target)

    def test_get_secret_volumes_abspath(self):
        secret1 = {
            'secret': ServiceSecret.parse({'source': 'secret1', 'target': '/d.txt'}),
            'file': 'c.txt'
        }
        service = Service(
            'web',
            client=self.mock_client,
            image='busybox',
            secrets=[secret1]
        )
        volumes = service.get_secret_volumes()

        assert volumes[0].source == secret1['file']
        assert volumes[0].target == secret1['secret'].target

    def test_get_secret_volumes_no_target(self):
        secret1 = {
            'secret': ServiceSecret.parse({'source': 'secret1'}),
            'file': 'c.txt'
        }
        service = Service(
            'web',
            client=self.mock_client,
            image='busybox',
            secrets=[secret1]
        )
        volumes = service.get_secret_volumes()

        assert volumes[0].source == secret1['file']
        assert volumes[0].target == '{}/{}'.format(SECRETS_PATH, secret1['secret'].source)


class RewriteBuildPathTest(unittest.TestCase):
    @mock.patch('compose.service.IS_WINDOWS_PLATFORM', True)
    def test_rewrite_url_no_prefix(self):
        urls = [
            'http://test.com',
            'https://test.com',
            'git://test.com',
            'github.com/test/test',
            '[email protected]',
        ]
        for u in urls:
            assert rewrite_build_path(u) == u

    @mock.patch('compose.service.IS_WINDOWS_PLATFORM', True)
    def test_rewrite_windows_path(self):
        assert rewrite_build_path('C:\\context') == WINDOWS_LONGPATH_PREFIX + 'C:\\context'
        assert rewrite_build_path(
            rewrite_build_path('C:\\context')
        ) == rewrite_build_path('C:\\context')

    @mock.patch('compose.service.IS_WINDOWS_PLATFORM', False)
    def test_rewrite_unix_path(self):
        assert rewrite_build_path('/context') == '/context'