jyapayne/compose
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
compose/tests/integration/service_test.py
Daniel Nephin 4b04280db8 Revert "Change special case from '_', None to ()" 
This reverts commit 677c50650c.

Revert "Modify service_test.py::ServiceTest::test_resolve_env to reflect new behavior"

This reverts commit 0019037712.

Revert "Mangle the tests. They pass for better or worse!"

This reverts commit 7ab9509ce6.

Revert "If an env var is passthrough but not defined on the host don't set it."

This reverts commit 6540efb3d3.

Signed-off-by: Daniel Nephin <dnephin@docker.com>
2016-02-23 11:13:57 -08:00

1056 lines
42 KiB
Python
Raw Blame History

from __future__ import absolute_import
from __future__ import unicode_literals
import os
import shutil
import tempfile
from os import path
from docker.errors import APIError
from six import StringIO
from six import text_type
from .. import mock
from .testcases import DockerClientTestCase
from .testcases import get_links
from .testcases import pull_busybox
from compose import __version__
from compose.config.types import VolumeFromSpec
from compose.config.types import VolumeSpec
from compose.const import LABEL_CONFIG_HASH
from compose.const import LABEL_CONTAINER_NUMBER
from compose.const import LABEL_ONE_OFF
from compose.const import LABEL_PROJECT
from compose.const import LABEL_SERVICE
from compose.const import LABEL_VERSION
from compose.container import Container
from compose.service import ConvergencePlan
from compose.service import ConvergenceStrategy
from compose.service import NetworkMode
from compose.service import Service
def create_and_start_container(service, **override_options):
container = service.create_container(**override_options)
return service.start_container(container)
class ServiceTest(DockerClientTestCase):
def test_containers(self):
foo = self.create_service('foo')
bar = self.create_service('bar')
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)
create_and_start_container(bar)
create_and_start_container(bar)
self.assertEqual(len(foo.containers()), 1)
self.assertEqual(len(bar.containers()), 2)
names = [c.name for c in bar.containers()]
self.assertIn('composetest_bar_1', names)
self.assertIn('composetest_bar_2', 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=True, 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')
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')
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')
def test_create_container_with_unspecified_volume(self):
service = self.create_service('db', volumes=[VolumeSpec.parse('/var/db')])
container = service.create_container()
service.start_container(container)
assert container.get_mount('/var/db')
def test_create_container_with_volume_driver(self):
service = self.create_service('db', volume_driver='foodriver')
container = service.create_container()
service.start_container(container)
self.assertEqual('foodriver', container.get('HostConfig.VolumeDriver'))
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)
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)
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)
def test_create_container_with_extra_hosts_list(self):
extra_hosts = ['somehost:162.242.195.82', 'otherhost:50.31.209.229']
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))
def test_create_container_with_extra_hosts_dicts(self):
extra_hosts = {'somehost': '162.242.195.82', 'otherhost': '50.31.209.229'}
extra_hosts_list = ['somehost:162.242.195.82', 'otherhost:50.31.209.229']
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))
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')
def test_create_container_with_read_only_root_fs(self):
read_only = True
service = self.create_service('db', read_only=read_only)
container = service.create_container()
service.start_container(container)
self.assertEqual(container.get('HostConfig.ReadonlyRootfs'), read_only, container.get('HostConfig'))
def test_create_container_with_security_opt(self):
security_opt = ['label:disable']
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))
def test_create_container_with_mac_address(self):
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')
def test_create_container_with_specified_volume(self):
host_path = '/tmp/host-path'
container_path = '/container-path'
service = self.create_service(
'db',
volumes=[VolumeSpec(host_path, container_path, 'rw')])
container = service.create_container()
service.start_container(container)
assert container.get_mount(container_path)
# 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)))
def test_recreate_preserves_volume_with_trailing_slash(self):
"""When the Compose file specifies a trailing slash in the container path, make
sure we copy the volume over when recreating.
"""
service = self.create_service('data', volumes=[VolumeSpec.parse('/data/')])
old_container = create_and_start_container(service)
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)
def test_duplicate_volume_trailing_slash(self):
"""
When an image specifies a volume, and the Compose file specifies a host path
but adds a trailing slash, make sure that we don't create duplicate binds.
"""
host_path = '/tmp/data'
container_path = '/data'
volumes = [VolumeSpec.parse('{}:{}/'.format(host_path, container_path))]
tmp_container = self.client.create_container(
'busybox', 'true',
volumes={container_path: {}},
labels={'com.docker.compose.test_image': 'true'},
)
image = self.client.commit(tmp_container)['Id']
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: {}},
)
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: {}},
)
self.assertEqual(service.containers(stopped=False), [new_container])
def test_create_container_with_volumes_from(self):
volume_service = self.create_service('data')
volume_container_1 = volume_service.create_container()
volume_container_2 = Container.create(
self.client,
image='busybox:latest',
command=["top"],
labels={LABEL_PROJECT: 'composetest'},
)
host_service = self.create_service(
'host',
volumes_from=[
VolumeFromSpec(volume_service, 'rw', 'service'),
VolumeFromSpec(volume_container_2, 'rw', 'container')
]
)
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'))
def test_execute_convergence_plan_recreate(self):
service = self.create_service(
'db',
environment={'FOO': '1'},
volumes=[VolumeSpec.parse('/etc')],
entrypoint=['top'],
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')
service.start_container(old_container)
old_container.inspect() # reload volume data
volume_path = old_container.get_mount('/etc')['Source']
num_containers_before = len(self.client.containers(all=True))
service.options['environment']['FOO'] = '2'
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)
self.assertIn(
'affinity:container==%s' % old_container.id,
new_container.get('Config.Env'))
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)
def test_execute_convergence_plan_recreate_twice(self):
service = self.create_service(
'db',
volumes=[VolumeSpec.parse('/etc')],
entrypoint=['top'],
command=['-d', '1'])
orig_container = service.create_container()
service.start_container(orig_container)
orig_container.inspect() # reload volume data
volume_path = orig_container.get_mount('/etc')['Source']
# Do this twice to reproduce the bug
for _ in range(2):
new_container, = service.execute_convergence_plan(
ConvergencePlan('recreate', [orig_container]))
assert new_container.get_mount('/etc')['Source'] == volume_path
assert ('affinity:container==%s' % orig_container.id in
new_container.get('Config.Env'))
orig_container = new_container
def test_execute_convergence_plan_when_containers_are_stopped(self):
service = self.create_service(
'db',
environment={'FOO': '1'},
volumes=[VolumeSpec.parse('/var/db')],
entrypoint=['top'],
command=['-d', '1']
)
service.create_container()
containers = service.containers(stopped=True)
self.assertEqual(len(containers), 1)
container, = containers
self.assertFalse(container.is_running)
service.execute_convergence_plan(ConvergencePlan('start', [container]))
containers = service.containers()
self.assertEqual(len(containers), 1)
container.inspect()
self.assertEqual(container, containers[0])
self.assertTrue(container.is_running)
def test_execute_convergence_plan_with_image_declared_volume(self):
service = Service(
project='composetest',
name='db',
client=self.client,
build={'context': 'tests/fixtures/dockerfile-with-volume'},
)
old_container = create_and_start_container(service)
self.assertEqual(
[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)
def test_execute_convergence_plan_when_image_volume_masks_config(self):
service = self.create_service(
'db',
build={'context': 'tests/fixtures/dockerfile-with-volume'},
)
old_container = create_and_start_container(service)
self.assertEqual(
[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')]
with mock.patch('compose.service.log') as mock_log:
new_container, = service.execute_convergence_plan(
ConvergencePlan('recreate', [old_container]))
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])
self.assertEqual(
[mount['Destination'] for mount in new_container.get('Mounts')],
['/data']
)
self.assertEqual(new_container.get_mount('/data')['Source'], volume_path)
def test_execute_convergence_plan_when_host_volume_is_removed(self):
host_path = '/tmp/host-path'
service = self.create_service(
'db',
build={'context': 'tests/fixtures/dockerfile-with-volume'},
volumes=[VolumeSpec(host_path, '/data', 'rw')])
old_container = create_and_start_container(service)
assert (
[mount['Destination'] for mount in old_container.get('Mounts')] ==
['/data']
)
service.options['volumes'] = []
with mock.patch('compose.service.log', autospec=True) as mock_log:
new_container, = service.execute_convergence_plan(
ConvergencePlan('recreate', [old_container]))
assert not mock_log.warn.called
assert (
[mount['Destination'] for mount in new_container.get('Mounts')],
['/data']
)
assert new_container.get_mount('/data')['Source'] != host_path
def test_execute_convergence_plan_without_start(self):
service = self.create_service(
'db',
build={'context': 'tests/fixtures/dockerfile-with-volume'}
)
containers = service.execute_convergence_plan(ConvergencePlan('create', []), start=False)
self.assertEqual(len(service.containers()), 0)
self.assertEqual(len(service.containers(stopped=True)), 1)
containers = service.execute_convergence_plan(ConvergencePlan('recreate', containers), start=False)
self.assertEqual(len(service.containers()), 0)
self.assertEqual(len(service.containers(stopped=True)), 1)
service.execute_convergence_plan(ConvergencePlan('start', containers), start=False)
self.assertEqual(len(service.containers()), 0)
self.assertEqual(len(service.containers(stopped=True)), 1)
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')
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')
def test_start_container_creates_links(self):
db = self.create_service('db')
web = self.create_service('web', links=[(db, None)])
create_and_start_container(db)
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'])
)
def test_start_container_creates_links_with_names(self):
db = self.create_service('db')
web = self.create_service('web', links=[(db, 'custom_link_name')])
create_and_start_container(db)
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'])
)
def test_start_container_with_external_links(self):
db = self.create_service('db')
web = self.create_service('web', external_links=['composetest_db_1',
'composetest_db_2',
'composetest_db_3:db_3'])
for _ in range(3):
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']),
)
def test_start_normal_container_does_not_create_links_to_its_own_service(self):
db = self.create_service('db')
create_and_start_container(db)
create_and_start_container(db)
c = create_and_start_container(db)
self.assertEqual(set(get_links(c)), set([]))
def test_start_one_off_container_creates_links_to_its_own_service(self):
db = self.create_service('db')
create_and_start_container(db)
create_and_start_container(db)
c = create_and_start_container(db, one_off=True)
self.assertEqual(
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(
name='test',
client=self.client,
build={'context': 'tests/fixtures/simple-dockerfile'},
project='composetest',
)
container = create_and_start_container(service)
container.wait()
self.assertIn(b'success', container.logs())
self.assertEqual(len(self.client.images(name='composetest_test')), 1)
def test_start_container_uses_tagged_image_if_it_exists(self):
self.check_build('tests/fixtures/simple-dockerfile', tag='composetest_test')
service = Service(
name='test',
client=self.client,
build={'context': 'this/does/not/exist/and/will/throw/error'},
project='composetest',
)
container = create_and_start_container(service)
container.wait()
self.assertIn(b'success', 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')
def test_build(self):
base_dir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base_dir)
with open(os.path.join(base_dir, 'Dockerfile'), 'w') as f:
f.write("FROM busybox\n")
self.create_service('web', build={'context': base_dir}).build()
assert self.client.inspect_image('composetest_web')
def test_build_non_ascii_filename(self):
base_dir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base_dir)
with open(os.path.join(base_dir, 'Dockerfile'), 'w') as f:
f.write("FROM busybox\n")
with open(os.path.join(base_dir.encode('utf8'), b'foo\xE2bar'), 'w') as f:
f.write("hello world\n")
self.create_service('web', build={'context': text_type(base_dir)}).build()
assert self.client.inspect_image('composetest_web')
def test_build_with_image_name(self):
base_dir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base_dir)
with open(os.path.join(base_dir, 'Dockerfile'), 'w') as f:
f.write("FROM busybox\n")
image_name = 'examples/composetest:latest'
self.addCleanup(self.client.remove_image, image_name)
self.create_service('web', build={'context': base_dir}, image=image_name).build()
assert self.client.inspect_image(image_name)
def test_build_with_git_url(self):
build_url = "https://github.com/dnephin/docker-build-from-url.git"
service = self.create_service('buildwithurl', build={'context': build_url})
self.addCleanup(self.client.remove_image, service.image_name)
service.build()
assert service.image()
def test_build_with_build_args(self):
base_dir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base_dir)
with open(os.path.join(base_dir, 'Dockerfile'), 'w') as f:
f.write("FROM busybox\n")
f.write("ARG build_version\n")
service = self.create_service('buildwithargs',
build={'context': text_type(base_dir),
'args': {"build_version": "1"}})
service.build()
assert service.image()
def test_start_container_stays_unpriviliged(self):
service = self.create_service('web')
container = create_and_start_container(service).inspect()
self.assertEqual(container['HostConfig']['Privileged'], False)
def test_start_container_becomes_priviliged(self):
service = self.create_service('web', privileged=True)
container = create_and_start_container(service).inspect()
self.assertEqual(container['HostConfig']['Privileged'], 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})
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'])
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')
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')
def test_port_with_explicit_interface(self):
service = self.create_service('web', ports=[
'127.0.0.1:8001:8000',
'0.0.0.0:9001:9000/udp',
])
container = create_and_start_container(service).inspect()
self.assertEqual(container['NetworkSettings']['Ports'], {
'8000/tcp': [
{
'HostIp': '127.0.0.1',
'HostPort': '8001',
},
],
'9000/udp': [
{
'HostIp': '0.0.0.0',
'HostPort': '9001',
},
],
})
def test_create_with_image_id(self):
# Get image id for the current busybox:latest
pull_busybox(self.client)
image_id = self.client.inspect_image('busybox:latest')['Id'][:12]
service = self.create_service('foo', image=image_id)
service.create_container()
def test_scale(self):
service = self.create_service('web')
service.scale(1)
self.assertEqual(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'])
service.scale(3)
self.assertEqual(len(service.containers()), 3)
service.scale(1)
self.assertEqual(len(service.containers()), 1)
service.scale(0)
self.assertEqual(len(service.containers()), 0)
def test_scale_with_stopped_containers(self):
"""
Given there are some stopped containers and scale is called with a
desired number that is the same as the number of stopped containers,
test that those containers are restarted and not removed/recreated.
"""
service = self.create_service('web')
next_number = service._next_container_number()
valid_numbers = [next_number, next_number + 1]
service.create_container(number=next_number)
service.create_container(number=next_number + 1)
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)
captured_output = mock_stderr.getvalue()
self.assertNotIn('Creating', captured_output)
self.assertIn('Starting', captured_output)
def test_scale_with_stopped_containers_and_needing_creation(self):
"""
Given there are some stopped containers and scale is called with a
desired number that is greater than the number of stopped containers,
test that those containers are restarted and required number are created.
"""
service = self.create_service('web')
next_number = service._next_container_number()
service.create_container(number=next_number, quiet=True)
for container in service.containers():
self.assertFalse(container.is_running)
with mock.patch('sys.stderr', new_callable=StringIO) as mock_stderr:
service.scale(2)
self.assertEqual(len(service.containers()), 2)
for container in service.containers():
self.assertTrue(container.is_running)
captured_output = mock_stderr.getvalue()
self.assertIn('Creating', captured_output)
self.assertIn('Starting', captured_output)
def test_scale_with_api_error(self):
"""Test that when scaling if the API returns an error, that error is handled
and the remaining threads continue.
"""
service = self.create_service('web')
next_number = service._next_container_number()
service.create_container(number=next_number, quiet=True)
with mock.patch(
'compose.container.Container.create',
side_effect=APIError(
message="testing",
response={},
explanation="Boom")):
with mock.patch('sys.stderr', new_callable=StringIO) as mock_stderr:
service.scale(3)
self.assertEqual(len(service.containers()), 1)
self.assertTrue(service.containers()[0].is_running)
self.assertIn("ERROR: for 2 Boom", mock_stderr.getvalue())
def test_scale_with_unexpected_exception(self):
"""Test that when scaling if the API returns an error, that is not of type
APIError, that error is re-raised.
"""
service = self.create_service('web')
next_number = service._next_container_number()
service.create_container(number=next_number, quiet=True)
with mock.patch(
'compose.container.Container.create',
side_effect=ValueError("BOOM")
):
with self.assertRaises(ValueError):
service.scale(3)
self.assertEqual(len(service.containers()), 1)
self.assertTrue(service.containers()[0].is_running)
@mock.patch('compose.service.log')
def test_scale_with_desired_number_already_achieved(self, mock_log):
"""
Test that calling scale with a desired number that is equal to the
number of containers already running results in no change.
"""
service = self.create_service('web')
next_number = service._next_container_number()
container = service.create_container(number=next_number, quiet=True)
container.start()
self.assertTrue(container.is_running)
self.assertEqual(len(service.containers()), 1)
service.scale(1)
self.assertEqual(len(service.containers()), 1)
container.inspect()
self.assertTrue(container.is_running)
captured_output = mock_log.info.call_args[0]
self.assertIn('Desired container number already achieved', captured_output)
@mock.patch('compose.service.log')
def test_scale_with_custom_container_name_outputs_warning(self, mock_log):
"""Test that calling scale on a service that has a custom container name
results in warning output.
"""
service = self.create_service('app', container_name='custom-container')
self.assertEqual(service.custom_container_name(), 'custom-container')
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
)
def test_scale_sets_ports(self):
service = self.create_service('web', ports=['8000'])
service.scale(2)
containers = service.containers()
self.assertEqual(len(containers), 2)
for container in containers:
self.assertEqual(list(container.inspect()['HostConfig']['PortBindings'].keys()), ['8000/tcp'])
def test_scale_with_immediate_exit(self):
service = self.create_service('web', image='busybox', command='true')
service.scale(2)
assert len(service.containers(stopped=True)) == 2
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')
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')
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')
def test_pid_mode_none_defined(self):
service = self.create_service('web', pid=None)
container = create_and_start_container(service)
self.assertEqual(container.get('HostConfig.PidMode'), '')
def test_pid_mode_host(self):
service = self.create_service('web', pid='host')
container = create_and_start_container(service)
self.assertEqual(container.get('HostConfig.PidMode'), 'host')
def test_dns_no_value(self):
service = self.create_service('web')
container = create_and_start_container(service)
self.assertIsNone(container.get('HostConfig.Dns'))
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'])
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')
def test_restart_on_failure_value(self):
service = self.create_service('web', restart={
'Name': 'on-failure',
'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)
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'])
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'])
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'])
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')
def test_split_env(self):
service = self.create_service('web', 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)
def test_env_from_file_combined_with_env(self):
service = self.create_service(
'web',
environment=['ONE=1', 'TWO=2', 'THREE=3'],
env_file=['tests/fixtures/env/one.env', 'tests/fixtures/env/two.env'])
env = create_and_start_container(service).environment
for k, v in {
'ONE': '1',
'TWO': '2',
'THREE': '3',
'FOO': 'baz',
'DOO': 'dah'
}.items():
self.assertEqual(env[k], v)
@mock.patch.dict(os.environ)
def test_resolve_env(self):
os.environ['FILE_DEF'] = 'E1'
os.environ['FILE_DEF_EMPTY'] = 'E2'
os.environ['ENV_DEF'] = 'E3'
service = self.create_service(
'web',
environment={
'FILE_DEF': 'F1',
'FILE_DEF_EMPTY': '',
'ENV_DEF': None,
'NO_DEF': None
}
)
env = create_and_start_container(service).environment
for k, v in {
'FILE_DEF': 'F1',
'FILE_DEF_EMPTY': '',
'ENV_DEF': 'E3',
'NO_DEF': ''
}.items():
self.assertEqual(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.assertEquals(service_config['NetworkMode'], 'default')
def test_labels(self):
labels_dict = {
'com.example.description': "Accounting webapp",
'com.example.department': "Finance",
'com.example.label-with-empty-value': "",
}
compose_labels = {
LABEL_CONTAINER_NUMBER: '1',
LABEL_ONE_OFF: 'False',
LABEL_PROJECT: 'composetest',
LABEL_SERVICE: 'web',
LABEL_VERSION: __version__,
}
expected = dict(labels_dict, **compose_labels)
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)
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)
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)
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')
container = create_and_start_container(service)
self.assertEqual(container.name, 'my-web-container')
one_off_container = service.create_container(one_off=True)
self.assertNotEqual(one_off_container.name, 'my-web-container')
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):
create_and_start_container(service)
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'])
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'])
def test_devices(self):
service = self.create_service('web', devices=["/dev/random:/dev/mapped-random"])
device_config = create_and_start_container(service).get('HostConfig.Devices')
device_dict = {
'PathOnHost': '/dev/random',
'CgroupPermissions': 'rwm',
'PathInContainer': '/dev/mapped-random'
}
self.assertEqual(1, len(device_config))
self.assertDictEqual(device_dict, device_config[0])
def test_duplicate_containers(self):
service = self.create_service('web')
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())
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]))
def converge(service,
strategy=ConvergenceStrategy.changed,
do_build=True):
"""Create a converge plan from a strategy and execute the plan."""
plan = service.convergence_plan(strategy)
return service.execute_convergence_plan(plan, do_build=do_build, timeout=1)
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)
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)
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)
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)
web = self.create_service('web', command=["top", "-d", "1"])
container = converge(web)[0]
self.assertIn(LABEL_CONFIG_HASH, container.labels)
Reference in a new issue View git blame Copy permalink