docker
/
docker-py
mirror of https://github.com/docker/docker-py.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adding port utils to split port ranges 

Signed-off-by: Yuval Kohavi <yuval.kohavi@gmail.com>
This commit is contained in:
Yuval Kohavi 2015-04-03 17:23:50 +03:00
parent 2f6579bf59
commit 278e801b89
3 changed files with 185 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
docker/utils/ports/__init__.py Normal file
View File

@ -0,0 +1,6 @@
from .ports import (
to_port_range,
split_port,
add_port_mapping,
build_port_bindings
) # flake8: noqa

84
docker/utils/ports/ports.py Normal file
View File

@ -0,0 +1,84 @@
def add_port_mapping(port_bindings, internal_port, external):
if internal_port in port_bindings:
port_bindings[internal_port].append(external)
else:
port_bindings[internal_port] = [external]
def add_port(port_bindings, internal_port_range, external_range):
if external_range is None:
for internal_port in internal_port_range:
add_port_mapping(port_bindings, internal_port, None)
else:
ports = zip(internal_port_range, external_range)
for internal_port, external_port in ports:
add_port_mapping(port_bindings, internal_port, external_port)
def build_port_bindings(ports):
port_bindings = {}
for port in ports:
internal_port_range, external_range = split_port(port)
add_port(port_bindings, internal_port_range, external_range)
return port_bindings
def to_port_range(port):
if not port:
return None
protocol = ""
if "/" in port:
parts = port.split("/")
if len(parts) != 2:
raise ValueError('Invalid port "%s", should be '
'[[remote_ip:]remote_port[-remote_port]:]'
'port[/protocol]' % port)
port, protocol = parts
protocol = "/" + protocol
parts = str(port).split('-')
if len(parts) == 1:
return ["%s%s" % (port, protocol)]
if len(parts) == 2:
full_port_range = range(int(parts[0]), int(parts[1]) + 1)
return ["%s%s" % (p, protocol) for p in full_port_range]
raise ValueError('Invalid port range "%s", should be '
'port or startport-endport' % port)
def split_port(port):
parts = str(port).split(':')
if not 1 <= len(parts) <= 3:
raise ValueError('Invalid port "%s", should be '
'[[remote_ip:]remote_port:]port[/protocol]' % port)
if len(parts) == 1:
internal_port, = parts
return to_port_range(internal_port), None
if len(parts) == 2:
external_port, internal_port = parts
internal_range = to_port_range(internal_port)
external_range = to_port_range(external_port)
if len(internal_range) != len(external_range):
raise ValueError('Port ranges don\'t match in length')
return internal_range, external_range
external_ip, external_port, internal_port = parts
internal_range = to_port_range(internal_port)
external_range = to_port_range(external_port)
if not external_range:
external_range = [None] * len(internal_range)
if len(internal_range) != len(external_range):
raise ValueError('Port ranges don\'t match in length')
return internal_range, [(external_ip, ex_port or None)
for ex_port in external_range]

95
tests/utils_test.py
View File

@ -8,6 +8,7 @@ from docker.utils import (
parse_repository_tag, parse_host, convert_filters, kwargs_from_env,
create_host_config
)
from docker.utils.ports import build_port_bindings, split_port
from docker.auth import resolve_authconfig
@ -165,6 +166,100 @@ class UtilsTest(unittest.TestCase):
resolve_authconfig(auth_config, 'does.not.exist') is None
)
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_range_with_host_ip_no_port(self):
internal_port, external_port = split_port("127.0.0.1::2000-2001")
self.assertEqual(internal_port, ["2000", "2001"])
self.assertEqual(external_port,
[("127.0.0.1", None), ("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_range_with_host_port(self):
internal_port, external_port = split_port("1000-1001:2000-2001")
self.assertEqual(internal_port, ["2000", "2001"])
self.assertEqual(external_port, ["1000", "1001"])
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_range_no_host_port(self):
internal_port, external_port = split_port("2000-2001")
self.assertEqual(internal_port, ["2000", "2001"])
self.assertEqual(external_port, None)
def test_split_port_range_with_protocol(self):
internal_port, external_port = split_port(
"127.0.0.1:1000-1001:2000-2001/udp")
self.assertEqual(internal_port, ["2000/udp", "2001/udp"])
self.assertEqual(external_port,
[("127.0.0.1", "1000"), ("127.0.0.1", "1001")])
def test_split_port_invalid(self):
with self.assertRaises(ValueError):
split_port("0.0.0.0:1000:2000:tcp")
def test_non_matching_length_port_ranges(self):
with self.assertRaises(ValueError):
split_port("0.0.0.0:1000-1010:2000-2002/tcp")
def test_port_and_range_invalid(self):
with self.assertRaises(ValueError):
split_port("0.0.0.0:1000:2000-2002/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_build_port_bindings_with_port_range(self):
port_bindings = build_port_bindings(["127.0.0.1:1000-1001:1000-1001"])
self.assertEqual(port_bindings["1000"], [("127.0.0.1", "1000")])
self.assertEqual(port_bindings["1001"], [("127.0.0.1", "1001")])
def test_build_port_bindings_with_matching_internal_port_ranges(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000-1001:1000-1001", "127.0.0.1:2000-2001:1000-1001"])
self.assertEqual(port_bindings["1000"],
[("127.0.0.1", "1000"), ("127.0.0.1", "2000")])
self.assertEqual(port_bindings["1001"],
[("127.0.0.1", "1001"), ("127.0.0.1", "2001")])
def test_build_port_bindings_with_nonmatching_internal_port_ranges(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")])
if __name__ == '__main__':
unittest.main()