slapos.slap: parse ipv6 and adds bracket s if missing.

Needed for requests, that now NEEDS brackets for ipv6.
......@@ -41,6 +41,7 @@ import re
import urlparse
from util import xml2dict
import netaddr
from xml.sax import saxutils
import zope.interface
from interface import slap as interface
......@@ -599,10 +600,31 @@ class ComputerPartition(SlapRequester):
)
return xml_marshaller.loads(xml)
def _addIpv6Brackets(url):
# if master_url contains an ipv6 without bracket, add it
# Note that this is mostly to limit specific issues with
# backward compatiblity, not to ensure generic detection.
api_scheme, api_netloc, api_path, api_query, api_fragment = urlparse.urlsplit(url)
try:
ip = netaddr.IPAddress(api_netloc)
port = None
except netaddr.AddrFormatError:
try:
ip = netaddr.IPAddress(':'.join(api_netloc.split(':')[:-1]))
port = api_netloc.split(':')[-1]
except netaddr.AddrFormatError:
ip = port = None
if ip and ip.version == 6:
api_netloc = '[%s]' % ip
if port:
api_netloc = '%s:%s' % (api_netloc, port)
url = urlparse.urlunsplit((api_scheme, api_netloc, api_path, api_query, api_fragment))
return url
class ConnectionHelper:
def __init__(self, master_url, key_file=None,
cert_file=None, master_ca_file=None, timeout=None):
master_url = _addIpv6Brackets(master_url)
if master_url.endswith('/'):
self.slapgrid_uri = master_url
else:
......
......@@ -78,6 +78,66 @@ class TestSlap(SlapMixin):
slap_instance.initializeConnection(self.server_url)
self.assertEquals(slap_instance._connection_helper.slapgrid_uri, self.server_url)
def test_slap_initialisation_ipv6_and_port(self):
"""
Asserts that slap correctly understand master_url containing
ipv6 and adds brackets if not there.
"""
slap_instance = slapos.slap.slap()
slap_instance.initializeConnection("http://1234:1234:1234:1234:1:1:1:1:5000/foo/")
self.assertEqual(
slap_instance._connection_helper.slapgrid_uri,
"http://[1234:1234:1234:1234:1:1:1:1]:5000/foo/"
)
def test_slap_initialisation_ipv6_without_port(self):
"""
Asserts that slap correctly understand master_url containing
ipv6 and adds brackets if not there.
"""
slap_instance = slapos.slap.slap()
slap_instance.initializeConnection("http://1234:1234:1234:1234:1:1:1:1/foo/")
self.assertEqual(
slap_instance._connection_helper.slapgrid_uri,
"http://[1234:1234:1234:1234:1:1:1:1]/foo/"
)
def test_slap_initialisation_ipv6_with_bracket(self):
"""
Asserts that slap correctly understand master_url containing
ipv6 and adds brackets if not there.
"""
slap_instance = slapos.slap.slap()
slap_instance.initializeConnection("http://[1234:1234:1234:1234:1:1:1:1]:5000/foo/")
self.assertEqual(
slap_instance._connection_helper.slapgrid_uri,
"http://[1234:1234:1234:1234:1:1:1:1]:5000/foo/"
)
def test_slap_initialisation_ipv4(self):
"""
Asserts that slap correctly understand master_url containing
ipv6 and adds brackets if not there.
"""
slap_instance = slapos.slap.slap()
slap_instance.initializeConnection("http://127.0.0.1:5000/foo/")
self.assertEqual(
slap_instance._connection_helper.slapgrid_uri,
"http://127.0.0.1:5000/foo/"
)
def test_slap_initialisation_hostname(self):
"""
Asserts that slap correctly understand master_url containing
ipv6 and adds brackets if not there.
"""
slap_instance = slapos.slap.slap()
slap_instance.initializeConnection("http://foo.com:5000/foo/")
self.assertEqual(
slap_instance._connection_helper.slapgrid_uri,
"http://foo.com:5000/foo/"
)
def test_registerComputer_with_new_guid(self):
"""
Asserts that calling slap.registerComputer with new guid returns
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment