Support for ipv6 in wsgi.Service

Enable wsgi.Service to listen on ipv6 address by checking
if the host specified is ipv6. Based on that set the
appropriate family in the eventlet.listen api

More tests for just the ipv6, ipv6 with app, ipv6+ssl with app
to make sure everything is working fine

Change-Id: I2772905128bdbc69dd0fafe4ced848f5c477d7c8

1 files changed, 63 insertions, 1 deletions

diff --git a/tests/unit/test_wsgi.py b/tests/unit/test_wsgi.py
index e4f975e..2dd1035 100644
--- a/tests/unit/test_wsgi.py
+++ b/tests/unit/test_wsgi.py
@@ -18,6 +18,7 @@
 import mock
 import os
 import routes
+import socket
 import ssl
 import urllib2
 import webob
@@ -480,6 +481,13 @@ class WSGIServerTest(utils.BaseTestCase):
         self.assertNotEqual(0, server.port)
         server.stop()
 
+    def test_start_random_port_with_ipv6(self):
+        server = wsgi.Service('test_random_port', 0, host="::1")
+        server.start()
+        self.assertEqual("::1", server.host)
+        self.assertNotEqual(0, server.port)
+        server.stop()
+
     def test_app(self):
         greetings = 'Hello, World!!!'
 
@@ -517,8 +525,39 @@ class WSGIServerTest(utils.BaseTestCase):
 
         server.stop()
 
-    def test_app_using_router_ssl(self):
+    def test_ipv6_listen_called_with_scope(self):
+        server = wsgi.Service("test_app",
+                              1234,
+                              host="fe80::204:acff:fe96:da87%eth0")
+
+        with mock.patch.object(wsgi.eventlet, 'listen') as mock_listen:
+            with mock.patch.object(socket, 'getaddrinfo') as mock_get_addr:
+                mock_get_addr.return_value = [
+                    (socket.AF_INET6,
+                     socket.SOCK_STREAM,
+                     socket.IPPROTO_TCP,
+                     '',
+                     ('fe80::204:acff:fe96:da87%eth0', 1234, 0, 2))
+                ]
+                server.start()
+
+                mock_get_addr.assert_called_once_with(
+                    "fe80::204:acff:fe96:da87%eth0",
+                    1234,
+                    socket.AF_UNSPEC,
+                    socket.SOCK_STREAM
+                )
+
+                mock_listen.assert_called_once_with(
+                    ('fe80::204:acff:fe96:da87%eth0', 1234, 0, 2),
+                    backlog=4096,
+                    family=socket.AF_INET6
+                )
+
+
+class WSGIServerWithSSLTest(utils.BaseTestCase):
 
+    def test_app_using_router_ssl(self):
         self.config(cert_file=os.path.join(TEST_VAR_DIR, 'certificate.crt'),
                     group="ssl")
         self.config(key_file=os.path.join(TEST_VAR_DIR, 'privatekey.key'),
@@ -540,3 +579,26 @@ class WSGIServerTest(utils.BaseTestCase):
         self.assertEquals(greetings, response.read())
 
         server.stop()
+
+    def test_app_using_router_ipv6_and_ssl(self):
+        self.config(cert_file=os.path.join(TEST_VAR_DIR, 'certificate.crt'),
+                    group="ssl")
+        self.config(key_file=os.path.join(TEST_VAR_DIR, 'privatekey.key'),
+                    group="ssl")
+
+        greetings = 'Hello, World!!!'
+
+        @webob.dec.wsgify
+        def hello(req):
+            return greetings
+
+        mapper = routes.Mapper()
+        mapper.connect(None, "/v1.0/{path_info:.*}", controller=hello)
+        router = wsgi.Router(mapper)
+        server = wsgi.Service(router, 0, host="::1")
+        server.start()
+
+        response = urllib2.urlopen('https://[::1]:%d/v1.0/' % server.port)
+        self.assertEquals(greetings, response.read())
+
+        server.stop()