using port binding from docker-py that includes port ranges

Author: yuval-k

compose/service.py

@@ -9,6 +9,7 @@
 import six
 from docker.errors import APIError
 from docker.utils import create_host_config, LogConfig
+from docker.utils.ports import build_port_bindings
 
 from .config import DOCKER_CONFIG_KEYS, merge_environment
 from .container import Container, get_container_name
@@ -654,34 +655,6 @@ def build_volume_binding(volume_spec):
     return volume_spec.external, internal
 
 
-def build_port_bindings(ports):
-    port_bindings = {}
-    for port in ports:
-        internal_port, external = split_port(port)
-        if internal_port in port_bindings:
-            port_bindings[internal_port].append(external)
-        else:
-            port_bindings[internal_port] = [external]
-    return port_bindings
-
-
-def split_port(port):
-    parts = str(port).split(':')
-    if not 1 <= len(parts) <= 3:
-        raise ConfigError('Invalid port "%s", should be '
-                          '[[remote_ip:]remote_port:]port[/protocol]' % port)
-
-    if len(parts) == 1:
-        internal_port, = parts
-        return internal_port, None
-    if len(parts) == 2:
-        external_port, internal_port = parts
-        return internal_port, external_port
-
-    external_ip, external_port, internal_port = parts
-    return internal_port, (external_ip, external_port or None)
-
-
 def build_extra_hosts(extra_hosts_config):
     if not extra_hosts_config:
         return {}

tests/unit/service_test.py

@@ -12,14 +12,12 @@
 from compose.service import (
     APIError,
     ConfigError,
-    build_port_bindings,
     build_volume_binding,
     get_container_data_volumes,
     get_container_name,
     merge_volume_bindings,
     parse_repository_tag,
     parse_volume_spec,
-    split_port,
 )
 
 
@@ -111,48 +109,6 @@ def test_get_volumes_from_service_no_container(self):
         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 = []