add test for ipv6 functionality

Signed-off-by: Matt Cipperly <[email protected]>

Author: mcipperly

tests/mockserver.py

@@ -16,21 +16,21 @@ class MockRecvServer(threading.Thread):
     Single threaded server accepts one connection and recv until EOF.
     """
 
-    def __init__(self, host='localhost', port=0):
+    def __init__(self, host='localhost', port=0, inet_family=socket.AF_INET):
         super(MockRecvServer, self).__init__()
 
         if host.startswith('unix://'):
             self.socket_proto = socket.AF_UNIX
             self.socket_type = socket.SOCK_STREAM
             self.socket_addr = host[len('unix://'):]
         else:
-            self.socket_proto = socket.AF_INET
+            self.socket_proto = inet_family
             self.socket_type = socket.SOCK_STREAM
             self.socket_addr = (host, port)
 
         self._sock = socket.socket(self.socket_proto, self.socket_type)
         self._sock.bind(self.socket_addr)
-        if self.socket_proto == socket.AF_INET:
+        if self.socket_proto == inet_family:
             self.port = self._sock.getsockname()[1]
 
         self._sock.listen(1)
@@ -76,7 +76,7 @@ def close(self):
             pass
 
         try:
-            conn = socket.socket(socket.AF_INET,
+            conn = socket.socket(self.socket_proto,
                                  socket.SOCK_STREAM)
             try:
                 conn.connect((self.socket_addr[0], self.port))

tests/test_sender.py

@@ -8,6 +8,7 @@
 import unittest
 from shutil import rmtree
 from tempfile import mkdtemp
+from unittest.mock import patch
 
 import msgpack
 
@@ -291,6 +292,37 @@ def recv(self, bufsize, flags=0):
             finally:
                 self._sender.socket = old_sock
 
+    def test_ipv6_only(self):
+        # Test if our host supports IPv6 before running this test
+        try:
+            socket.gethostbyaddr('::1')
+        except socket.herror:
+            self.skipTest("Host does not support IPv6, cannot run this test")
+
+        self.tearDown()
+
+        real_getaddrinfo = socket.getaddrinfo
+
+        def _fake_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
+            if family == socket.AF_INET:
+                raise socket.gaierror("mock: IPv6 Only")
+            else:
+                return real_getaddrinfo(host, port, family, type, proto, flags)
+
+        self._server = mockserver.MockRecvServer(host='localhost',
+                                                 inet_family=socket.AF_INET6)
+
+
+        with patch('socket.getaddrinfo', side_effect=_fake_getaddrinfo):
+            sender = fluent.sender.FluentSender(tag='test',
+                                                host='localhost',
+                                                port=self._server.port)
+            sender.emit('foo', {'bar': 'baz'})
+            sender._close()
+            data = self.get_data()
+            self.assertEqual(len(data), 1)
+            self.assertEqual(data[0][2], {'bar': 'baz'})
+
     @unittest.skipIf(sys.platform == "win32", "Unix socket not supported")
     def test_unix_socket(self):
         self.tearDown()