Use split_port for ports format check

Rather than implement the logic a second time, use docker-py
split_port function to test if the ports is valid.

Signed-off-by: Mazz Mosley <mazz@houseofmnowster.com>
This commit is contained in:
Mazz Mosley 2015-08-07 15:26:26 +01:00
parent 2e428f94ca
commit df74b131ff
3 changed files with 12 additions and 29 deletions

View File

@ -75,15 +75,10 @@
"pid": {"type": "string"}, "pid": {"type": "string"},
"ports": { "ports": {
"oneOf": [ "type": "array",
{"type": "string", "format": "ports"}, "items": {"type": "string"},
{ "uniqueItems": true,
"type": "array", "format": "ports"
"items": {"type": "string"},
"uniqueItems": true,
"format": "ports"
}
]
}, },
"privileged": {"type": "string"}, "privileged": {"type": "string"},

View File

@ -1,5 +1,6 @@
import os import os
from docker.utils.ports import split_port
import json import json
from jsonschema import Draft4Validator, FormatChecker, ValidationError from jsonschema import Draft4Validator, FormatChecker, ValidationError
@ -26,26 +27,13 @@ DOCKER_CONFIG_HINTS = {
VALID_NAME_CHARS = '[a-zA-Z0-9\._\-]' VALID_NAME_CHARS = '[a-zA-Z0-9\._\-]'
@FormatChecker.cls_checks(format="ports", raises=ValidationError("Ports is incorrectly formatted.")) @FormatChecker.cls_checks(format="ports", raises=ValidationError("Invalid port formatting, it should be '[[remote_ip:]remote_port:]port[/protocol]'"))
def format_ports(instance): def format_ports(instance):
def _is_valid(port): try:
if ':' in port or '/' in port: split_port(instance)
return True except ValueError:
try:
int(port)
return True
except ValueError:
return False
return False return False
return True
if isinstance(instance, list):
for port in instance:
if not _is_valid(port):
return False
return True
elif isinstance(instance, str):
return _is_valid(instance)
return False
def get_unsupported_config_msg(service_name, error_key): def get_unsupported_config_msg(service_name, error_key):

View File

@ -76,7 +76,7 @@ class ConfigTest(unittest.TestCase):
def test_config_invalid_ports_format_validation(self): def test_config_invalid_ports_format_validation(self):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
for invalid_ports in [{"1": "8000"}, "whatport"]: for invalid_ports in [{"1": "8000"}, "whatport", "625", "8000:8050"]:
config.load( config.load(
config.ConfigDetails( config.ConfigDetails(
{'web': {'image': 'busybox', 'ports': invalid_ports}}, {'web': {'image': 'busybox', 'ports': invalid_ports}},
@ -86,7 +86,7 @@ class ConfigTest(unittest.TestCase):
) )
def test_config_valid_ports_format_validation(self): def test_config_valid_ports_format_validation(self):
valid_ports = [["8000", "9000"], "625", "8000:8050", ["8000/8050"]] valid_ports = [["8000", "9000"], ["8000/8050"], ["8000"]]
for ports in valid_ports: for ports in valid_ports:
config.load( config.load(
config.ConfigDetails( config.ConfigDetails(