mirror of https://github.com/docker/compose.git
663 lines
24 KiB
Python
663 lines
24 KiB
Python
from __future__ import unicode_literals
|
|
from __future__ import absolute_import
|
|
|
|
from .. import unittest
|
|
import mock
|
|
|
|
import docker
|
|
from docker.utils import LogConfig
|
|
|
|
from .. import mock
|
|
from .. import unittest
|
|
from compose.const import LABEL_CONFIG_HASH
|
|
from compose.const import LABEL_ONE_OFF
|
|
from compose.const import LABEL_PROJECT
|
|
from compose.const import LABEL_SERVICE
|
|
from compose.container import Container
|
|
from compose.service import ConfigError
|
|
from compose.service import ContainerNet
|
|
from compose.service import NeedsBuildError
|
|
from compose.service import Net
|
|
from compose.service import NoSuchImageError
|
|
from compose.service import Service
|
|
from compose.service import ServiceNet
|
|
from compose.service import build_port_bindings
|
|
from compose.service import build_volume_binding
|
|
from compose.service import get_container_data_volumes
|
|
from compose.service import merge_volume_bindings
|
|
from compose.service import parse_repository_tag
|
|
from compose.service import parse_volume_spec
|
|
from compose.service import split_port
|
|
|
|
|
|
class ServiceTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.mock_client = mock.create_autospec(docker.Client)
|
|
|
|
def test_name_validations(self):
|
|
self.assertRaises(ConfigError, lambda: Service(name='', image='foo'))
|
|
|
|
self.assertRaises(ConfigError, lambda: Service(name=' ', image='foo'))
|
|
self.assertRaises(ConfigError, lambda: Service(name='/', image='foo'))
|
|
self.assertRaises(ConfigError, lambda: Service(name='!', image='foo'))
|
|
self.assertRaises(ConfigError, lambda: Service(name='\xe2', image='foo'))
|
|
|
|
Service('a', image='foo')
|
|
Service('foo', image='foo')
|
|
Service('foo-bar', image='foo')
|
|
Service('foo.bar', image='foo')
|
|
Service('foo_bar', image='foo')
|
|
Service('_', image='foo')
|
|
Service('___', image='foo')
|
|
Service('-', image='foo')
|
|
Service('--', image='foo')
|
|
Service('.__.', image='foo')
|
|
|
|
def test_project_validation(self):
|
|
self.assertRaises(ConfigError, lambda: Service('bar'))
|
|
self.assertRaises(ConfigError, lambda: Service(name='foo', project='>', image='foo'))
|
|
|
|
Service(name='foo', project='bar.bar__', image='foo')
|
|
|
|
def test_containers(self):
|
|
service = Service('db', self.mock_client, 'myproject', image='foo')
|
|
self.mock_client.containers.return_value = []
|
|
self.assertEqual(service.containers(), [])
|
|
|
|
def test_containers_with_containers(self):
|
|
self.mock_client.containers.return_value = [
|
|
dict(Name=str(i), Image='foo', Id=i) for i in range(3)
|
|
]
|
|
service = Service('db', self.mock_client, 'myproject', image='foo')
|
|
self.assertEqual([c.id for c in service.containers()], range(3))
|
|
|
|
expected_labels = [
|
|
'{0}=myproject'.format(LABEL_PROJECT),
|
|
'{0}=db'.format(LABEL_SERVICE),
|
|
'{0}=False'.format(LABEL_ONE_OFF),
|
|
]
|
|
|
|
self.mock_client.containers.assert_called_once_with(
|
|
all=False,
|
|
filters={'label': expected_labels})
|
|
|
|
def test_container_without_name(self):
|
|
self.mock_client.containers.return_value = [
|
|
{'Image': 'foo', 'Id': '1', 'Name': '1'},
|
|
{'Image': 'foo', 'Id': '2', 'Name': None},
|
|
{'Image': 'foo', 'Id': '3'},
|
|
]
|
|
service = Service('db', self.mock_client, 'myproject', image='foo')
|
|
|
|
self.assertEqual([c.id for c in service.containers()], ['1'])
|
|
self.assertEqual(service._next_container_number(), 2)
|
|
self.assertEqual(service.get_container(1).id, '1')
|
|
|
|
def test_get_volumes_from_container(self):
|
|
container_id = 'aabbccddee'
|
|
service = Service(
|
|
'test',
|
|
image='foo',
|
|
volumes_from=[mock.Mock(id=container_id, spec=Container)])
|
|
|
|
self.assertEqual(service._get_volumes_from(), [container_id])
|
|
|
|
def test_get_volumes_from_service_container_exists(self):
|
|
container_ids = ['aabbccddee', '12345']
|
|
from_service = mock.create_autospec(Service)
|
|
from_service.containers.return_value = [
|
|
mock.Mock(id=container_id, spec=Container)
|
|
for container_id in container_ids
|
|
]
|
|
service = Service('test', volumes_from=[from_service], image='foo')
|
|
|
|
self.assertEqual(service._get_volumes_from(), container_ids)
|
|
|
|
def test_get_volumes_from_service_no_container(self):
|
|
container_id = 'abababab'
|
|
from_service = mock.create_autospec(Service)
|
|
from_service.containers.return_value = []
|
|
from_service.create_container.return_value = mock.Mock(
|
|
id=container_id,
|
|
spec=Container)
|
|
service = Service('test', image='foo', volumes_from=[from_service])
|
|
|
|
self.assertEqual(service._get_volumes_from(), [container_id])
|
|
from_service.create_container.assert_called_once_with()
|
|
|
|
def test_split_port_with_host_ip(self):
|
|
internal_port, external_port = split_port("127.0.0.1:1000:2000")
|
|
self.assertEqual(internal_port, "2000")
|
|
self.assertEqual(external_port, ("127.0.0.1", "1000"))
|
|
|
|
def test_split_port_with_protocol(self):
|
|
internal_port, external_port = split_port("127.0.0.1:1000:2000/udp")
|
|
self.assertEqual(internal_port, "2000/udp")
|
|
self.assertEqual(external_port, ("127.0.0.1", "1000"))
|
|
|
|
def test_split_port_with_host_ip_no_port(self):
|
|
internal_port, external_port = split_port("127.0.0.1::2000")
|
|
self.assertEqual(internal_port, "2000")
|
|
self.assertEqual(external_port, ("127.0.0.1", None))
|
|
|
|
def test_split_port_with_host_port(self):
|
|
internal_port, external_port = split_port("1000:2000")
|
|
self.assertEqual(internal_port, "2000")
|
|
self.assertEqual(external_port, "1000")
|
|
|
|
def test_split_port_no_host_port(self):
|
|
internal_port, external_port = split_port("2000")
|
|
self.assertEqual(internal_port, "2000")
|
|
self.assertEqual(external_port, None)
|
|
|
|
def test_split_port_invalid(self):
|
|
with self.assertRaises(ConfigError):
|
|
split_port("0.0.0.0:1000:2000:tcp")
|
|
|
|
def test_build_port_bindings_with_one_port(self):
|
|
port_bindings = build_port_bindings(["127.0.0.1:1000:1000"])
|
|
self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
|
|
|
|
def test_build_port_bindings_with_matching_internal_ports(self):
|
|
port_bindings = build_port_bindings(["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"])
|
|
self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000"), ("127.0.0.1", "2000")])
|
|
|
|
def test_build_port_bindings_with_nonmatching_internal_ports(self):
|
|
port_bindings = build_port_bindings(["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"])
|
|
self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
|
|
self.assertEqual(port_bindings["2000"], [("127.0.0.1", "2000")])
|
|
|
|
def test_split_domainname_none(self):
|
|
service = Service('foo', image='foo', hostname='name', client=self.mock_client)
|
|
self.mock_client.containers.return_value = []
|
|
opts = service._get_container_create_options({'image': 'foo'}, 1)
|
|
self.assertEqual(opts['hostname'], 'name', 'hostname')
|
|
self.assertFalse('domainname' in opts, 'domainname')
|
|
|
|
def test_memory_swap_limit(self):
|
|
service = Service(name='foo', image='foo', hostname='name', client=self.mock_client, mem_limit=1000000000, memswap_limit=2000000000)
|
|
self.mock_client.containers.return_value = []
|
|
opts = service._get_container_create_options({'some': 'overrides'}, 1)
|
|
self.assertEqual(opts['host_config']['MemorySwap'], 2000000000)
|
|
self.assertEqual(opts['host_config']['Memory'], 1000000000)
|
|
|
|
def test_log_opt(self):
|
|
log_opt = {'address': 'tcp://192.168.0.42:123'}
|
|
service = Service(name='foo', image='foo', hostname='name', client=self.mock_client, log_driver='syslog', log_opt=log_opt)
|
|
self.mock_client.containers.return_value = []
|
|
opts = service._get_container_create_options({'some': 'overrides'}, 1)
|
|
|
|
self.assertIsInstance(opts['host_config']['LogConfig'], LogConfig)
|
|
self.assertEqual(opts['host_config']['LogConfig'].type, 'syslog')
|
|
self.assertEqual(opts['host_config']['LogConfig'].config, log_opt)
|
|
|
|
def test_split_domainname_fqdn(self):
|
|
service = Service(
|
|
'foo',
|
|
hostname='name.domain.tld',
|
|
image='foo',
|
|
client=self.mock_client)
|
|
self.mock_client.containers.return_value = []
|
|
opts = service._get_container_create_options({'image': 'foo'}, 1)
|
|
self.assertEqual(opts['hostname'], 'name', 'hostname')
|
|
self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
|
|
|
|
def test_split_domainname_both(self):
|
|
service = Service(
|
|
'foo',
|
|
hostname='name',
|
|
image='foo',
|
|
domainname='domain.tld',
|
|
client=self.mock_client)
|
|
self.mock_client.containers.return_value = []
|
|
opts = service._get_container_create_options({'image': 'foo'}, 1)
|
|
self.assertEqual(opts['hostname'], 'name', 'hostname')
|
|
self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
|
|
|
|
def test_split_domainname_weird(self):
|
|
service = Service(
|
|
'foo',
|
|
hostname='name.sub',
|
|
domainname='domain.tld',
|
|
image='foo',
|
|
client=self.mock_client)
|
|
self.mock_client.containers.return_value = []
|
|
opts = service._get_container_create_options({'image': 'foo'}, 1)
|
|
self.assertEqual(opts['hostname'], 'name.sub', 'hostname')
|
|
self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
|
|
|
|
def test_get_container_create_options_with_name_option(self):
|
|
service = Service(
|
|
'foo',
|
|
image='foo',
|
|
client=self.mock_client,
|
|
container_name='foo1')
|
|
name = 'the_new_name'
|
|
opts = service._get_container_create_options(
|
|
{'name': name},
|
|
1,
|
|
one_off=True)
|
|
self.assertEqual(opts['name'], name)
|
|
|
|
def test_get_container_create_options_does_not_mutate_options(self):
|
|
labels = {'thing': 'real'}
|
|
environment = {'also': 'real'}
|
|
service = Service(
|
|
'foo',
|
|
image='foo',
|
|
labels=dict(labels),
|
|
client=self.mock_client,
|
|
environment=dict(environment),
|
|
)
|
|
self.mock_client.inspect_image.return_value = {'Id': 'abcd'}
|
|
prev_container = mock.Mock(
|
|
id='ababab',
|
|
image_config={'ContainerConfig': {}})
|
|
|
|
opts = service._get_container_create_options(
|
|
{},
|
|
1,
|
|
previous_container=prev_container)
|
|
|
|
self.assertEqual(service.options['labels'], labels)
|
|
self.assertEqual(service.options['environment'], environment)
|
|
|
|
self.assertEqual(
|
|
opts['labels'][LABEL_CONFIG_HASH],
|
|
'3c85881a8903b9d73a06c41860c8be08acce1494ab4cf8408375966dccd714de')
|
|
self.assertEqual(
|
|
opts['environment'],
|
|
{
|
|
'affinity:container': '=ababab',
|
|
'also': 'real',
|
|
}
|
|
)
|
|
|
|
def test_get_container_not_found(self):
|
|
self.mock_client.containers.return_value = []
|
|
service = Service('foo', client=self.mock_client, image='foo')
|
|
|
|
self.assertRaises(ValueError, service.get_container)
|
|
|
|
@mock.patch('compose.service.Container', autospec=True)
|
|
def test_get_container(self, mock_container_class):
|
|
container_dict = dict(Name='default_foo_2')
|
|
self.mock_client.containers.return_value = [container_dict]
|
|
service = Service('foo', image='foo', client=self.mock_client)
|
|
|
|
container = service.get_container(number=2)
|
|
self.assertEqual(container, mock_container_class.from_ps.return_value)
|
|
mock_container_class.from_ps.assert_called_once_with(
|
|
self.mock_client, container_dict)
|
|
|
|
@mock.patch('compose.service.log', autospec=True)
|
|
def test_pull_image(self, mock_log):
|
|
service = Service('foo', client=self.mock_client, image='someimage:sometag')
|
|
service.pull()
|
|
self.mock_client.pull.assert_called_once_with(
|
|
'someimage',
|
|
tag='sometag',
|
|
stream=True)
|
|
mock_log.info.assert_called_once_with('Pulling foo (someimage:sometag)...')
|
|
|
|
def test_pull_image_no_tag(self):
|
|
service = Service('foo', client=self.mock_client, image='ababab')
|
|
service.pull()
|
|
self.mock_client.pull.assert_called_once_with(
|
|
'ababab',
|
|
tag='latest',
|
|
stream=True)
|
|
|
|
@mock.patch('compose.service.Container', autospec=True)
|
|
def test_recreate_container(self, _):
|
|
mock_container = mock.create_autospec(Container)
|
|
service = Service('foo', client=self.mock_client, image='someimage')
|
|
service.image = lambda: {'Id': 'abc123'}
|
|
new_container = service.recreate_container(mock_container)
|
|
|
|
mock_container.stop.assert_called_once_with(timeout=10)
|
|
self.mock_client.rename.assert_called_once_with(
|
|
mock_container.id,
|
|
'%s_%s' % (mock_container.short_id, mock_container.name))
|
|
|
|
new_container.start.assert_called_once_with()
|
|
mock_container.remove.assert_called_once_with()
|
|
|
|
@mock.patch('compose.service.Container', autospec=True)
|
|
def test_recreate_container_with_timeout(self, _):
|
|
mock_container = mock.create_autospec(Container)
|
|
self.mock_client.inspect_image.return_value = {'Id': 'abc123'}
|
|
service = Service('foo', client=self.mock_client, image='someimage')
|
|
service.recreate_container(mock_container, timeout=1)
|
|
|
|
mock_container.stop.assert_called_once_with(timeout=1)
|
|
|
|
def test_parse_repository_tag(self):
|
|
self.assertEqual(parse_repository_tag("root"), ("root", ""))
|
|
self.assertEqual(parse_repository_tag("root:tag"), ("root", "tag"))
|
|
self.assertEqual(parse_repository_tag("user/repo"), ("user/repo", ""))
|
|
self.assertEqual(parse_repository_tag("user/repo:tag"), ("user/repo", "tag"))
|
|
self.assertEqual(parse_repository_tag("url:5000/repo"), ("url:5000/repo", ""))
|
|
self.assertEqual(parse_repository_tag("url:5000/repo:tag"), ("url:5000/repo", "tag"))
|
|
|
|
@mock.patch('compose.service.Container', autospec=True)
|
|
def test_create_container_latest_is_used_when_no_tag_specified(self, mock_container):
|
|
service = Service('foo', client=self.mock_client, image='someimage')
|
|
images = []
|
|
|
|
def pull(repo, tag=None, **kwargs):
|
|
self.assertEqual('someimage', repo)
|
|
self.assertEqual('latest', tag)
|
|
images.append({'Id': 'abc123'})
|
|
return []
|
|
|
|
service.image = lambda *args, **kwargs: mock_get_image(images)
|
|
self.mock_client.pull = pull
|
|
|
|
service.create_container()
|
|
self.assertEqual(1, len(images))
|
|
|
|
def test_create_container_with_build(self):
|
|
service = Service('foo', client=self.mock_client, build='.')
|
|
|
|
images = []
|
|
service.image = lambda *args, **kwargs: mock_get_image(images)
|
|
service.build = lambda: images.append({'Id': 'abc123'})
|
|
|
|
service.create_container(do_build=True)
|
|
self.assertEqual(1, len(images))
|
|
|
|
def test_create_container_no_build(self):
|
|
service = Service('foo', client=self.mock_client, build='.')
|
|
service.image = lambda: {'Id': 'abc123'}
|
|
|
|
service.create_container(do_build=False)
|
|
self.assertFalse(self.mock_client.build.called)
|
|
|
|
def test_create_container_no_build_but_needs_build(self):
|
|
service = Service('foo', client=self.mock_client, build='.')
|
|
service.image = lambda *args, **kwargs: mock_get_image([])
|
|
|
|
with self.assertRaises(NeedsBuildError):
|
|
service.create_container(do_build=False)
|
|
|
|
def test_build_does_not_pull(self):
|
|
self.mock_client.build.return_value = [
|
|
'{"stream": "Successfully built 12345"}',
|
|
]
|
|
|
|
service = Service('foo', client=self.mock_client, build='.')
|
|
service.build()
|
|
|
|
self.assertEqual(self.mock_client.build.call_count, 1)
|
|
self.assertFalse(self.mock_client.build.call_args[1]['pull'])
|
|
|
|
def test_config_dict(self):
|
|
self.mock_client.inspect_image.return_value = {'Id': 'abcd'}
|
|
service = Service(
|
|
'foo',
|
|
image='example.com/foo',
|
|
client=self.mock_client,
|
|
net=ServiceNet(Service('other')),
|
|
links=[(Service('one'), 'one')],
|
|
volumes_from=[Service('two')])
|
|
|
|
config_dict = service.config_dict()
|
|
expected = {
|
|
'image_id': 'abcd',
|
|
'options': {'image': 'example.com/foo'},
|
|
'links': [('one', 'one')],
|
|
'net': 'other',
|
|
'volumes_from': ['two'],
|
|
}
|
|
self.assertEqual(config_dict, expected)
|
|
|
|
def test_config_dict_with_net_from_container(self):
|
|
self.mock_client.inspect_image.return_value = {'Id': 'abcd'}
|
|
container = Container(
|
|
self.mock_client,
|
|
{'Id': 'aaabbb', 'Name': '/foo_1'})
|
|
service = Service(
|
|
'foo',
|
|
image='example.com/foo',
|
|
client=self.mock_client,
|
|
net=container)
|
|
|
|
config_dict = service.config_dict()
|
|
expected = {
|
|
'image_id': 'abcd',
|
|
'options': {'image': 'example.com/foo'},
|
|
'links': [],
|
|
'net': 'aaabbb',
|
|
'volumes_from': [],
|
|
}
|
|
self.assertEqual(config_dict, expected)
|
|
|
|
|
|
class NetTestCase(unittest.TestCase):
|
|
|
|
def test_net(self):
|
|
net = Net('host')
|
|
self.assertEqual(net.id, 'host')
|
|
self.assertEqual(net.mode, 'host')
|
|
self.assertEqual(net.service_name, None)
|
|
|
|
def test_net_container(self):
|
|
container_id = 'abcd'
|
|
net = ContainerNet(Container(None, {'Id': container_id}))
|
|
self.assertEqual(net.id, container_id)
|
|
self.assertEqual(net.mode, 'container:' + container_id)
|
|
self.assertEqual(net.service_name, None)
|
|
|
|
def test_net_service(self):
|
|
container_id = 'bbbb'
|
|
service_name = 'web'
|
|
mock_client = mock.create_autospec(docker.Client)
|
|
mock_client.containers.return_value = [
|
|
{'Id': container_id, 'Name': container_id, 'Image': 'abcd'},
|
|
]
|
|
|
|
service = Service(name=service_name, client=mock_client)
|
|
net = ServiceNet(service)
|
|
|
|
self.assertEqual(net.id, service_name)
|
|
self.assertEqual(net.mode, 'container:' + container_id)
|
|
self.assertEqual(net.service_name, service_name)
|
|
|
|
def test_net_service_no_containers(self):
|
|
service_name = 'web'
|
|
mock_client = mock.create_autospec(docker.Client)
|
|
mock_client.containers.return_value = []
|
|
|
|
service = Service(name=service_name, client=mock_client)
|
|
net = ServiceNet(service)
|
|
|
|
self.assertEqual(net.id, service_name)
|
|
self.assertEqual(net.mode, None)
|
|
self.assertEqual(net.service_name, service_name)
|
|
|
|
|
|
def mock_get_image(images):
|
|
if images:
|
|
return images[0]
|
|
else:
|
|
raise NoSuchImageError()
|
|
|
|
|
|
class ServiceVolumesTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.mock_client = mock.create_autospec(docker.Client)
|
|
|
|
def test_parse_volume_spec_only_one_path(self):
|
|
spec = parse_volume_spec('/the/volume')
|
|
self.assertEqual(spec, (None, '/the/volume', 'rw'))
|
|
|
|
def test_parse_volume_spec_internal_and_external(self):
|
|
spec = parse_volume_spec('external:interval')
|
|
self.assertEqual(spec, ('external', 'interval', 'rw'))
|
|
|
|
def test_parse_volume_spec_with_mode(self):
|
|
spec = parse_volume_spec('external:interval:ro')
|
|
self.assertEqual(spec, ('external', 'interval', 'ro'))
|
|
|
|
spec = parse_volume_spec('external:interval:z')
|
|
self.assertEqual(spec, ('external', 'interval', 'z'))
|
|
|
|
def test_parse_volume_spec_too_many_parts(self):
|
|
with self.assertRaises(ConfigError):
|
|
parse_volume_spec('one:two:three:four')
|
|
|
|
def test_build_volume_binding(self):
|
|
binding = build_volume_binding(parse_volume_spec('/outside:/inside'))
|
|
self.assertEqual(binding, ('/inside', '/outside:/inside:rw'))
|
|
|
|
def test_get_container_data_volumes(self):
|
|
options = [
|
|
'/host/volume:/host/volume:ro',
|
|
'/new/volume',
|
|
'/existing/volume',
|
|
]
|
|
|
|
self.mock_client.inspect_image.return_value = {
|
|
'ContainerConfig': {
|
|
'Volumes': {
|
|
'/mnt/image/data': {},
|
|
}
|
|
}
|
|
}
|
|
container = Container(self.mock_client, {
|
|
'Image': 'ababab',
|
|
'Volumes': {
|
|
'/host/volume': '/host/volume',
|
|
'/existing/volume': '/var/lib/docker/aaaaaaaa',
|
|
'/removed/volume': '/var/lib/docker/bbbbbbbb',
|
|
'/mnt/image/data': '/var/lib/docker/cccccccc',
|
|
},
|
|
}, has_been_inspected=True)
|
|
|
|
expected = {
|
|
'/existing/volume': '/var/lib/docker/aaaaaaaa:/existing/volume:rw',
|
|
'/mnt/image/data': '/var/lib/docker/cccccccc:/mnt/image/data:rw',
|
|
}
|
|
|
|
binds = get_container_data_volumes(container, options)
|
|
self.assertEqual(binds, expected)
|
|
|
|
def test_merge_volume_bindings(self):
|
|
options = [
|
|
'/host/volume:/host/volume:ro',
|
|
'/host/rw/volume:/host/rw/volume',
|
|
'/new/volume',
|
|
'/existing/volume',
|
|
]
|
|
|
|
self.mock_client.inspect_image.return_value = {
|
|
'ContainerConfig': {'Volumes': {}}
|
|
}
|
|
|
|
intermediate_container = Container(self.mock_client, {
|
|
'Image': 'ababab',
|
|
'Volumes': {'/existing/volume': '/var/lib/docker/aaaaaaaa'},
|
|
}, has_been_inspected=True)
|
|
|
|
expected = [
|
|
'/host/volume:/host/volume:ro',
|
|
'/host/rw/volume:/host/rw/volume:rw',
|
|
'/var/lib/docker/aaaaaaaa:/existing/volume:rw',
|
|
]
|
|
|
|
binds = merge_volume_bindings(options, intermediate_container)
|
|
self.assertEqual(set(binds), set(expected))
|
|
|
|
def test_mount_same_host_path_to_two_volumes(self):
|
|
service = Service(
|
|
'web',
|
|
image='busybox',
|
|
volumes=[
|
|
'/host/path:/data1',
|
|
'/host/path:/data2',
|
|
],
|
|
client=self.mock_client,
|
|
)
|
|
|
|
self.mock_client.inspect_image.return_value = {
|
|
'Id': 'ababab',
|
|
'ContainerConfig': {
|
|
'Volumes': {}
|
|
}
|
|
}
|
|
|
|
create_options = service._get_container_create_options(
|
|
override_options={},
|
|
number=1,
|
|
)
|
|
|
|
self.assertEqual(
|
|
set(create_options['host_config']['Binds']),
|
|
set([
|
|
'/host/path:/data1:rw',
|
|
'/host/path:/data2:rw',
|
|
]),
|
|
)
|
|
|
|
def test_different_host_path_in_container_json(self):
|
|
service = Service(
|
|
'web',
|
|
image='busybox',
|
|
volumes=['/host/path:/data'],
|
|
client=self.mock_client,
|
|
)
|
|
|
|
self.mock_client.inspect_image.return_value = {
|
|
'Id': 'ababab',
|
|
'ContainerConfig': {
|
|
'Volumes': {
|
|
'/data': {},
|
|
}
|
|
}
|
|
}
|
|
|
|
self.mock_client.inspect_container.return_value = {
|
|
'Id': '123123123',
|
|
'Image': 'ababab',
|
|
'Volumes': {
|
|
'/data': '/mnt/sda1/host/path',
|
|
},
|
|
}
|
|
|
|
create_options = service._get_container_create_options(
|
|
override_options={},
|
|
number=1,
|
|
previous_container=Container(self.mock_client, {'Id': '123123123'}),
|
|
)
|
|
|
|
self.assertEqual(
|
|
create_options['host_config']['Binds'],
|
|
['/mnt/sda1/host/path:/data:rw'],
|
|
)
|
|
|
|
def test_create_with_special_volume_mode(self):
|
|
self.mock_client.inspect_image.return_value = {'Id': 'imageid'}
|
|
|
|
create_calls = []
|
|
|
|
def create_container(*args, **kwargs):
|
|
create_calls.append((args, kwargs))
|
|
return {'Id': 'containerid'}
|
|
|
|
self.mock_client.create_container = create_container
|
|
|
|
volumes = ['/tmp:/foo:z']
|
|
|
|
Service(
|
|
'web',
|
|
client=self.mock_client,
|
|
image='busybox',
|
|
volumes=volumes,
|
|
).create_container()
|
|
|
|
self.assertEqual(len(create_calls), 1)
|
|
self.assertEqual(create_calls[0][1]['host_config']['Binds'], volumes)
|