path: root/tests/integration/api_service_test.py

# -*- coding: utf-8 -*-

import random
import time

import docker
import pytest
import six

from ..helpers import (
    force_leave_swarm, requires_api_version, requires_experimental
)
from .base import BaseAPIIntegrationTest, TEST_IMG


class ServiceTest(BaseAPIIntegrationTest):
    @classmethod
    def setup_class(cls):
        client = cls.get_client_instance()
        force_leave_swarm(client)
        cls._init_swarm(client)

    @classmethod
    def teardown_class(cls):
        client = cls.get_client_instance()
        force_leave_swarm(client)

    def tearDown(self):
        for service in self.client.services(filters={'name': 'dockerpytest_'}):
            try:
                self.client.remove_service(service['ID'])
            except docker.errors.APIError:
                pass
        super(ServiceTest, self).tearDown()

    def get_service_name(self):
        return 'dockerpytest_{0:x}'.format(random.getrandbits(64))

    def get_service_container(self, service_name, attempts=20, interval=0.5,
                              include_stopped=False):
        # There is some delay between the service's creation and the creation
        # of the service's containers. This method deals with the uncertainty
        # when trying to retrieve the container associated with a service.
        while True:
            containers = self.client.containers(
                filters={'name': [service_name]}, quiet=True,
                all=include_stopped
            )
            if len(containers) > 0:
                return containers[0]
            attempts -= 1
            if attempts <= 0:
                return None
            time.sleep(interval)

    def create_simple_service(self, name=None, labels=None):
        if name:
            name = 'dockerpytest_{0}'.format(name)
        else:
            name = self.get_service_name()

        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        return name, self.client.create_service(
            task_tmpl, name=name, labels=labels
        )

    @requires_api_version('1.24')
    def test_list_services(self):
        services = self.client.services()
        assert isinstance(services, list)

        test_services = self.client.services(filters={'name': 'dockerpytest_'})
        assert len(test_services) == 0
        self.create_simple_service()
        test_services = self.client.services(filters={'name': 'dockerpytest_'})
        assert len(test_services) == 1
        assert 'dockerpytest_' in test_services[0]['Spec']['Name']

    @requires_api_version('1.24')
    def test_list_services_filter_by_label(self):
        test_services = self.client.services(filters={'label': 'test_label'})
        assert len(test_services) == 0
        self.create_simple_service(labels={'test_label': 'testing'})
        test_services = self.client.services(filters={'label': 'test_label'})
        assert len(test_services) == 1
        assert test_services[0]['Spec']['Labels']['test_label'] == 'testing'

    def test_inspect_service_by_id(self):
        svc_name, svc_id = self.create_simple_service()
        svc_info = self.client.inspect_service(svc_id)
        assert 'ID' in svc_info
        assert svc_info['ID'] == svc_id['ID']

    def test_inspect_service_by_name(self):
        svc_name, svc_id = self.create_simple_service()
        svc_info = self.client.inspect_service(svc_name)
        assert 'ID' in svc_info
        assert svc_info['ID'] == svc_id['ID']

    @requires_api_version('1.29')
    def test_inspect_service_insert_defaults(self):
        svc_name, svc_id = self.create_simple_service()
        svc_info = self.client.inspect_service(svc_id)
        svc_info_defaults = self.client.inspect_service(
            svc_id, insert_defaults=True
        )
        assert svc_info != svc_info_defaults
        assert 'RollbackConfig' in svc_info_defaults['Spec']
        assert 'RollbackConfig' not in svc_info['Spec']

    def test_remove_service_by_id(self):
        svc_name, svc_id = self.create_simple_service()
        assert self.client.remove_service(svc_id)
        test_services = self.client.services(filters={'name': 'dockerpytest_'})
        assert len(test_services) == 0

    def test_remove_service_by_name(self):
        svc_name, svc_id = self.create_simple_service()
        assert self.client.remove_service(svc_name)
        test_services = self.client.services(filters={'name': 'dockerpytest_'})
        assert len(test_services) == 0

    def test_create_service_simple(self):
        name, svc_id = self.create_simple_service()
        assert self.client.inspect_service(svc_id)
        services = self.client.services(filters={'name': name})
        assert len(services) == 1
        assert services[0]['ID'] == svc_id['ID']

    @requires_api_version('1.25')
    @requires_experimental(until='1.29')
    def test_service_logs(self):
        name, svc_id = self.create_simple_service()
        assert self.get_service_container(name, include_stopped=True)
        attempts = 20
        while True:
            if attempts == 0:
                self.fail('No service logs produced by endpoint')
                return
            logs = self.client.service_logs(svc_id, stdout=True, is_tty=False)
            try:
                log_line = next(logs)
            except StopIteration:
                attempts -= 1
                time.sleep(0.1)
                continue
            else:
                break

        if six.PY3:
            log_line = log_line.decode('utf-8')
        assert 'hello\n' in log_line

    def test_create_service_custom_log_driver(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['echo', 'hello']
        )
        log_cfg = docker.types.DriverConfig('none')
        task_tmpl = docker.types.TaskTemplate(
            container_spec, log_driver=log_cfg
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        res_template = svc_info['Spec']['TaskTemplate']
        assert 'LogDriver' in res_template
        assert 'Name' in res_template['LogDriver']
        assert res_template['LogDriver']['Name'] == 'none'

    def test_create_service_with_volume_mount(self):
        vol_name = self.get_service_name()
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['ls'],
            mounts=[
                docker.types.Mount(target='/test', source=vol_name)
            ]
        )
        self.tmp_volumes.append(vol_name)
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        cspec = svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert 'Mounts' in cspec
        assert len(cspec['Mounts']) == 1
        mount = cspec['Mounts'][0]
        assert mount['Target'] == '/test'
        assert mount['Source'] == vol_name
        assert mount['Type'] == 'volume'

    def test_create_service_with_resources_constraints(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        resources = docker.types.Resources(
            cpu_limit=4000000, mem_limit=3 * 1024 * 1024 * 1024,
            cpu_reservation=3500000, mem_reservation=2 * 1024 * 1024 * 1024
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec, resources=resources
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        res_template = svc_info['Spec']['TaskTemplate']
        assert 'Resources' in res_template
        assert res_template['Resources']['Limits'] == resources['Limits']
        assert res_template['Resources']['Reservations'] == resources[
            'Reservations'
        ]

    def _create_service_with_generic_resources(self, generic_resources):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])

        resources = docker.types.Resources(
            generic_resources=generic_resources
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec, resources=resources
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        return resources, self.client.inspect_service(svc_id)

    @requires_api_version('1.32')
    def test_create_service_with_generic_resources(self):
        successful = [{
            'input': [
                {'DiscreteResourceSpec': {'Kind': 'gpu', 'Value': 1}},
                {'NamedResourceSpec': {'Kind': 'gpu', 'Value': 'test'}}
            ]}, {
            'input': {'gpu': 2, 'mpi': 'latest'},
            'expected': [
                {'DiscreteResourceSpec': {'Kind': 'gpu', 'Value': 2}},
                {'NamedResourceSpec': {'Kind': 'mpi', 'Value': 'latest'}}
            ]}
        ]

        for test in successful:
            t = test['input']
            resrcs, svc_info = self._create_service_with_generic_resources(t)

            assert 'TaskTemplate' in svc_info['Spec']
            res_template = svc_info['Spec']['TaskTemplate']
            assert 'Resources' in res_template
            res_reservations = res_template['Resources']['Reservations']
            assert res_reservations == resrcs['Reservations']
            assert 'GenericResources' in res_reservations

            def _key(d, specs=('DiscreteResourceSpec', 'NamedResourceSpec')):
                return [d.get(s, {}).get('Kind', '') for s in specs]

            actual = res_reservations['GenericResources']
            expected = test.get('expected', test['input'])
            assert sorted(actual, key=_key) == sorted(expected, key=_key)

    @requires_api_version('1.32')
    def test_create_service_with_invalid_generic_resources(self):
        for test_input in ['1', 1.0, lambda: '1', {1, 2}]:
            with pytest.raises(docker.errors.InvalidArgument):
                self._create_service_with_generic_resources(test_input)

    def test_create_service_with_update_config(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        update_config = docker.types.UpdateConfig(
            parallelism=10, delay=5, failure_action='pause'
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, update_config=update_config, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'UpdateConfig' in svc_info['Spec']
        uc = svc_info['Spec']['UpdateConfig']
        assert update_config['Parallelism'] == uc['Parallelism']
        assert update_config['Delay'] == uc['Delay']
        assert update_config['FailureAction'] == uc['FailureAction']

    @requires_api_version('1.28')
    def test_create_service_with_failure_action_rollback(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        update_config = docker.types.UpdateConfig(failure_action='rollback')
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, update_config=update_config, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'UpdateConfig' in svc_info['Spec']
        uc = svc_info['Spec']['UpdateConfig']
        assert update_config['FailureAction'] == uc['FailureAction']

    @requires_api_version('1.25')
    def test_create_service_with_update_config_monitor(self):
        container_spec = docker.types.ContainerSpec('busybox', ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        update_config = docker.types.UpdateConfig(
            monitor=300000000, max_failure_ratio=0.4
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, update_config=update_config, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'UpdateConfig' in svc_info['Spec']
        uc = svc_info['Spec']['UpdateConfig']
        assert update_config['Monitor'] == uc['Monitor']
        assert update_config['MaxFailureRatio'] == uc['MaxFailureRatio']

    @requires_api_version('1.28')
    def test_create_service_with_rollback_config(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        rollback_cfg = docker.types.RollbackConfig(
            parallelism=10, delay=5, failure_action='pause',
            monitor=300000000, max_failure_ratio=0.4
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, rollback_config=rollback_cfg, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'RollbackConfig' in svc_info['Spec']
        rc = svc_info['Spec']['RollbackConfig']
        assert rollback_cfg['Parallelism'] == rc['Parallelism']
        assert rollback_cfg['Delay'] == rc['Delay']
        assert rollback_cfg['FailureAction'] == rc['FailureAction']
        assert rollback_cfg['Monitor'] == rc['Monitor']
        assert rollback_cfg['MaxFailureRatio'] == rc['MaxFailureRatio']

    def test_create_service_with_restart_policy(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        policy = docker.types.RestartPolicy(
            docker.types.RestartPolicy.condition_types.ANY,
            delay=5, max_attempts=5
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec, restart_policy=policy
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'RestartPolicy' in svc_info['Spec']['TaskTemplate']
        assert policy == svc_info['Spec']['TaskTemplate']['RestartPolicy']

    def test_create_service_with_custom_networks(self):
        net1 = self.client.create_network(
            'dockerpytest_1', driver='overlay', ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(net1['Id'])
        net2 = self.client.create_network(
            'dockerpytest_2', driver='overlay', ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(net2['Id'])
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, networks=[
                'dockerpytest_1', {'Target': 'dockerpytest_2'}
            ]
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Networks' in svc_info['Spec']
        assert svc_info['Spec']['Networks'] == [
            {'Target': net1['Id']}, {'Target': net2['Id']}
        ]

    def test_create_service_with_network_attachment_config(self):
        network = self.client.create_network(
            'dockerpytest_1', driver='overlay', ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(network['Id'])
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        network_config = docker.types.NetworkAttachmentConfig(
            target='dockerpytest_1',
            aliases=['dockerpytest_1_alias'],
            options={
                'foo': 'bar'
            }
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec,
            networks=[network_config]
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Networks' in svc_info['Spec']['TaskTemplate']
        service_networks_info = svc_info['Spec']['TaskTemplate']['Networks']
        assert len(service_networks_info) == 1
        assert service_networks_info[0]['Target'] == network['Id']
        assert service_networks_info[0]['Aliases'] == ['dockerpytest_1_alias']
        assert service_networks_info[0]['DriverOpts'] == {'foo': 'bar'}

    def test_create_service_with_placement(self):
        node_id = self.client.nodes()[0]['ID']
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(
            container_spec, placement=['node.id=={}'.format(node_id)]
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Placement' in svc_info['Spec']['TaskTemplate']
        assert (svc_info['Spec']['TaskTemplate']['Placement'] ==
                {'Constraints': ['node.id=={}'.format(node_id)]})

    def test_create_service_with_placement_object(self):
        node_id = self.client.nodes()[0]['ID']
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        placemt = docker.types.Placement(
            constraints=['node.id=={}'.format(node_id)]
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec, placement=placemt
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Placement' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['Placement'] == placemt

    @requires_api_version('1.30')
    def test_create_service_with_placement_platform(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        placemt = docker.types.Placement(platforms=[('x86_64', 'linux')])
        task_tmpl = docker.types.TaskTemplate(
            container_spec, placement=placemt
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Placement' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['Placement'] == placemt

    @requires_api_version('1.27')
    def test_create_service_with_placement_preferences(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        placemt = docker.types.Placement(preferences=[
            {'Spread': {'SpreadDescriptor': 'com.dockerpy.test'}}
        ])
        task_tmpl = docker.types.TaskTemplate(
            container_spec, placement=placemt
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Placement' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['Placement'] == placemt

    @requires_api_version('1.27')
    def test_create_service_with_placement_preferences_tuple(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        placemt = docker.types.Placement(preferences=(
            ('spread', 'com.dockerpy.test'),
        ))
        task_tmpl = docker.types.TaskTemplate(
            container_spec, placement=placemt
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Placement' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['Placement'] == placemt

    def test_create_service_with_endpoint_spec(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        endpoint_spec = docker.types.EndpointSpec(ports={
            12357: (1990, 'udp'),
            12562: (678,),
            53243: 8080,
        })
        svc_id = self.client.create_service(
            task_tmpl, name=name, endpoint_spec=endpoint_spec
        )
        svc_info = self.client.inspect_service(svc_id)
        ports = svc_info['Spec']['EndpointSpec']['Ports']
        for port in ports:
            if port['PublishedPort'] == 12562:
                assert port['TargetPort'] == 678
                assert port['Protocol'] == 'tcp'
            elif port['PublishedPort'] == 53243:
                assert port['TargetPort'] == 8080
                assert port['Protocol'] == 'tcp'
            elif port['PublishedPort'] == 12357:
                assert port['TargetPort'] == 1990
                assert port['Protocol'] == 'udp'
            else:
                self.fail('Invalid port specification: {0}'.format(port))

        assert len(ports) == 3

    @requires_api_version('1.32')
    def test_create_service_with_endpoint_spec_host_publish_mode(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        endpoint_spec = docker.types.EndpointSpec(ports={
            12357: (1990, None, 'host'),
        })
        svc_id = self.client.create_service(
            task_tmpl, name=name, endpoint_spec=endpoint_spec
        )
        svc_info = self.client.inspect_service(svc_id)
        ports = svc_info['Spec']['EndpointSpec']['Ports']
        assert len(ports) == 1
        port = ports[0]
        assert port['PublishedPort'] == 12357
        assert port['TargetPort'] == 1990
        assert port['Protocol'] == 'tcp'
        assert port['PublishMode'] == 'host'

    def test_create_service_with_env(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['true'], env={'DOCKER_PY_TEST': 1}
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec,
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        con_spec = svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert 'Env' in con_spec
        assert con_spec['Env'] == ['DOCKER_PY_TEST=1']

    @requires_api_version('1.29')
    def test_create_service_with_update_order(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        update_config = docker.types.UpdateConfig(
            parallelism=10, delay=5, order='start-first'
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, update_config=update_config, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'UpdateConfig' in svc_info['Spec']
        uc = svc_info['Spec']['UpdateConfig']
        assert update_config['Parallelism'] == uc['Parallelism']
        assert update_config['Delay'] == uc['Delay']
        assert update_config['Order'] == uc['Order']

    @requires_api_version('1.25')
    def test_create_service_with_tty(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['true'], tty=True
        )
        task_tmpl = docker.types.TaskTemplate(
            container_spec,
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        con_spec = svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert 'TTY' in con_spec
        assert con_spec['TTY'] is True

    @requires_api_version('1.25')
    def test_create_service_with_tty_dict(self):
        container_spec = {
            'Image': TEST_IMG,
            'Command': ['true'],
            'TTY': True
        }
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        con_spec = svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert 'TTY' in con_spec
        assert con_spec['TTY'] is True

    def test_create_service_global_mode(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, mode='global'
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Mode' in svc_info['Spec']
        assert 'Global' in svc_info['Spec']['Mode']

    def test_create_service_replicated_mode(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name,
            mode=docker.types.ServiceMode('replicated', 5)
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Mode' in svc_info['Spec']
        assert 'Replicated' in svc_info['Spec']['Mode']
        assert svc_info['Spec']['Mode']['Replicated'] == {'Replicas': 5}

    @requires_api_version('1.25')
    def test_update_service_force_update(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ForceUpdate' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['ForceUpdate'] == 0
        version_index = svc_info['Version']['Index']

        task_tmpl = docker.types.TaskTemplate(container_spec, force_update=10)
        self.client.update_service(name, version_index, task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert svc_info['Spec']['TaskTemplate']['ForceUpdate'] == 10

    @requires_api_version('1.25')
    def test_create_service_with_secret(self):
        secret_name = 'favorite_touhou'
        secret_data = b'phantasmagoria of flower view'
        secret_id = self.client.create_secret(secret_name, secret_data)
        self.tmp_secrets.append(secret_id)
        secret_ref = docker.types.SecretReference(secret_id, secret_name)
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], secrets=[secret_ref]
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Secrets' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        secrets = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Secrets']
        assert secrets[0] == secret_ref

        container = self.get_service_container(name)
        assert container is not None
        exec_id = self.client.exec_create(
            container, 'cat /run/secrets/{0}'.format(secret_name)
        )
        assert self.client.exec_start(exec_id) == secret_data

    @requires_api_version('1.25')
    def test_create_service_with_unicode_secret(self):
        secret_name = 'favorite_touhou'
        secret_data = u'東方花映塚'
        secret_id = self.client.create_secret(secret_name, secret_data)
        self.tmp_secrets.append(secret_id)
        secret_ref = docker.types.SecretReference(secret_id, secret_name)
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], secrets=[secret_ref]
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Secrets' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        secrets = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Secrets']
        assert secrets[0] == secret_ref

        container = self.get_service_container(name)
        assert container is not None
        exec_id = self.client.exec_create(
            container, 'cat /run/secrets/{0}'.format(secret_name)
        )
        container_secret = self.client.exec_start(exec_id)
        container_secret = container_secret.decode('utf-8')
        assert container_secret == secret_data

    @requires_api_version('1.30')
    def test_create_service_with_config(self):
        config_name = 'favorite_touhou'
        config_data = b'phantasmagoria of flower view'
        config_id = self.client.create_config(config_name, config_data)
        self.tmp_configs.append(config_id)
        config_ref = docker.types.ConfigReference(config_id, config_name)
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], configs=[config_ref]
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Configs' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        configs = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Configs']
        assert configs[0] == config_ref

        container = self.get_service_container(name)
        assert container is not None
        exec_id = self.client.exec_create(
            container, 'cat /{0}'.format(config_name)
        )
        assert self.client.exec_start(exec_id) == config_data

    @requires_api_version('1.30')
    def test_create_service_with_unicode_config(self):
        config_name = 'favorite_touhou'
        config_data = u'東方花映塚'
        config_id = self.client.create_config(config_name, config_data)
        self.tmp_configs.append(config_id)
        config_ref = docker.types.ConfigReference(config_id, config_name)
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], configs=[config_ref]
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Configs' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        configs = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Configs']
        assert configs[0] == config_ref

        container = self.get_service_container(name)
        assert container is not None
        exec_id = self.client.exec_create(
            container, 'cat /{0}'.format(config_name)
        )
        container_config = self.client.exec_start(exec_id)
        container_config = container_config.decode('utf-8')
        assert container_config == config_data

    @requires_api_version('1.25')
    def test_create_service_with_hosts(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], hosts={
                'foobar': '127.0.0.1',
                'baz': '8.8.8.8',
            }
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Hosts' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        hosts = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Hosts']
        assert len(hosts) == 2
        assert '127.0.0.1 foobar' in hosts
        assert '8.8.8.8 baz' in hosts

    @requires_api_version('1.25')
    def test_create_service_with_hostname(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], hostname='foobar.baz.com'
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Hostname' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert (
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['Hostname'] ==
            'foobar.baz.com'
        )

    @requires_api_version('1.25')
    def test_create_service_with_groups(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], groups=['shrinemaidens', 'youkais']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Groups' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        groups = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Groups']
        assert len(groups) == 2
        assert 'shrinemaidens' in groups
        assert 'youkais' in groups

    @requires_api_version('1.25')
    def test_create_service_with_dns_config(self):
        dns_config = docker.types.DNSConfig(
            nameservers=['8.8.8.8', '8.8.4.4'],
            search=['local'], options=['debug']
        )
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], dns_config=dns_config
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'DNSConfig' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert (
            dns_config ==
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['DNSConfig']
        )

    @requires_api_version('1.25')
    def test_create_service_with_healthcheck(self):
        second = 1000000000
        hc = docker.types.Healthcheck(
            test='true', retries=3, timeout=1 * second,
            start_period=3 * second, interval=int(second / 2),
        )
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], healthcheck=hc
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert (
            'Healthcheck' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        )
        assert (
            hc ==
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['Healthcheck']
        )

    @requires_api_version('1.28')
    def test_create_service_with_readonly(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], read_only=True
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert (
            'ReadOnly' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        )
        assert svc_info['Spec']['TaskTemplate']['ContainerSpec']['ReadOnly']

    @requires_api_version('1.28')
    def test_create_service_with_stop_signal(self):
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], stop_signal='SIGINT'
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert (
            'StopSignal' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        )
        assert (
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['StopSignal'] ==
            'SIGINT'
        )

    @requires_api_version('1.30')
    def test_create_service_with_privileges(self):
        priv = docker.types.Privileges(selinux_disable=True)
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], privileges=priv
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert (
            'Privileges' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        )
        privileges = (
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['Privileges']
        )
        assert privileges['SELinuxContext']['Disable'] is True

    @requires_api_version('1.38')
    def test_create_service_with_init(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['sleep', '999'], init=True
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Init' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert (
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['Init'] is True
        )

    @requires_api_version('1.25')
    def test_update_service_with_defaults_name(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert 'Name' in svc_info['Spec']
        assert svc_info['Spec']['Name'] == name
        version_index = svc_info['Version']['Index']

        task_tmpl = docker.types.TaskTemplate(container_spec, force_update=10)
        self._update_service(
            svc_id, name, version_index, task_tmpl, fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'Name' in svc_info['Spec']
        assert svc_info['Spec']['Name'] == name

    @requires_api_version('1.25')
    def test_update_service_with_defaults_labels(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, labels={'service.label': 'SampleLabel'}
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Labels' in svc_info['Spec']
        assert 'service.label' in svc_info['Spec']['Labels']
        assert svc_info['Spec']['Labels']['service.label'] == 'SampleLabel'
        version_index = svc_info['Version']['Index']

        task_tmpl = docker.types.TaskTemplate(container_spec, force_update=10)
        self._update_service(
            svc_id, name, version_index, task_tmpl, name=name,
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'Labels' in svc_info['Spec']
        assert 'service.label' in svc_info['Spec']['Labels']
        assert svc_info['Spec']['Labels']['service.label'] == 'SampleLabel'

    def test_update_service_with_defaults_mode(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name,
            mode=docker.types.ServiceMode(mode='replicated', replicas=2)
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Mode' in svc_info['Spec']
        assert 'Replicated' in svc_info['Spec']['Mode']
        assert 'Replicas' in svc_info['Spec']['Mode']['Replicated']
        assert svc_info['Spec']['Mode']['Replicated']['Replicas'] == 2
        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, labels={'force': 'update'},
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'Mode' in svc_info['Spec']
        assert 'Replicated' in svc_info['Spec']['Mode']
        assert 'Replicas' in svc_info['Spec']['Mode']['Replicated']
        assert svc_info['Spec']['Mode']['Replicated']['Replicas'] == 2

    def test_update_service_with_defaults_container_labels(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello'],
            labels={'container.label': 'SampleLabel'}
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, labels={'service.label': 'SampleLabel'}
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        assert 'Labels' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        labels = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Labels']
        assert labels['container.label'] == 'SampleLabel'
        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, labels={'force': 'update'},
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        assert 'Labels' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        labels = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Labels']
        assert labels['container.label'] == 'SampleLabel'

        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        self._update_service(
            svc_id, name, new_index, task_tmpl, fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        newer_index = svc_info['Version']['Index']
        assert newer_index > new_index
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        assert 'Labels' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        labels = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Labels']
        assert labels['container.label'] == 'SampleLabel'

    def test_update_service_with_defaults_update_config(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        update_config = docker.types.UpdateConfig(
            parallelism=10, delay=5, failure_action='pause'
        )
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, update_config=update_config, name=name
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'UpdateConfig' in svc_info['Spec']
        uc = svc_info['Spec']['UpdateConfig']
        assert update_config['Parallelism'] == uc['Parallelism']
        assert update_config['Delay'] == uc['Delay']
        assert update_config['FailureAction'] == uc['FailureAction']
        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, labels={'force': 'update'},
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'UpdateConfig' in svc_info['Spec']
        uc = svc_info['Spec']['UpdateConfig']
        assert update_config['Parallelism'] == uc['Parallelism']
        assert update_config['Delay'] == uc['Delay']
        assert update_config['FailureAction'] == uc['FailureAction']

    def test_update_service_with_defaults_networks(self):
        net1 = self.client.create_network(
            'dockerpytest_1', driver='overlay', ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(net1['Id'])
        net2 = self.client.create_network(
            'dockerpytest_2', driver='overlay', ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(net2['Id'])
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, networks=[
                'dockerpytest_1', {'Target': 'dockerpytest_2'}
            ]
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Networks' in svc_info['Spec']
        assert svc_info['Spec']['Networks'] == [
            {'Target': net1['Id']}, {'Target': net2['Id']}
        ]

        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, labels={'force': 'update'},
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'Networks' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['Networks'] == [
            {'Target': net1['Id']}, {'Target': net2['Id']}
        ]

        self._update_service(
            svc_id, name, new_index, networks=[net1['Id']],
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Networks' in svc_info['Spec']['TaskTemplate']
        assert svc_info['Spec']['TaskTemplate']['Networks'] == [
            {'Target': net1['Id']}
        ]

    def test_update_service_with_defaults_endpoint_spec(self):
        container_spec = docker.types.ContainerSpec(TEST_IMG, ['true'])
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        endpoint_spec = docker.types.EndpointSpec(ports={
            12357: (1990, 'udp'),
            12562: (678,),
            53243: 8080,
        })
        svc_id = self.client.create_service(
            task_tmpl, name=name, endpoint_spec=endpoint_spec
        )
        svc_info = self.client.inspect_service(svc_id)
        print(svc_info)
        ports = svc_info['Spec']['EndpointSpec']['Ports']
        for port in ports:
            if port['PublishedPort'] == 12562:
                assert port['TargetPort'] == 678
                assert port['Protocol'] == 'tcp'
            elif port['PublishedPort'] == 53243:
                assert port['TargetPort'] == 8080
                assert port['Protocol'] == 'tcp'
            elif port['PublishedPort'] == 12357:
                assert port['TargetPort'] == 1990
                assert port['Protocol'] == 'udp'
            else:
                self.fail('Invalid port specification: {0}'.format(port))

        assert len(ports) == 3

        svc_info = self.client.inspect_service(svc_id)
        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, labels={'force': 'update'},
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index

        ports = svc_info['Spec']['EndpointSpec']['Ports']
        for port in ports:
            if port['PublishedPort'] == 12562:
                assert port['TargetPort'] == 678
                assert port['Protocol'] == 'tcp'
            elif port['PublishedPort'] == 53243:
                assert port['TargetPort'] == 8080
                assert port['Protocol'] == 'tcp'
            elif port['PublishedPort'] == 12357:
                assert port['TargetPort'] == 1990
                assert port['Protocol'] == 'udp'
            else:
                self.fail('Invalid port specification: {0}'.format(port))

        assert len(ports) == 3

    @requires_api_version('1.25')
    def test_update_service_remove_healthcheck(self):
        second = 1000000000
        hc = docker.types.Healthcheck(
            test='true', retries=3, timeout=1 * second,
            start_period=3 * second, interval=int(second / 2),
        )
        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], healthcheck=hc
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(task_tmpl, name=name)
        svc_info = self.client.inspect_service(svc_id)
        assert (
            'Healthcheck' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        )
        assert (
            hc ==
            svc_info['Spec']['TaskTemplate']['ContainerSpec']['Healthcheck']
        )

        container_spec = docker.types.ContainerSpec(
            TEST_IMG, ['sleep', '999'], healthcheck={}
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)

        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, task_tmpl, fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        container_spec = svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert (
            'Healthcheck' not in container_spec or
            not container_spec['Healthcheck']
        )

    def test_update_service_remove_labels(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, labels={'service.label': 'SampleLabel'}
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Labels' in svc_info['Spec']
        assert 'service.label' in svc_info['Spec']['Labels']
        assert svc_info['Spec']['Labels']['service.label'] == 'SampleLabel'
        version_index = svc_info['Version']['Index']

        self._update_service(
            svc_id, name, version_index, labels={}, fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert not svc_info['Spec'].get('Labels')

    def test_update_service_remove_container_labels(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello'],
            labels={'container.label': 'SampleLabel'}
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, labels={'service.label': 'SampleLabel'}
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        assert 'Labels' in svc_info['Spec']['TaskTemplate']['ContainerSpec']
        labels = svc_info['Spec']['TaskTemplate']['ContainerSpec']['Labels']
        assert labels['container.label'] == 'SampleLabel'
        version_index = svc_info['Version']['Index']

        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello'],
            labels={}
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        self._update_service(
            svc_id, name, version_index, task_tmpl, fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index
        assert 'TaskTemplate' in svc_info['Spec']
        assert 'ContainerSpec' in svc_info['Spec']['TaskTemplate']
        container_spec = svc_info['Spec']['TaskTemplate']['ContainerSpec']
        assert not container_spec.get('Labels')

    @requires_api_version('1.29')
    def test_update_service_with_network_change(self):
        container_spec = docker.types.ContainerSpec(
            'busybox', ['echo', 'hello']
        )
        task_tmpl = docker.types.TaskTemplate(container_spec)
        net1 = self.client.create_network(
            self.get_service_name(), driver='overlay',
            ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(net1['Id'])
        net2 = self.client.create_network(
            self.get_service_name(), driver='overlay',
            ipam={'Driver': 'default'}
        )
        self.tmp_networks.append(net2['Id'])
        name = self.get_service_name()
        svc_id = self.client.create_service(
            task_tmpl, name=name, networks=[net1['Id']]
        )
        svc_info = self.client.inspect_service(svc_id)
        assert 'Networks' in svc_info['Spec']
        assert len(svc_info['Spec']['Networks']) > 0
        assert svc_info['Spec']['Networks'][0]['Target'] == net1['Id']

        svc_info = self.client.inspect_service(svc_id)
        version_index = svc_info['Version']['Index']

        task_tmpl = docker.types.TaskTemplate(container_spec)
        self._update_service(
            svc_id, name, version_index, task_tmpl, name=name,
            networks=[net2['Id']], fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        task_template = svc_info['Spec']['TaskTemplate']
        assert 'Networks' in task_template
        assert len(task_template['Networks']) > 0
        assert task_template['Networks'][0]['Target'] == net2['Id']

        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']
        assert new_index > version_index

        self._update_service(
            svc_id, name, new_index, name=name, networks=[net1['Id']],
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        task_template = svc_info['Spec']['TaskTemplate']
        assert 'ContainerSpec' in task_template
        new_spec = task_template['ContainerSpec']
        assert 'Image' in new_spec
        assert new_spec['Image'].split(':')[0] == 'busybox'
        assert 'Command' in new_spec
        assert new_spec['Command'] == ['echo', 'hello']
        assert 'Networks' in task_template
        assert len(task_template['Networks']) > 0
        assert task_template['Networks'][0]['Target'] == net1['Id']

        svc_info = self.client.inspect_service(svc_id)
        new_index = svc_info['Version']['Index']

        task_tmpl = docker.types.TaskTemplate(
            container_spec, networks=[net2['Id']]
        )
        self._update_service(
            svc_id, name, new_index, task_tmpl, name=name,
            fetch_current_spec=True
        )
        svc_info = self.client.inspect_service(svc_id)
        task_template = svc_info['Spec']['TaskTemplate']
        assert 'Networks' in task_template
        assert len(task_template['Networks']) > 0
        assert task_template['Networks'][0]['Target'] == net2['Id']

    def _update_service(self, svc_id, *args, **kwargs):
        # service update tests seem to be a bit flaky
        # give them a chance to retry the update with a new version index
        try:
            self.client.update_service(*args, **kwargs)
        except docker.errors.APIError as e:
            if e.explanation.endswith("update out of sequence"):
                svc_info = self.client.inspect_service(svc_id)
                version_index = svc_info['Version']['Index']

                if len(args) > 1:
                    args = (args[0], version_index) + args[2:]
                else:
                    kwargs['version'] = version_index

                self.client.update_service(*args, **kwargs)
            else:
                raise