mirror of https://github.com/docker/compose.git
499 lines
21 KiB
Python
499 lines
21 KiB
Python
from __future__ import unicode_literals
|
|
from __future__ import absolute_import
|
|
import os
|
|
from os import path
|
|
import mock
|
|
|
|
from compose import Service
|
|
from compose.service import CannotBeScaledError
|
|
from compose.container import Container
|
|
from docker.errors import APIError
|
|
from .testcases import DockerClientTestCase
|
|
|
|
|
|
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_start_stop(self):
|
|
service = self.create_service('scalingtest')
|
|
self.assertEqual(len(service.containers(stopped=True)), 0)
|
|
|
|
service.create_container()
|
|
self.assertEqual(len(service.containers()), 0)
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
|
|
service.start()
|
|
self.assertEqual(len(service.containers()), 1)
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
|
|
service.stop(timeout=1)
|
|
self.assertEqual(len(service.containers()), 0)
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
|
|
service.stop(timeout=1)
|
|
self.assertEqual(len(service.containers()), 0)
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
|
|
def test_kill_remove(self):
|
|
service = self.create_service('scalingtest')
|
|
|
|
create_and_start_container(service)
|
|
self.assertEqual(len(service.containers()), 1)
|
|
|
|
service.remove_stopped()
|
|
self.assertEqual(len(service.containers()), 1)
|
|
|
|
service.kill()
|
|
self.assertEqual(len(service.containers()), 0)
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
|
|
service.remove_stopped()
|
|
self.assertEqual(len(service.containers(stopped=True)), 0)
|
|
|
|
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=['/var/db'])
|
|
container = service.create_container()
|
|
service.start_container(container)
|
|
self.assertIn('/var/db', container.inspect()['Volumes'])
|
|
|
|
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.inspect()['Config']['CpuShares'], 73)
|
|
|
|
def test_create_container_with_specified_volume(self):
|
|
host_path = '/tmp/host-path'
|
|
container_path = '/container-path'
|
|
|
|
service = self.create_service('db', volumes=['%s:%s' % (host_path, container_path)])
|
|
container = service.create_container()
|
|
service.start_container(container)
|
|
|
|
volumes = container.inspect()['Volumes']
|
|
self.assertIn(container_path, volumes)
|
|
|
|
# Match the last component ("host-path"), because boot2docker symlinks /tmp
|
|
actual_host_path = volumes[container_path]
|
|
self.assertTrue(path.basename(actual_host_path) == path.basename(host_path),
|
|
msg=("Last component differs: %s, %s" % (actual_host_path, host_path)))
|
|
|
|
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=["/bin/sleep", "300"])
|
|
host_service = self.create_service('host', volumes_from=[volume_service, volume_container_2])
|
|
host_container = host_service.create_container()
|
|
host_service.start_container(host_container)
|
|
self.assertIn(volume_container_1.id,
|
|
host_container.get('HostConfig.VolumesFrom'))
|
|
self.assertIn(volume_container_2.id,
|
|
host_container.get('HostConfig.VolumesFrom'))
|
|
|
|
def test_recreate_containers(self):
|
|
service = self.create_service(
|
|
'db',
|
|
environment={'FOO': '1'},
|
|
volumes=['/etc'],
|
|
entrypoint=['sleep'],
|
|
command=['300']
|
|
)
|
|
old_container = service.create_container()
|
|
self.assertEqual(old_container.dictionary['Config']['Entrypoint'], ['sleep'])
|
|
self.assertEqual(old_container.dictionary['Config']['Cmd'], ['300'])
|
|
self.assertIn('FOO=1', old_container.dictionary['Config']['Env'])
|
|
self.assertEqual(old_container.name, 'composetest_db_1')
|
|
service.start_container(old_container)
|
|
volume_path = old_container.inspect()['Volumes']['/etc']
|
|
|
|
num_containers_before = len(self.client.containers(all=True))
|
|
|
|
service.options['environment']['FOO'] = '2'
|
|
tuples = service.recreate_containers()
|
|
self.assertEqual(len(tuples), 1)
|
|
|
|
intermediate_container = tuples[0][0]
|
|
new_container = tuples[0][1]
|
|
self.assertEqual(intermediate_container.dictionary['Config']['Entrypoint'], ['/bin/echo'])
|
|
|
|
self.assertEqual(new_container.dictionary['Config']['Entrypoint'], ['sleep'])
|
|
self.assertEqual(new_container.dictionary['Config']['Cmd'], ['300'])
|
|
self.assertIn('FOO=2', new_container.dictionary['Config']['Env'])
|
|
self.assertEqual(new_container.name, 'composetest_db_1')
|
|
self.assertEqual(new_container.inspect()['Volumes']['/etc'], volume_path)
|
|
self.assertIn(intermediate_container.id, new_container.dictionary['HostConfig']['VolumesFrom'])
|
|
|
|
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,
|
|
intermediate_container.id)
|
|
|
|
def test_recreate_containers_when_containers_are_stopped(self):
|
|
service = self.create_service(
|
|
'db',
|
|
environment={'FOO': '1'},
|
|
volumes=['/var/db'],
|
|
entrypoint=['sleep'],
|
|
command=['300']
|
|
)
|
|
service.create_container()
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
service.recreate_containers()
|
|
self.assertEqual(len(service.containers(stopped=True)), 1)
|
|
|
|
def test_recreate_containers_with_image_declared_volume(self):
|
|
service = Service(
|
|
project='composetest',
|
|
name='db',
|
|
client=self.client,
|
|
build='tests/fixtures/dockerfile-with-volume',
|
|
)
|
|
|
|
old_container = create_and_start_container(service)
|
|
self.assertEqual(old_container.get('Volumes').keys(), ['/data'])
|
|
volume_path = old_container.get('Volumes')['/data']
|
|
|
|
service.recreate_containers()
|
|
new_container = service.containers()[0]
|
|
service.start_container(new_container)
|
|
self.assertEqual(new_container.get('Volumes').keys(), ['/data'])
|
|
self.assertEqual(new_container.get('Volumes')['/data'], volume_path)
|
|
|
|
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(web.containers()[0].links()),
|
|
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(web.containers()[0].links()),
|
|
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(web.containers()[0].links()),
|
|
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(c.links()), 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(c.links()),
|
|
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='tests/fixtures/simple-dockerfile',
|
|
project='composetest',
|
|
)
|
|
container = create_and_start_container(service)
|
|
container.wait()
|
|
self.assertIn('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.client.build('tests/fixtures/simple-dockerfile', tag='composetest_test')
|
|
service = Service(
|
|
name='test',
|
|
client=self.client,
|
|
build='this/does/not/exist/and/will/throw/error',
|
|
project='composetest',
|
|
)
|
|
container = create_and_start_container(service)
|
|
container.wait()
|
|
self.assertIn('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_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_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_on_service_that_cannot_be_scaled(self):
|
|
service = self.create_service('web', ports=['8000:8000'])
|
|
self.assertRaises(CannotBeScaledError, lambda: service.scale(1))
|
|
|
|
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_network_mode_none(self):
|
|
service = self.create_service('web', net='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', net='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', net='host')
|
|
container = create_and_start_container(service)
|
|
self.assertEqual(container.get('HostConfig.NetworkMode'), '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_single_value(self):
|
|
service = self.create_service('web', dns='8.8.8.8')
|
|
container = create_and_start_container(service)
|
|
self.assertEqual(container.get('HostConfig.Dns'), ['8.8.8.8'])
|
|
|
|
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='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='on-failure: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_no_value(self):
|
|
service = self.create_service('web')
|
|
container = create_and_start_container(service)
|
|
self.assertIsNone(container.get('HostConfig.DnsSearch'))
|
|
|
|
def test_dns_search_single_value(self):
|
|
service = self.create_service('web', dns_search='example.com')
|
|
container = create_and_start_container(service)
|
|
self.assertEqual(container.get('HostConfig.DnsSearch'), ['example.com'])
|
|
|
|
def test_dns_search_list(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)
|