From: Mircea Ulinic <mirucha@cloudflare.com>
Date: Wed, 1 Feb 2017 13:18:46 +0000 (+0000)
Subject: Add tests for source_ip and source_port
X-Git-Tag: v4.5.0~39^2~1
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=fab3b14bb827b1f5145ece4bffa89728a8b6f6e3;p=thirdparty%2Ftornado.git

Add tests for source_ip and source_port
---

diff --git a/tornado/test/tcpclient_test.py b/tornado/test/tcpclient_test.py
index 1a4201e6b..2deca8abb 100644
--- a/tornado/test/tcpclient_test.py
+++ b/tornado/test/tcpclient_test.py
@@ -81,9 +81,11 @@ class TCPClientTest(AsyncTestCase):
             self.skipTest("localhost does not resolve to ipv6")
 
     @gen_test
-    def do_test_connect(self, family, host):
+    def do_test_connect(self, family, host, source_ip=None, source_port=None):
         port = self.start_server(family)
-        stream = yield self.client.connect(host, port)
+        stream = yield self.client.connect(host, port,
+                                           source_ip=source_ip,
+                                           source_port=source_port)
         with closing(stream):
             stream.write(b"hello")
             data = yield self.server.streams[0].read_bytes(5)
@@ -125,6 +127,32 @@ class TCPClientTest(AsyncTestCase):
         with self.assertRaises(IOError):
             yield self.client.connect('127.0.0.1', port)
 
+    def test_source_ip_fail(self):
+        '''
+        Fail when trying to use the source IP Address '8.8.8.8'.
+        '''
+        self.assertRaises(socket.error,
+                          self.do_test_connect,
+                          socket.AF_INET,
+                          '127.0.0.1',
+                          source_ip='8.8.8.8')
+
+    def test_source_ip_success(self):
+        '''
+        Success when trying to use the source IP Address '127.0.0.1'
+        '''
+        self.do_test_connect(socket.AF_INET, '127.0.0.1', source_ip='127.0.0.1')
+
+    def test_source_port_fail(self):
+        '''
+        Fail when trying to use source port 1.
+        '''
+        self.assertRaises(socket.error,
+                          self.do_test_connect,
+                          socket.AF_INET,
+                          '127.0.0.1',
+                          source_port=1)
+
 
 class TestConnectorSplit(unittest.TestCase):
     def test_one_family(self):