openstack/zaqar
Code Issues Proposed changes

Merge "Escape IPv6 address in square brackets"

Browse Source
This commit is contained in:
Jenkins 2017-10-13 03:34:40 +00:00 committed by Gerrit Code Review
commit dda183920d
2 changed files with 24 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
zaqar/tests/unit/transport/websocket/test_protocol.py
View File

@ -19,6 +19,7 @@ import ddt
import mock import mock
from oslo_utils import uuidutils from oslo_utils import uuidutils
import zaqar
from zaqar.tests.unit.transport.websocket import base from zaqar.tests.unit.transport.websocket import base
from zaqar.tests.unit.transport.websocket import utils as test_utils from zaqar.tests.unit.transport.websocket import utils as test_utils
@ -92,3 +93,15 @@ class TestMessagingProtocol(base.TestBase):
self.protocol.onMessage(req, in_binary) self.protocol.onMessage(req, in_binary)
resp = loads(send_mock.call_args[0][0]) resp = loads(send_mock.call_args[0][0])
self.assertEqual(200, resp['headers']['status']) self.assertEqual(200, resp['headers']['status'])
@mock.patch.object(zaqar.transport.websocket.factory, 'ProtocolFactory')
def test_ipv6_escaped(self, mock_pf):
delattr(self.transport, '_lazy_factory')
self.transport.factory()
self.assertEqual('ws://127.0.0.1:9000', mock_pf.mock_calls[0][1][0])
mock_pf.reset_mock()
with mock.patch.object(self.transport._ws_conf, 'bind', "1::4"):
delattr(self.transport, '_lazy_factory')
self.transport.factory()
self.assertEqual('ws://[1::4]:9000', mock_pf.mock_calls[0][1][0])

12
zaqar/transport/websocket/driver.py
View File

@ -17,6 +17,7 @@ import socket
from oslo_config import cfg from oslo_config import cfg
from oslo_log import log as logging from oslo_log import log as logging
from oslo_utils import netutils
try: try:
import asyncio import asyncio
@ -59,6 +60,14 @@ def _config_options():
return [(_WS_GROUP, _WS_OPTIONS)] return [(_WS_GROUP, _WS_OPTIONS)]
# TODO(derekh): use escape_ipv6 from oslo.utils once available
def _escape_ipv6(address):
"""Escape an IP address in square brackets if IPv6"""
if netutils.is_valid_ipv6(address):
return "[%s]" % address
return address
class Driver(base.DriverBase): class Driver(base.DriverBase):
def __init__(self, conf, api, cache): def __init__(self, conf, api, cache):
@ -78,7 +87,8 @@ class Driver(base.DriverBase):
@decorators.lazy_property(write=False) @decorators.lazy_property(write=False)
def factory(self): def factory(self):
uri = 'ws://' + self._ws_conf.bind + ':' + str(self._ws_conf.port) uri = 'ws://' + _escape_ipv6(self._ws_conf.bind) + ':' + \
str(self._ws_conf.port)
return factory.ProtocolFactory( return factory.ProtocolFactory(
uri, uri,
handler=self._api, handler=self._api,