tyler.durden/compose
1
0
Fork 0
mirror of https://github.com/docker/compose.git synced 2025-07-21 04:34:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Some additional validation for container ports, and a couple extra test cases.

Browse Source 
Signed-off-by: Daniel Nephin <dnephin@gmail.com>
This commit is contained in:
Daniel Nephin 2014-08-25 22:17:33 -04:00
parent 07fa169fd2
commit 24044fa704
2 changed files with 29 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
fig/service.py
View File

@ -245,7 +245,7 @@ class Service(object):
def start_or_create_containers(self): def start_or_create_containers(self):
containers = self.containers(stopped=True) containers = self.containers(stopped=True)
if len(containers) == 0: if not containers:
log.info("Creating %s..." % self.next_container_name()) log.info("Creating %s..." % self.next_container_name())
new_container = self.create_container() new_container = self.create_container()
return [self.start_container(new_container)] return [self.start_container(new_container)]
@ -451,21 +451,21 @@ def build_volume_binding(volume_spec):
def split_port(port): def split_port(port):
port = str(port) parts = str(port).split(':')
external_ip = None if not 1 <= len(parts) <= 3:
if ':' in port: raise ConfigError('Invalid port "%s", should be '
external_port, internal_port = port.rsplit(':', 1) '[[remote_ip:]remote_port:]port[/protocol]' % port)
if ':' in external_port:
external_ip, external_port = external_port.split(':', 1) if len(parts) == 1:
else: internal_port, = parts
external_port, internal_port = (None, port) return internal_port, None
if external_ip: if len(parts) == 2:
if external_port: external_port, internal_port = parts
external_port = (external_ip, external_port)
else:
external_port = (external_ip,)
return internal_port, external_port return internal_port, external_port
external_ip, external_port, internal_port = parts
return internal_port, (external_ip, external_port or None)
def split_env(env): def split_env(env):
if '=' in env: if '=' in env:

16
tests/unit/service_test.py
View File

@ -38,23 +38,35 @@ class ServiceTest(unittest.TestCase):
self.assertRaises(ConfigError, lambda: Service(name='foo', port=['8000'])) self.assertRaises(ConfigError, lambda: Service(name='foo', port=['8000']))
Service(name='foo', ports=['8000']) Service(name='foo', ports=['8000'])
def test_split_port(self): def test_split_port_with_host_ip(self):
internal_port, external_port = split_port("127.0.0.1:1000:2000") internal_port, external_port = split_port("127.0.0.1:1000:2000")
self.assertEqual(internal_port, "2000") self.assertEqual(internal_port, "2000")
self.assertEqual(external_port, ("127.0.0.1", "1000")) 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") internal_port, external_port = split_port("127.0.0.1:1000:2000/udp")
self.assertEqual(internal_port, "2000/udp") self.assertEqual(internal_port, "2000/udp")
self.assertEqual(external_port, ("127.0.0.1", "1000")) 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") internal_port, external_port = split_port("127.0.0.1::2000")
self.assertEqual(internal_port, "2000") self.assertEqual(internal_port, "2000")
self.assertEqual(external_port, ("127.0.0.1",)) 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") internal_port, external_port = split_port("1000:2000")
self.assertEqual(internal_port, "2000") self.assertEqual(internal_port, "2000")
self.assertEqual(external_port, "1000") 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_split_domainname_none(self): def test_split_domainname_none(self):
service = Service('foo', service = Service('foo',
hostname = 'name', hostname = 'name',