mirror of https://github.com/docker/docker-py.git
379 lines
13 KiB
Python
379 lines
13 KiB
Python
import unittest
|
|
|
|
import docker
|
|
import pytest
|
|
|
|
from .. import helpers
|
|
from .base import TEST_API_VERSION
|
|
from docker.errors import InvalidArgument
|
|
from docker.types.services import ServiceMode
|
|
|
|
|
|
class ServiceTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
helpers.force_leave_swarm(client)
|
|
client.swarm.init('127.0.0.1', listen_addr=helpers.swarm_listen_addr())
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
helpers.force_leave_swarm(docker.from_env(version=TEST_API_VERSION))
|
|
|
|
def test_create(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
name = helpers.random_name()
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=name,
|
|
labels={'foo': 'bar'},
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300",
|
|
container_labels={'container': 'label'},
|
|
rollback_config={'order': 'start-first'}
|
|
)
|
|
assert service.name == name
|
|
assert service.attrs['Spec']['Labels']['foo'] == 'bar'
|
|
container_spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert "alpine" in container_spec['Image']
|
|
assert container_spec['Labels'] == {'container': 'label'}
|
|
spec_rollback = service.attrs['Spec'].get('RollbackConfig', None)
|
|
assert spec_rollback is not None
|
|
assert ('Order' in spec_rollback and
|
|
spec_rollback['Order'] == 'start-first')
|
|
|
|
def test_create_with_network(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
name = helpers.random_name()
|
|
network = client.networks.create(
|
|
helpers.random_name(), driver='overlay'
|
|
)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=name,
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300",
|
|
networks=[network.id]
|
|
)
|
|
assert 'Networks' in service.attrs['Spec']['TaskTemplate']
|
|
networks = service.attrs['Spec']['TaskTemplate']['Networks']
|
|
assert len(networks) == 1
|
|
assert networks[0]['Target'] == network.id
|
|
|
|
def test_get(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
name = helpers.random_name()
|
|
service = client.services.create(
|
|
name=name,
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
service = client.services.get(service.id)
|
|
assert service.name == name
|
|
|
|
def test_list_remove(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
name=helpers.random_name(),
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
assert service in client.services.list()
|
|
service.remove()
|
|
assert service not in client.services.list()
|
|
|
|
def test_tasks(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service1 = client.services.create(
|
|
name=helpers.random_name(),
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
service2 = client.services.create(
|
|
name=helpers.random_name(),
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
tasks = []
|
|
while len(tasks) == 0:
|
|
tasks = service1.tasks()
|
|
assert len(tasks) == 1
|
|
assert tasks[0]['ServiceID'] == service1.id
|
|
|
|
tasks = []
|
|
while len(tasks) == 0:
|
|
tasks = service2.tasks()
|
|
assert len(tasks) == 1
|
|
assert tasks[0]['ServiceID'] == service2.id
|
|
|
|
def test_update(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
service.update(
|
|
# create argument
|
|
name=service.name,
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
container_spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert container_spec['Command'] == ["sleep", "600"]
|
|
|
|
def test_update_retains_service_labels(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
labels={'service.label': 'SampleLabel'},
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
service.update(
|
|
# create argument
|
|
name=service.name,
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
labels = service.attrs['Spec']['Labels']
|
|
assert labels == {'service.label': 'SampleLabel'}
|
|
|
|
def test_update_retains_container_labels(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300",
|
|
container_labels={'container.label': 'SampleLabel'}
|
|
)
|
|
service.update(
|
|
# create argument
|
|
name=service.name,
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
container_spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert container_spec['Labels'] == {'container.label': 'SampleLabel'}
|
|
|
|
def test_update_remove_service_labels(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
labels={'service.label': 'SampleLabel'},
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
service.update(
|
|
# create argument
|
|
name=service.name,
|
|
labels={},
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
assert not service.attrs['Spec'].get('Labels')
|
|
|
|
@pytest.mark.xfail(reason='Flaky test')
|
|
def test_update_retains_networks(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
network_name = helpers.random_name()
|
|
network = client.networks.create(
|
|
network_name, driver='overlay'
|
|
)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
networks=[network.id],
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
service.reload()
|
|
service.update(
|
|
# create argument
|
|
name=service.name,
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
networks = service.attrs['Spec']['TaskTemplate']['Networks']
|
|
assert networks == [{'Target': network.id}]
|
|
|
|
def test_scale_service(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
tasks = []
|
|
while len(tasks) == 0:
|
|
tasks = service.tasks()
|
|
assert len(tasks) == 1
|
|
service.update(
|
|
mode=docker.types.ServiceMode('replicated', replicas=2),
|
|
)
|
|
while len(tasks) == 1:
|
|
tasks = service.tasks()
|
|
assert len(tasks) >= 2
|
|
# check that the container spec is not overridden with None
|
|
service.reload()
|
|
spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert spec.get('Command') == ['sleep', '300']
|
|
|
|
def test_scale_method_service(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300",
|
|
)
|
|
tasks = []
|
|
while len(tasks) == 0:
|
|
tasks = service.tasks()
|
|
assert len(tasks) == 1
|
|
service.scale(2)
|
|
while len(tasks) == 1:
|
|
tasks = service.tasks()
|
|
assert len(tasks) >= 2
|
|
# check that the container spec is not overridden with None
|
|
service.reload()
|
|
spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert spec.get('Command') == ['sleep', '300']
|
|
|
|
def test_scale_method_global_service(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
mode = ServiceMode('global')
|
|
service = client.services.create(
|
|
name=helpers.random_name(),
|
|
image="alpine",
|
|
command="sleep 300",
|
|
mode=mode
|
|
)
|
|
tasks = []
|
|
while len(tasks) == 0:
|
|
tasks = service.tasks()
|
|
assert len(tasks) == 1
|
|
with pytest.raises(InvalidArgument):
|
|
service.scale(2)
|
|
|
|
assert len(tasks) == 1
|
|
service.reload()
|
|
spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert spec.get('Command') == ['sleep', '300']
|
|
|
|
@helpers.requires_api_version('1.25')
|
|
def test_force_update_service(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
initial_version = service.version
|
|
assert service.update(
|
|
# create argument
|
|
name=service.name,
|
|
# task template argument
|
|
force_update=10,
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
assert service.version > initial_version
|
|
|
|
@helpers.requires_api_version('1.25')
|
|
def test_force_update_service_using_bool(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
initial_version = service.version
|
|
assert service.update(
|
|
# create argument
|
|
name=service.name,
|
|
# task template argument
|
|
force_update=True,
|
|
# ContainerSpec argument
|
|
command="sleep 600"
|
|
)
|
|
service.reload()
|
|
assert service.version > initial_version
|
|
|
|
@helpers.requires_api_version('1.25')
|
|
def test_force_update_service_using_shorthand_method(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
service = client.services.create(
|
|
# create arguments
|
|
name=helpers.random_name(),
|
|
# ContainerSpec arguments
|
|
image="alpine",
|
|
command="sleep 300"
|
|
)
|
|
initial_version = service.version
|
|
assert service.force_update()
|
|
service.reload()
|
|
assert service.version > initial_version
|
|
|
|
@helpers.requires_api_version('1.41')
|
|
def test_create_cap_add(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
name = helpers.random_name()
|
|
service = client.services.create(
|
|
name=name,
|
|
labels={'foo': 'bar'},
|
|
image="alpine",
|
|
command="sleep 300",
|
|
container_labels={'container': 'label'},
|
|
cap_add=["CAP_SYSLOG"]
|
|
)
|
|
assert service.name == name
|
|
assert service.attrs['Spec']['Labels']['foo'] == 'bar'
|
|
container_spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert "alpine" in container_spec['Image']
|
|
assert container_spec['Labels'] == {'container': 'label'}
|
|
assert "CAP_SYSLOG" in container_spec["CapabilityAdd"]
|
|
|
|
@helpers.requires_api_version('1.41')
|
|
def test_create_cap_drop(self):
|
|
client = docker.from_env(version=TEST_API_VERSION)
|
|
name = helpers.random_name()
|
|
service = client.services.create(
|
|
name=name,
|
|
labels={'foo': 'bar'},
|
|
image="alpine",
|
|
command="sleep 300",
|
|
container_labels={'container': 'label'},
|
|
cap_drop=["CAP_SYSLOG"]
|
|
)
|
|
assert service.name == name
|
|
assert service.attrs['Spec']['Labels']['foo'] == 'bar'
|
|
container_spec = service.attrs['Spec']['TaskTemplate']['ContainerSpec']
|
|
assert "alpine" in container_spec['Image']
|
|
assert container_spec['Labels'] == {'container': 'label'}
|
|
assert "CAP_SYSLOG" in container_spec["CapabilityDrop"]
|