|
@@ -2,6 +2,7 @@ from __future__ import absolute_import
|
|
|
from __future__ import unicode_literals
|
|
|
|
|
|
import os
|
|
|
+import re
|
|
|
import shutil
|
|
|
import tempfile
|
|
|
from distutils.spawn import find_executable
|
|
@@ -62,41 +63,41 @@ class ServiceTest(DockerClientTestCase):
|
|
|
|
|
|
create_and_start_container(foo)
|
|
|
|
|
|
- self.assertEqual(len(foo.containers()), 1)
|
|
|
- self.assertEqual(foo.containers()[0].name, 'composetest_foo_1')
|
|
|
- self.assertEqual(len(bar.containers()), 0)
|
|
|
+ assert len(foo.containers()) == 1
|
|
|
+ assert foo.containers()[0].name == 'composetest_foo_1'
|
|
|
+ assert len(bar.containers()) == 0
|
|
|
|
|
|
create_and_start_container(bar)
|
|
|
create_and_start_container(bar)
|
|
|
|
|
|
- self.assertEqual(len(foo.containers()), 1)
|
|
|
- self.assertEqual(len(bar.containers()), 2)
|
|
|
+ assert len(foo.containers()) == 1
|
|
|
+ assert len(bar.containers()) == 2
|
|
|
|
|
|
names = [c.name for c in bar.containers()]
|
|
|
- self.assertIn('composetest_bar_1', names)
|
|
|
- self.assertIn('composetest_bar_2', names)
|
|
|
+ assert 'composetest_bar_1' in names
|
|
|
+ assert 'composetest_bar_2' in names
|
|
|
|
|
|
def test_containers_one_off(self):
|
|
|
db = self.create_service('db')
|
|
|
container = db.create_container(one_off=True)
|
|
|
- self.assertEqual(db.containers(stopped=True), [])
|
|
|
- self.assertEqual(db.containers(one_off=OneOffFilter.only, stopped=True), [container])
|
|
|
+ assert db.containers(stopped=True) == []
|
|
|
+ assert db.containers(one_off=OneOffFilter.only, stopped=True) == [container]
|
|
|
|
|
|
def test_project_is_added_to_container_name(self):
|
|
|
service = self.create_service('web')
|
|
|
create_and_start_container(service)
|
|
|
- self.assertEqual(service.containers()[0].name, 'composetest_web_1')
|
|
|
+ assert service.containers()[0].name == 'composetest_web_1'
|
|
|
|
|
|
def test_create_container_with_one_off(self):
|
|
|
db = self.create_service('db')
|
|
|
container = db.create_container(one_off=True)
|
|
|
- self.assertEqual(container.name, 'composetest_db_run_1')
|
|
|
+ assert container.name == 'composetest_db_run_1'
|
|
|
|
|
|
def test_create_container_with_one_off_when_existing_container_is_running(self):
|
|
|
db = self.create_service('db')
|
|
|
db.start()
|
|
|
container = db.create_container(one_off=True)
|
|
|
- self.assertEqual(container.name, 'composetest_db_run_1')
|
|
|
+ assert container.name == 'composetest_db_run_1'
|
|
|
|
|
|
def test_create_container_with_unspecified_volume(self):
|
|
|
service = self.create_service('db', volumes=[VolumeSpec.parse('/var/db')])
|
|
@@ -108,20 +109,20 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', volume_driver='foodriver')
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual('foodriver', container.get('HostConfig.VolumeDriver'))
|
|
|
+ assert 'foodriver' == container.get('HostConfig.VolumeDriver')
|
|
|
|
|
|
@pytest.mark.skipif(SWARM_SKIP_CPU_SHARES, reason='Swarm --cpu-shares bug')
|
|
|
def test_create_container_with_cpu_shares(self):
|
|
|
service = self.create_service('db', cpu_shares=73)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.CpuShares'), 73)
|
|
|
+ assert container.get('HostConfig.CpuShares') == 73
|
|
|
|
|
|
def test_create_container_with_cpu_quota(self):
|
|
|
service = self.create_service('db', cpu_quota=40000)
|
|
|
container = service.create_container()
|
|
|
container.start()
|
|
|
- self.assertEqual(container.get('HostConfig.CpuQuota'), 40000)
|
|
|
+ assert container.get('HostConfig.CpuQuota') == 40000
|
|
|
|
|
|
@v2_2_only()
|
|
|
def test_create_container_with_cpu_count(self):
|
|
@@ -129,7 +130,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', cpu_count=2)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.CpuCount'), 2)
|
|
|
+ assert container.get('HostConfig.CpuCount') == 2
|
|
|
|
|
|
@v2_2_only()
|
|
|
@pytest.mark.skipif(not IS_WINDOWS_PLATFORM, reason='cpu_percent is not supported for Linux')
|
|
@@ -138,7 +139,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', cpu_percent=12)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.CpuPercent'), 12)
|
|
|
+ assert container.get('HostConfig.CpuPercent') == 12
|
|
|
|
|
|
@v2_2_only()
|
|
|
def test_create_container_with_cpus(self):
|
|
@@ -146,14 +147,14 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', cpus=1)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.NanoCpus'), 1000000000)
|
|
|
+ assert container.get('HostConfig.NanoCpus') == 1000000000
|
|
|
|
|
|
def test_create_container_with_shm_size(self):
|
|
|
self.require_api_version('1.22')
|
|
|
service = self.create_service('db', shm_size=67108864)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.ShmSize'), 67108864)
|
|
|
+ assert container.get('HostConfig.ShmSize') == 67108864
|
|
|
|
|
|
def test_create_container_with_init_bool(self):
|
|
|
self.require_api_version('1.25')
|
|
@@ -184,7 +185,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', extra_hosts=extra_hosts)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(set(container.get('HostConfig.ExtraHosts')), set(extra_hosts))
|
|
|
+ assert set(container.get('HostConfig.ExtraHosts')) == set(extra_hosts)
|
|
|
|
|
|
def test_create_container_with_extra_hosts_dicts(self):
|
|
|
extra_hosts = {'somehost': '162.242.195.82', 'otherhost': '50.31.209.229'}
|
|
@@ -192,13 +193,13 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', extra_hosts=extra_hosts)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(set(container.get('HostConfig.ExtraHosts')), set(extra_hosts_list))
|
|
|
+ assert set(container.get('HostConfig.ExtraHosts')) == set(extra_hosts_list)
|
|
|
|
|
|
def test_create_container_with_cpu_set(self):
|
|
|
service = self.create_service('db', cpuset='0')
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.CpusetCpus'), '0')
|
|
|
+ assert container.get('HostConfig.CpusetCpus') == '0'
|
|
|
|
|
|
def test_create_container_with_read_only_root_fs(self):
|
|
|
read_only = True
|
|
@@ -240,7 +241,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', security_opt=security_opt)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(set(container.get('HostConfig.SecurityOpt')), set(security_opt))
|
|
|
+ assert set(container.get('HostConfig.SecurityOpt')) == set(security_opt)
|
|
|
|
|
|
@pytest.mark.xfail(True, reason='Not supported on most drivers')
|
|
|
def test_create_container_with_storage_opt(self):
|
|
@@ -248,7 +249,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', storage_opt=storage_opt)
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.get('HostConfig.StorageOpt'), storage_opt)
|
|
|
+ assert container.get('HostConfig.StorageOpt') == storage_opt
|
|
|
|
|
|
def test_create_container_with_oom_kill_disable(self):
|
|
|
self.require_api_version('1.20')
|
|
@@ -260,7 +261,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', mac_address='02:42:ac:11:65:43')
|
|
|
container = service.create_container()
|
|
|
service.start_container(container)
|
|
|
- self.assertEqual(container.inspect()['Config']['MacAddress'], '02:42:ac:11:65:43')
|
|
|
+ assert container.inspect()['Config']['MacAddress'] == '02:42:ac:11:65:43'
|
|
|
|
|
|
def test_create_container_with_specified_volume(self):
|
|
|
host_path = '/tmp/host-path'
|
|
@@ -276,8 +277,9 @@ class ServiceTest(DockerClientTestCase):
|
|
|
# Match the last component ("host-path"), because boot2docker symlinks /tmp
|
|
|
actual_host_path = container.get_mount(container_path)['Source']
|
|
|
|
|
|
- self.assertTrue(path.basename(actual_host_path) == path.basename(host_path),
|
|
|
- msg=("Last component differs: %s, %s" % (actual_host_path, host_path)))
|
|
|
+ assert path.basename(actual_host_path) == path.basename(host_path), (
|
|
|
+ "Last component differs: %s, %s" % (actual_host_path, host_path)
|
|
|
+ )
|
|
|
|
|
|
@v2_3_only()
|
|
|
def test_create_container_with_host_mount(self):
|
|
@@ -371,7 +373,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
volume_path = old_container.get_mount('/data')['Source']
|
|
|
|
|
|
new_container = service.recreate_container(old_container)
|
|
|
- self.assertEqual(new_container.get_mount('/data')['Source'], volume_path)
|
|
|
+ assert new_container.get_mount('/data')['Source'] == volume_path
|
|
|
|
|
|
def test_duplicate_volume_trailing_slash(self):
|
|
|
"""
|
|
@@ -393,20 +395,14 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('db', image=image, volumes=volumes)
|
|
|
old_container = create_and_start_container(service)
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- old_container.get('Config.Volumes'),
|
|
|
- {container_path: {}},
|
|
|
- )
|
|
|
+ assert old_container.get('Config.Volumes') == {container_path: {}}
|
|
|
|
|
|
service = self.create_service('db', image=image, volumes=volumes)
|
|
|
new_container = service.recreate_container(old_container)
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- new_container.get('Config.Volumes'),
|
|
|
- {container_path: {}},
|
|
|
- )
|
|
|
+ assert new_container.get('Config.Volumes') == {container_path: {}}
|
|
|
|
|
|
- self.assertEqual(service.containers(stopped=False), [new_container])
|
|
|
+ assert service.containers(stopped=False) == [new_container]
|
|
|
|
|
|
def test_create_container_with_volumes_from(self):
|
|
|
volume_service = self.create_service('data')
|
|
@@ -429,10 +425,8 @@ class ServiceTest(DockerClientTestCase):
|
|
|
)
|
|
|
host_container = host_service.create_container()
|
|
|
host_service.start_container(host_container)
|
|
|
- self.assertIn(volume_container_1.id + ':rw',
|
|
|
- host_container.get('HostConfig.VolumesFrom'))
|
|
|
- self.assertIn(volume_container_2.id + ':rw',
|
|
|
- host_container.get('HostConfig.VolumesFrom'))
|
|
|
+ assert volume_container_1.id + ':rw' in host_container.get('HostConfig.VolumesFrom')
|
|
|
+ assert volume_container_2.id + ':rw' in host_container.get('HostConfig.VolumesFrom')
|
|
|
|
|
|
def test_execute_convergence_plan_recreate(self):
|
|
|
service = self.create_service(
|
|
@@ -443,10 +437,10 @@ class ServiceTest(DockerClientTestCase):
|
|
|
command=['-d', '1']
|
|
|
)
|
|
|
old_container = service.create_container()
|
|
|
- self.assertEqual(old_container.get('Config.Entrypoint'), ['top'])
|
|
|
- self.assertEqual(old_container.get('Config.Cmd'), ['-d', '1'])
|
|
|
- self.assertIn('FOO=1', old_container.get('Config.Env'))
|
|
|
- self.assertEqual(old_container.name, 'composetest_db_1')
|
|
|
+ assert old_container.get('Config.Entrypoint') == ['top']
|
|
|
+ assert old_container.get('Config.Cmd') == ['-d', '1']
|
|
|
+ assert 'FOO=1' in old_container.get('Config.Env')
|
|
|
+ assert old_container.name == 'composetest_db_1'
|
|
|
service.start_container(old_container)
|
|
|
old_container.inspect() # reload volume data
|
|
|
volume_path = old_container.get_mount('/etc')['Source']
|
|
@@ -457,11 +451,11 @@ class ServiceTest(DockerClientTestCase):
|
|
|
new_container, = service.execute_convergence_plan(
|
|
|
ConvergencePlan('recreate', [old_container]))
|
|
|
|
|
|
- self.assertEqual(new_container.get('Config.Entrypoint'), ['top'])
|
|
|
- self.assertEqual(new_container.get('Config.Cmd'), ['-d', '1'])
|
|
|
- self.assertIn('FOO=2', new_container.get('Config.Env'))
|
|
|
- self.assertEqual(new_container.name, 'composetest_db_1')
|
|
|
- self.assertEqual(new_container.get_mount('/etc')['Source'], volume_path)
|
|
|
+ assert new_container.get('Config.Entrypoint') == ['top']
|
|
|
+ assert new_container.get('Config.Cmd') == ['-d', '1']
|
|
|
+ assert 'FOO=2' in new_container.get('Config.Env')
|
|
|
+ assert new_container.name == 'composetest_db_1'
|
|
|
+ assert new_container.get_mount('/etc')['Source'] == volume_path
|
|
|
if not is_cluster(self.client):
|
|
|
assert (
|
|
|
'affinity:container==%s' % old_container.id in
|
|
@@ -472,11 +466,10 @@ class ServiceTest(DockerClientTestCase):
|
|
|
# on the same node.
|
|
|
assert old_container.get('Node.Name') == new_container.get('Node.Name')
|
|
|
|
|
|
- self.assertEqual(len(self.client.containers(all=True)), num_containers_before)
|
|
|
- self.assertNotEqual(old_container.id, new_container.id)
|
|
|
- self.assertRaises(APIError,
|
|
|
- self.client.inspect_container,
|
|
|
- old_container.id)
|
|
|
+ assert len(self.client.containers(all=True)) == num_containers_before
|
|
|
+ assert old_container.id != new_container.id
|
|
|
+ with pytest.raises(APIError):
|
|
|
+ self.client.inspect_container(old_container.id)
|
|
|
|
|
|
def test_execute_convergence_plan_recreate_twice(self):
|
|
|
service = self.create_service(
|
|
@@ -550,17 +543,17 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service.create_container()
|
|
|
|
|
|
containers = service.containers(stopped=True)
|
|
|
- self.assertEqual(len(containers), 1)
|
|
|
+ assert len(containers) == 1
|
|
|
container, = containers
|
|
|
- self.assertFalse(container.is_running)
|
|
|
+ assert not container.is_running
|
|
|
|
|
|
service.execute_convergence_plan(ConvergencePlan('start', [container]))
|
|
|
|
|
|
containers = service.containers()
|
|
|
- self.assertEqual(len(containers), 1)
|
|
|
+ assert len(containers) == 1
|
|
|
container.inspect()
|
|
|
- self.assertEqual(container, containers[0])
|
|
|
- self.assertTrue(container.is_running)
|
|
|
+ assert container == containers[0]
|
|
|
+ assert container.is_running
|
|
|
|
|
|
def test_execute_convergence_plan_with_image_declared_volume(self):
|
|
|
service = Service(
|
|
@@ -571,19 +564,14 @@ class ServiceTest(DockerClientTestCase):
|
|
|
)
|
|
|
|
|
|
old_container = create_and_start_container(service)
|
|
|
- self.assertEqual(
|
|
|
- [mount['Destination'] for mount in old_container.get('Mounts')], ['/data']
|
|
|
- )
|
|
|
+ assert [mount['Destination'] for mount in old_container.get('Mounts')] == ['/data']
|
|
|
volume_path = old_container.get_mount('/data')['Source']
|
|
|
|
|
|
new_container, = service.execute_convergence_plan(
|
|
|
ConvergencePlan('recreate', [old_container]))
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- [mount['Destination'] for mount in new_container.get('Mounts')],
|
|
|
- ['/data']
|
|
|
- )
|
|
|
- self.assertEqual(new_container.get_mount('/data')['Source'], volume_path)
|
|
|
+ assert [mount['Destination'] for mount in new_container.get('Mounts')] == ['/data']
|
|
|
+ assert new_container.get_mount('/data')['Source'] == volume_path
|
|
|
|
|
|
def test_execute_convergence_plan_when_image_volume_masks_config(self):
|
|
|
service = self.create_service(
|
|
@@ -592,10 +580,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
)
|
|
|
|
|
|
old_container = create_and_start_container(service)
|
|
|
- self.assertEqual(
|
|
|
- [mount['Destination'] for mount in old_container.get('Mounts')],
|
|
|
- ['/data']
|
|
|
- )
|
|
|
+ assert [mount['Destination'] for mount in old_container.get('Mounts')] == ['/data']
|
|
|
volume_path = old_container.get_mount('/data')['Source']
|
|
|
|
|
|
service.options['volumes'] = [VolumeSpec.parse('/tmp:/data')]
|
|
@@ -606,15 +591,10 @@ class ServiceTest(DockerClientTestCase):
|
|
|
|
|
|
mock_log.warn.assert_called_once_with(mock.ANY)
|
|
|
_, args, kwargs = mock_log.warn.mock_calls[0]
|
|
|
- self.assertIn(
|
|
|
- "Service \"db\" is using volume \"/data\" from the previous container",
|
|
|
- args[0])
|
|
|
+ assert "Service \"db\" is using volume \"/data\" from the previous container" in args[0]
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- [mount['Destination'] for mount in new_container.get('Mounts')],
|
|
|
- ['/data']
|
|
|
- )
|
|
|
- self.assertEqual(new_container.get_mount('/data')['Source'], volume_path)
|
|
|
+ assert [mount['Destination'] for mount in new_container.get('Mounts')] == ['/data']
|
|
|
+ assert new_container.get_mount('/data')['Source'] == volume_path
|
|
|
|
|
|
def test_execute_convergence_plan_when_host_volume_is_removed(self):
|
|
|
host_path = '/tmp/host-path'
|
|
@@ -667,12 +647,12 @@ class ServiceTest(DockerClientTestCase):
|
|
|
def test_start_container_passes_through_options(self):
|
|
|
db = self.create_service('db')
|
|
|
create_and_start_container(db, environment={'FOO': 'BAR'})
|
|
|
- self.assertEqual(db.containers()[0].environment['FOO'], 'BAR')
|
|
|
+ assert db.containers()[0].environment['FOO'] == 'BAR'
|
|
|
|
|
|
def test_start_container_inherits_options_from_constructor(self):
|
|
|
db = self.create_service('db', environment={'FOO': 'BAR'})
|
|
|
create_and_start_container(db)
|
|
|
- self.assertEqual(db.containers()[0].environment['FOO'], 'BAR')
|
|
|
+ assert db.containers()[0].environment['FOO'] == 'BAR'
|
|
|
|
|
|
@no_cluster('No legacy links support in Swarm')
|
|
|
def test_start_container_creates_links(self):
|
|
@@ -683,13 +663,11 @@ class ServiceTest(DockerClientTestCase):
|
|
|
create_and_start_container(db)
|
|
|
create_and_start_container(web)
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- set(get_links(web.containers()[0])),
|
|
|
- set([
|
|
|
- 'composetest_db_1', 'db_1',
|
|
|
- 'composetest_db_2', 'db_2',
|
|
|
- 'db'])
|
|
|
- )
|
|
|
+ assert set(get_links(web.containers()[0])) == set([
|
|
|
+ 'composetest_db_1', 'db_1',
|
|
|
+ 'composetest_db_2', 'db_2',
|
|
|
+ 'db'
|
|
|
+ ])
|
|
|
|
|
|
@no_cluster('No legacy links support in Swarm')
|
|
|
def test_start_container_creates_links_with_names(self):
|
|
@@ -700,13 +678,11 @@ class ServiceTest(DockerClientTestCase):
|
|
|
create_and_start_container(db)
|
|
|
create_and_start_container(web)
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- set(get_links(web.containers()[0])),
|
|
|
- set([
|
|
|
- 'composetest_db_1', 'db_1',
|
|
|
- 'composetest_db_2', 'db_2',
|
|
|
- 'custom_link_name'])
|
|
|
- )
|
|
|
+ assert set(get_links(web.containers()[0])) == set([
|
|
|
+ 'composetest_db_1', 'db_1',
|
|
|
+ 'composetest_db_2', 'db_2',
|
|
|
+ 'custom_link_name'
|
|
|
+ ])
|
|
|
|
|
|
@no_cluster('No legacy links support in Swarm')
|
|
|
def test_start_container_with_external_links(self):
|
|
@@ -719,13 +695,11 @@ class ServiceTest(DockerClientTestCase):
|
|
|
create_and_start_container(db)
|
|
|
create_and_start_container(web)
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- set(get_links(web.containers()[0])),
|
|
|
- set([
|
|
|
- 'composetest_db_1',
|
|
|
- 'composetest_db_2',
|
|
|
- 'db_3']),
|
|
|
- )
|
|
|
+ assert set(get_links(web.containers()[0])) == set([
|
|
|
+ 'composetest_db_1',
|
|
|
+ 'composetest_db_2',
|
|
|
+ 'db_3'
|
|
|
+ ])
|
|
|
|
|
|
@no_cluster('No legacy links support in Swarm')
|
|
|
def test_start_normal_container_does_not_create_links_to_its_own_service(self):
|
|
@@ -735,7 +709,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
create_and_start_container(db)
|
|
|
|
|
|
c = create_and_start_container(db)
|
|
|
- self.assertEqual(set(get_links(c)), set([]))
|
|
|
+ assert set(get_links(c)) == set([])
|
|
|
|
|
|
@no_cluster('No legacy links support in Swarm')
|
|
|
def test_start_one_off_container_creates_links_to_its_own_service(self):
|
|
@@ -746,13 +720,11 @@ class ServiceTest(DockerClientTestCase):
|
|
|
|
|
|
c = create_and_start_container(db, one_off=OneOffFilter.only)
|
|
|
|
|
|
- self.assertEqual(
|
|
|
- set(get_links(c)),
|
|
|
- set([
|
|
|
- 'composetest_db_1', 'db_1',
|
|
|
- 'composetest_db_2', 'db_2',
|
|
|
- 'db'])
|
|
|
- )
|
|
|
+ assert set(get_links(c)) == set([
|
|
|
+ 'composetest_db_1', 'db_1',
|
|
|
+ 'composetest_db_2', 'db_2',
|
|
|
+ 'db'
|
|
|
+ ])
|
|
|
|
|
|
def test_start_container_builds_images(self):
|
|
|
service = Service(
|
|
@@ -763,7 +735,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
)
|
|
|
container = create_and_start_container(service)
|
|
|
container.wait()
|
|
|
- self.assertIn(b'success', container.logs())
|
|
|
+ assert b'success' in container.logs()
|
|
|
assert len(self.client.images(name='composetest_test')) >= 1
|
|
|
|
|
|
def test_start_container_uses_tagged_image_if_it_exists(self):
|
|
@@ -776,13 +748,13 @@ class ServiceTest(DockerClientTestCase):
|
|
|
)
|
|
|
container = create_and_start_container(service)
|
|
|
container.wait()
|
|
|
- self.assertIn(b'success', container.logs())
|
|
|
+ assert b'success' in container.logs()
|
|
|
|
|
|
def test_start_container_creates_ports(self):
|
|
|
service = self.create_service('web', ports=[8000])
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertEqual(list(container['NetworkSettings']['Ports'].keys()), ['8000/tcp'])
|
|
|
- self.assertNotEqual(container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'], '8000')
|
|
|
+ assert list(container['NetworkSettings']['Ports'].keys()) == ['8000/tcp']
|
|
|
+ assert container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'] != '8000'
|
|
|
|
|
|
def test_build(self):
|
|
|
base_dir = tempfile.mkdtemp()
|
|
@@ -959,34 +931,34 @@ class ServiceTest(DockerClientTestCase):
|
|
|
def test_start_container_stays_unprivileged(self):
|
|
|
service = self.create_service('web')
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertEqual(container['HostConfig']['Privileged'], False)
|
|
|
+ assert container['HostConfig']['Privileged'] is False
|
|
|
|
|
|
def test_start_container_becomes_privileged(self):
|
|
|
service = self.create_service('web', privileged=True)
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertEqual(container['HostConfig']['Privileged'], True)
|
|
|
+ assert container['HostConfig']['Privileged'] is True
|
|
|
|
|
|
def test_expose_does_not_publish_ports(self):
|
|
|
service = self.create_service('web', expose=["8000"])
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertEqual(container['NetworkSettings']['Ports'], {'8000/tcp': None})
|
|
|
+ assert container['NetworkSettings']['Ports'] == {'8000/tcp': None}
|
|
|
|
|
|
def test_start_container_creates_port_with_explicit_protocol(self):
|
|
|
service = self.create_service('web', ports=['8000/udp'])
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertEqual(list(container['NetworkSettings']['Ports'].keys()), ['8000/udp'])
|
|
|
+ assert list(container['NetworkSettings']['Ports'].keys()) == ['8000/udp']
|
|
|
|
|
|
def test_start_container_creates_fixed_external_ports(self):
|
|
|
service = self.create_service('web', ports=['8000:8000'])
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertIn('8000/tcp', container['NetworkSettings']['Ports'])
|
|
|
- self.assertEqual(container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'], '8000')
|
|
|
+ assert '8000/tcp' in container['NetworkSettings']['Ports']
|
|
|
+ assert container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'] == '8000'
|
|
|
|
|
|
def test_start_container_creates_fixed_external_ports_when_it_is_different_to_internal_port(self):
|
|
|
service = self.create_service('web', ports=['8001:8000'])
|
|
|
container = create_and_start_container(service).inspect()
|
|
|
- self.assertIn('8000/tcp', container['NetworkSettings']['Ports'])
|
|
|
- self.assertEqual(container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'], '8001')
|
|
|
+ assert '8000/tcp' in container['NetworkSettings']['Ports']
|
|
|
+ assert container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'] == '8001'
|
|
|
|
|
|
def test_port_with_explicit_interface(self):
|
|
|
service = self.create_service('web', ports=[
|
|
@@ -1026,21 +998,21 @@ class ServiceTest(DockerClientTestCase):
|
|
|
def test_scale(self):
|
|
|
service = self.create_service('web')
|
|
|
service.scale(1)
|
|
|
- self.assertEqual(len(service.containers()), 1)
|
|
|
+ assert len(service.containers()) == 1
|
|
|
|
|
|
# Ensure containers don't have stdout or stdin connected
|
|
|
container = service.containers()[0]
|
|
|
config = container.inspect()['Config']
|
|
|
- self.assertFalse(config['AttachStderr'])
|
|
|
- self.assertFalse(config['AttachStdout'])
|
|
|
- self.assertFalse(config['AttachStdin'])
|
|
|
+ assert not config['AttachStderr']
|
|
|
+ assert not config['AttachStdout']
|
|
|
+ assert not config['AttachStdin']
|
|
|
|
|
|
service.scale(3)
|
|
|
- self.assertEqual(len(service.containers()), 3)
|
|
|
+ assert len(service.containers()) == 3
|
|
|
service.scale(1)
|
|
|
- self.assertEqual(len(service.containers()), 1)
|
|
|
+ assert len(service.containers()) == 1
|
|
|
service.scale(0)
|
|
|
- self.assertEqual(len(service.containers()), 0)
|
|
|
+ assert len(service.containers()) == 0
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
|
SWARM_SKIP_CONTAINERS_ALL,
|
|
@@ -1061,12 +1033,12 @@ class ServiceTest(DockerClientTestCase):
|
|
|
with mock.patch('sys.stderr', new_callable=StringIO) as mock_stderr:
|
|
|
service.scale(2)
|
|
|
for container in service.containers():
|
|
|
- self.assertTrue(container.is_running)
|
|
|
- self.assertTrue(container.number in valid_numbers)
|
|
|
+ assert container.is_running
|
|
|
+ assert container.number in valid_numbers
|
|
|
|
|
|
captured_output = mock_stderr.getvalue()
|
|
|
- self.assertNotIn('Creating', captured_output)
|
|
|
- self.assertIn('Starting', captured_output)
|
|
|
+ assert 'Creating' not in captured_output
|
|
|
+ assert 'Starting' in captured_output
|
|
|
|
|
|
def test_scale_with_stopped_containers_and_needing_creation(self):
|
|
|
"""
|
|
@@ -1079,18 +1051,18 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service.create_container(number=next_number, quiet=True)
|
|
|
|
|
|
for container in service.containers():
|
|
|
- self.assertFalse(container.is_running)
|
|
|
+ assert not container.is_running
|
|
|
|
|
|
with mock.patch('sys.stderr', new_callable=StringIO) as mock_stderr:
|
|
|
service.scale(2)
|
|
|
|
|
|
- self.assertEqual(len(service.containers()), 2)
|
|
|
+ assert len(service.containers()) == 2
|
|
|
for container in service.containers():
|
|
|
- self.assertTrue(container.is_running)
|
|
|
+ assert container.is_running
|
|
|
|
|
|
captured_output = mock_stderr.getvalue()
|
|
|
- self.assertIn('Creating', captured_output)
|
|
|
- self.assertIn('Starting', captured_output)
|
|
|
+ assert 'Creating' in captured_output
|
|
|
+ assert 'Starting' in captured_output
|
|
|
|
|
|
def test_scale_with_api_error(self):
|
|
|
"""Test that when scaling if the API returns an error, that error is handled
|
|
@@ -1129,11 +1101,11 @@ class ServiceTest(DockerClientTestCase):
|
|
|
'compose.container.Container.create',
|
|
|
side_effect=ValueError("BOOM")
|
|
|
):
|
|
|
- with self.assertRaises(ValueError):
|
|
|
+ with pytest.raises(ValueError):
|
|
|
service.scale(3)
|
|
|
|
|
|
- self.assertEqual(len(service.containers()), 1)
|
|
|
- self.assertTrue(service.containers()[0].is_running)
|
|
|
+ assert len(service.containers()) == 1
|
|
|
+ assert service.containers()[0].is_running
|
|
|
|
|
|
@mock.patch('compose.service.log')
|
|
|
def test_scale_with_desired_number_already_achieved(self, mock_log):
|
|
@@ -1164,28 +1136,23 @@ class ServiceTest(DockerClientTestCase):
|
|
|
results in warning output.
|
|
|
"""
|
|
|
service = self.create_service('app', container_name='custom-container')
|
|
|
- self.assertEqual(service.custom_container_name, 'custom-container')
|
|
|
+ assert service.custom_container_name == 'custom-container'
|
|
|
|
|
|
with pytest.raises(OperationFailedError):
|
|
|
service.scale(3)
|
|
|
|
|
|
captured_output = mock_log.warn.call_args[0][0]
|
|
|
|
|
|
- self.assertEqual(len(service.containers()), 1)
|
|
|
- self.assertIn(
|
|
|
- "Remove the custom name to scale the service.",
|
|
|
- captured_output
|
|
|
- )
|
|
|
+ assert len(service.containers()) == 1
|
|
|
+ assert "Remove the custom name to scale the service." in captured_output
|
|
|
|
|
|
def test_scale_sets_ports(self):
|
|
|
service = self.create_service('web', ports=['8000'])
|
|
|
service.scale(2)
|
|
|
containers = service.containers()
|
|
|
- self.assertEqual(len(containers), 2)
|
|
|
+ assert len(containers) == 2
|
|
|
for container in containers:
|
|
|
- self.assertEqual(
|
|
|
- list(container.get('HostConfig.PortBindings')),
|
|
|
- ['8000/tcp'])
|
|
|
+ assert list(container.get('HostConfig.PortBindings')) == ['8000/tcp']
|
|
|
|
|
|
def test_scale_with_immediate_exit(self):
|
|
|
service = self.create_service('web', image='busybox', command='true')
|
|
@@ -1195,54 +1162,54 @@ class ServiceTest(DockerClientTestCase):
|
|
|
def test_network_mode_none(self):
|
|
|
service = self.create_service('web', network_mode=NetworkMode('none'))
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.NetworkMode'), 'none')
|
|
|
+ assert container.get('HostConfig.NetworkMode') == 'none'
|
|
|
|
|
|
def test_network_mode_bridged(self):
|
|
|
service = self.create_service('web', network_mode=NetworkMode('bridge'))
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.NetworkMode'), 'bridge')
|
|
|
+ assert container.get('HostConfig.NetworkMode') == 'bridge'
|
|
|
|
|
|
def test_network_mode_host(self):
|
|
|
service = self.create_service('web', network_mode=NetworkMode('host'))
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.NetworkMode'), 'host')
|
|
|
+ assert container.get('HostConfig.NetworkMode') == 'host'
|
|
|
|
|
|
def test_pid_mode_none_defined(self):
|
|
|
service = self.create_service('web', pid_mode=None)
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.PidMode'), '')
|
|
|
+ assert container.get('HostConfig.PidMode') == ''
|
|
|
|
|
|
def test_pid_mode_host(self):
|
|
|
service = self.create_service('web', pid_mode=PidMode('host'))
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.PidMode'), 'host')
|
|
|
+ assert container.get('HostConfig.PidMode') == 'host'
|
|
|
|
|
|
@v2_1_only()
|
|
|
def test_userns_mode_none_defined(self):
|
|
|
service = self.create_service('web', userns_mode=None)
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.UsernsMode'), '')
|
|
|
+ assert container.get('HostConfig.UsernsMode') == ''
|
|
|
|
|
|
@v2_1_only()
|
|
|
def test_userns_mode_host(self):
|
|
|
service = self.create_service('web', userns_mode='host')
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.UsernsMode'), 'host')
|
|
|
+ assert container.get('HostConfig.UsernsMode') == 'host'
|
|
|
|
|
|
def test_dns_no_value(self):
|
|
|
service = self.create_service('web')
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertIsNone(container.get('HostConfig.Dns'))
|
|
|
+ assert container.get('HostConfig.Dns') is None
|
|
|
|
|
|
def test_dns_list(self):
|
|
|
service = self.create_service('web', dns=['8.8.8.8', '9.9.9.9'])
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.Dns'), ['8.8.8.8', '9.9.9.9'])
|
|
|
+ assert container.get('HostConfig.Dns') == ['8.8.8.8', '9.9.9.9']
|
|
|
|
|
|
def test_mem_swappiness(self):
|
|
|
service = self.create_service('web', mem_swappiness=11)
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.MemorySwappiness'), 11)
|
|
|
+ assert container.get('HostConfig.MemorySwappiness') == 11
|
|
|
|
|
|
def test_mem_reservation(self):
|
|
|
service = self.create_service('web', mem_reservation='20m')
|
|
@@ -1252,12 +1219,12 @@ class ServiceTest(DockerClientTestCase):
|
|
|
def test_restart_always_value(self):
|
|
|
service = self.create_service('web', restart={'Name': 'always'})
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.RestartPolicy.Name'), 'always')
|
|
|
+ assert container.get('HostConfig.RestartPolicy.Name') == 'always'
|
|
|
|
|
|
def test_oom_score_adj_value(self):
|
|
|
service = self.create_service('web', oom_score_adj=500)
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.OomScoreAdj'), 500)
|
|
|
+ assert container.get('HostConfig.OomScoreAdj') == 500
|
|
|
|
|
|
def test_group_add_value(self):
|
|
|
service = self.create_service('web', group_add=["root", "1"])
|
|
@@ -1281,34 +1248,34 @@ class ServiceTest(DockerClientTestCase):
|
|
|
'MaximumRetryCount': 5
|
|
|
})
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.RestartPolicy.Name'), 'on-failure')
|
|
|
- self.assertEqual(container.get('HostConfig.RestartPolicy.MaximumRetryCount'), 5)
|
|
|
+ assert container.get('HostConfig.RestartPolicy.Name') == 'on-failure'
|
|
|
+ assert container.get('HostConfig.RestartPolicy.MaximumRetryCount') == 5
|
|
|
|
|
|
def test_cap_add_list(self):
|
|
|
service = self.create_service('web', cap_add=['SYS_ADMIN', 'NET_ADMIN'])
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.CapAdd'), ['SYS_ADMIN', 'NET_ADMIN'])
|
|
|
+ assert container.get('HostConfig.CapAdd') == ['SYS_ADMIN', 'NET_ADMIN']
|
|
|
|
|
|
def test_cap_drop_list(self):
|
|
|
service = self.create_service('web', cap_drop=['SYS_ADMIN', 'NET_ADMIN'])
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.CapDrop'), ['SYS_ADMIN', 'NET_ADMIN'])
|
|
|
+ assert container.get('HostConfig.CapDrop') == ['SYS_ADMIN', 'NET_ADMIN']
|
|
|
|
|
|
def test_dns_search(self):
|
|
|
service = self.create_service('web', dns_search=['dc1.example.com', 'dc2.example.com'])
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.DnsSearch'), ['dc1.example.com', 'dc2.example.com'])
|
|
|
+ assert container.get('HostConfig.DnsSearch') == ['dc1.example.com', 'dc2.example.com']
|
|
|
|
|
|
@v2_only()
|
|
|
def test_tmpfs(self):
|
|
|
service = self.create_service('web', tmpfs=['/run'])
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.get('HostConfig.Tmpfs'), {'/run': ''})
|
|
|
+ assert container.get('HostConfig.Tmpfs') == {'/run': ''}
|
|
|
|
|
|
def test_working_dir_param(self):
|
|
|
service = self.create_service('container', working_dir='/working/dir/sample')
|
|
|
container = service.create_container()
|
|
|
- self.assertEqual(container.get('Config.WorkingDir'), '/working/dir/sample')
|
|
|
+ assert container.get('Config.WorkingDir') == '/working/dir/sample'
|
|
|
|
|
|
def test_split_env(self):
|
|
|
service = self.create_service(
|
|
@@ -1316,7 +1283,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
environment=['NORMAL=F1', 'CONTAINS_EQUALS=F=2', 'TRAILING_EQUALS='])
|
|
|
env = create_and_start_container(service).environment
|
|
|
for k, v in {'NORMAL': 'F1', 'CONTAINS_EQUALS': 'F=2', 'TRAILING_EQUALS': ''}.items():
|
|
|
- self.assertEqual(env[k], v)
|
|
|
+ assert env[k] == v
|
|
|
|
|
|
def test_env_from_file_combined_with_env(self):
|
|
|
service = self.create_service(
|
|
@@ -1331,7 +1298,7 @@ class ServiceTest(DockerClientTestCase):
|
|
|
'FOO': 'baz',
|
|
|
'DOO': 'dah'
|
|
|
}.items():
|
|
|
- self.assertEqual(env[k], v)
|
|
|
+ assert env[k] == v
|
|
|
|
|
|
@v3_only()
|
|
|
def test_build_with_cachefrom(self):
|
|
@@ -1370,14 +1337,14 @@ class ServiceTest(DockerClientTestCase):
|
|
|
'ENV_DEF': 'E3',
|
|
|
'NO_DEF': None
|
|
|
}.items():
|
|
|
- self.assertEqual(env[k], v)
|
|
|
+ assert env[k] == v
|
|
|
|
|
|
def test_with_high_enough_api_version_we_get_default_network_mode(self):
|
|
|
# TODO: remove this test once minimum docker version is 1.8.x
|
|
|
with mock.patch.object(self.client, '_version', '1.20'):
|
|
|
service = self.create_service('web')
|
|
|
service_config = service._get_container_host_config({})
|
|
|
- self.assertEqual(service_config['NetworkMode'], 'default')
|
|
|
+ assert service_config['NetworkMode'] == 'default'
|
|
|
|
|
|
def test_labels(self):
|
|
|
labels_dict = {
|
|
@@ -1398,52 +1365,53 @@ class ServiceTest(DockerClientTestCase):
|
|
|
service = self.create_service('web', labels=labels_dict)
|
|
|
labels = create_and_start_container(service).labels.items()
|
|
|
for pair in expected.items():
|
|
|
- self.assertIn(pair, labels)
|
|
|
+ assert pair in labels
|
|
|
|
|
|
def test_empty_labels(self):
|
|
|
labels_dict = {'foo': '', 'bar': ''}
|
|
|
service = self.create_service('web', labels=labels_dict)
|
|
|
labels = create_and_start_container(service).labels.items()
|
|
|
for name in labels_dict:
|
|
|
- self.assertIn((name, ''), labels)
|
|
|
+ assert (name, '') in labels
|
|
|
|
|
|
def test_stop_signal(self):
|
|
|
stop_signal = 'SIGINT'
|
|
|
service = self.create_service('web', stop_signal=stop_signal)
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.stop_signal, stop_signal)
|
|
|
+ assert container.stop_signal == stop_signal
|
|
|
|
|
|
def test_custom_container_name(self):
|
|
|
service = self.create_service('web', container_name='my-web-container')
|
|
|
- self.assertEqual(service.custom_container_name, 'my-web-container')
|
|
|
+ assert service.custom_container_name == 'my-web-container'
|
|
|
|
|
|
container = create_and_start_container(service)
|
|
|
- self.assertEqual(container.name, 'my-web-container')
|
|
|
+ assert container.name == 'my-web-container'
|
|
|
|
|
|
one_off_container = service.create_container(one_off=True)
|
|
|
- self.assertNotEqual(one_off_container.name, 'my-web-container')
|
|
|
+ assert one_off_container.name != 'my-web-container'
|
|
|
|
|
|
@pytest.mark.skipif(True, reason="Broken on 1.11.0 - 17.03.0")
|
|
|
def test_log_drive_invalid(self):
|
|
|
service = self.create_service('web', logging={'driver': 'xxx'})
|
|
|
expected_error_msg = "logger: no log driver named 'xxx' is registered"
|
|
|
|
|
|
- with self.assertRaisesRegexp(APIError, expected_error_msg):
|
|
|
+ with pytest.raises(APIError) as excinfo:
|
|
|
create_and_start_container(service)
|
|
|
+ assert re.search(expected_error_msg, excinfo.value)
|
|
|
|
|
|
def test_log_drive_empty_default_jsonfile(self):
|
|
|
service = self.create_service('web')
|
|
|
log_config = create_and_start_container(service).log_config
|
|
|
|
|
|
- self.assertEqual('json-file', log_config['Type'])
|
|
|
- self.assertFalse(log_config['Config'])
|
|
|
+ assert 'json-file' == log_config['Type']
|
|
|
+ assert not log_config['Config']
|
|
|
|
|
|
def test_log_drive_none(self):
|
|
|
service = self.create_service('web', logging={'driver': 'none'})
|
|
|
log_config = create_and_start_container(service).log_config
|
|
|
|
|
|
- self.assertEqual('none', log_config['Type'])
|
|
|
- self.assertFalse(log_config['Config'])
|
|
|
+ assert 'none' == log_config['Type']
|
|
|
+ assert not log_config['Config']
|
|
|
|
|
|
def test_devices(self):
|
|
|
service = self.create_service('web', devices=["/dev/random:/dev/mapped-random"])
|
|
@@ -1455,8 +1423,8 @@ class ServiceTest(DockerClientTestCase):
|
|
|
'PathInContainer': '/dev/mapped-random'
|
|
|
}
|
|
|
|
|
|
- self.assertEqual(1, len(device_config))
|
|
|
- self.assertDictEqual(device_dict, device_config[0])
|
|
|
+ assert 1 == len(device_config)
|
|
|
+ assert device_dict == device_config[0]
|
|
|
|
|
|
def test_duplicate_containers(self):
|
|
|
service = self.create_service('web')
|
|
@@ -1464,14 +1432,14 @@ class ServiceTest(DockerClientTestCase):
|
|
|
options = service._get_container_create_options({}, 1)
|
|
|
original = Container.create(service.client, **options)
|
|
|
|
|
|
- self.assertEqual(set(service.containers(stopped=True)), set([original]))
|
|
|
- self.assertEqual(set(service.duplicate_containers()), set())
|
|
|
+ assert set(service.containers(stopped=True)) == set([original])
|
|
|
+ assert set(service.duplicate_containers()) == set()
|
|
|
|
|
|
options['name'] = 'temporary_container_name'
|
|
|
duplicate = Container.create(service.client, **options)
|
|
|
|
|
|
- self.assertEqual(set(service.containers(stopped=True)), set([original, duplicate]))
|
|
|
- self.assertEqual(set(service.duplicate_containers()), set([duplicate]))
|
|
|
+ assert set(service.containers(stopped=True)) == set([original, duplicate])
|
|
|
+ assert set(service.duplicate_containers()) == set([duplicate])
|
|
|
|
|
|
|
|
|
def converge(service, strategy=ConvergenceStrategy.changed):
|
|
@@ -1485,24 +1453,24 @@ class ConfigHashTest(DockerClientTestCase):
|
|
|
def test_no_config_hash_when_one_off(self):
|
|
|
web = self.create_service('web')
|
|
|
container = web.create_container(one_off=True)
|
|
|
- self.assertNotIn(LABEL_CONFIG_HASH, container.labels)
|
|
|
+ assert LABEL_CONFIG_HASH not in container.labels
|
|
|
|
|
|
def test_no_config_hash_when_overriding_options(self):
|
|
|
web = self.create_service('web')
|
|
|
container = web.create_container(environment={'FOO': '1'})
|
|
|
- self.assertNotIn(LABEL_CONFIG_HASH, container.labels)
|
|
|
+ assert LABEL_CONFIG_HASH not in container.labels
|
|
|
|
|
|
def test_config_hash_with_custom_labels(self):
|
|
|
web = self.create_service('web', labels={'foo': '1'})
|
|
|
container = converge(web)[0]
|
|
|
- self.assertIn(LABEL_CONFIG_HASH, container.labels)
|
|
|
- self.assertIn('foo', container.labels)
|
|
|
+ assert LABEL_CONFIG_HASH in container.labels
|
|
|
+ assert 'foo' in container.labels
|
|
|
|
|
|
def test_config_hash_sticks_around(self):
|
|
|
web = self.create_service('web', command=["top"])
|
|
|
container = converge(web)[0]
|
|
|
- self.assertIn(LABEL_CONFIG_HASH, container.labels)
|
|
|
+ assert LABEL_CONFIG_HASH in container.labels
|
|
|
|
|
|
web = self.create_service('web', command=["top", "-d", "1"])
|
|
|
container = converge(web)[0]
|
|
|
- self.assertIn(LABEL_CONFIG_HASH, container.labels)
|
|
|
+ assert LABEL_CONFIG_HASH in container.labels
|