mock dns responses to make running tests easier and more efficient

README.rst

@@ -67,10 +67,6 @@ stomp.py has been perfunctorily tested on:
 
 For testing locally, you'll need to install docker. Once installed:
 
-#. Add test domain names to your hosts file like so: 
-      |  172.17.0.2  my.example.com
-      |  172.17.0.2  my.example.org
-      |  172.17.0.2  my.example.net
 #. Install dependencies:
         poetry install
 #. Install optional dependencies (if testing with ssl)
@@ -84,6 +80,11 @@ For testing locally, you'll need to install docker. Once installed:
 #. Cleanup the container afterwards if you don't need it any more:
         make remove-docker
 
+If you want to connect to the test services locally (other than from the included tests), you'll want to add test domain names to your hosts file like so:
+      |  172.17.0.2  my.example.com
+      |  172.17.0.2  my.example.org
+      |  172.17.0.2  my.example.net
+
 
 .. _`STOMP`: http://stomp.github.io
 .. _`STOMP v1.0`: http://stomp.github.io/stomp-specification-1.0.html

tests/test_basic.py

@@ -1,4 +1,5 @@
 import signal
+import socket
 from time import monotonic
 
 import stomp
@@ -27,6 +28,19 @@ def invalidconn(testlistener):
     yield conn
 
 
+@pytest.fixture()
+def ipv4only(monkeypatch):
+    """
+    Filter DNS requests to return only IPv4 results. This is useful to avoid long timeouts
+    when tests attempt to connect to the IPv6 address, but the service is only listening
+    on the IPv4 address.
+    """
+    real_getaddrinfo = socket.getaddrinfo
+    def getaddrinfo_ipv4(*args, **kw):
+        return [a for a in real_getaddrinfo(*args, **kw) if a[0] == socket.AF_INET]
+    monkeypatch.setattr(socket, "getaddrinfo", getaddrinfo_ipv4)
+
+
 class TestBasic(object):
 
     def test_subscribe_and_send(self, conn, testlistener):
@@ -42,7 +56,7 @@ def test_subscribe_and_send(self, conn, testlistener):
         assert "content-type" in headers
         assert headers["content-type"] == "application/json"
 
-    def test_default_to_localhost(self):
+    def test_default_to_localhost(self, ipv4only):
         conn = stomp.Connection()
         listener = TestListener("123", print_to_log=True)
         queuename = "/queue/test1-%s" % listener.timestamp
@@ -198,7 +212,7 @@ def test_specialchars(self, conn):
         assert "special-3" in headers
         assert "test with newline \n" == headers["special-3"]
 
-    def test_host_bind_port(self):
+    def test_host_bind_port(self, ipv4only):
         conn = stomp.Connection(bind_host_port=("localhost", next_free_port()))
         listener = TestListener("981", print_to_log=True)
         queuename = "/queue/testbind-%s" % listener.timestamp

tests/test_ss.py

@@ -25,7 +25,7 @@ def test_alternate_hosts(self, server):
         conn = stomp.Connection([("192.0.2.0", 10000), ("127.0.0.1", 60000)], timeout=1, prefer_localhost=False)
         listener = TestListener(print_to_log=True)
         conn.set_listener('', listener)
-        conn.connect()
+        conn.connect(wait=True)
 
     def test_disconnect(self, server):
         server.add_frame('''CONNECTED

tests/test_ssl_sni.py

@@ -1,3 +1,4 @@
+import socket
 import stomp
 from stomp.listener import TestListener
 from .testutils import *
@@ -9,17 +10,16 @@ class TestSNIMQSend(object):
 
     - Start the docker container
 
-    - Add a couple fully qualified hostnames to your /etc/hosts
-        # SNI test hosts
-        172.17.0.2 my.example.com
-        172.17.0.2 my.example.org
-
     Connections with SNI to "my.example.com" will be routed to the STOMP server on port 62613.
     Connections without SNI won't be routed.
 
     """
 
-    def testconnect(self):
+    def testconnect(self, monkeypatch):
+        def getaddrinfo_fake(host, port, *args, **kw):
+            """Always return the IP address of the container."""
+            return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('172.17.0.2', port))]
+        monkeypatch.setattr(socket, "getaddrinfo", getaddrinfo_fake)
         if not is_inside_travis():
             logging.info("Running ipv6 test")
             receipt_id = str(uuid.uuid4())

tests/testutils.py

@@ -159,7 +159,7 @@ def get_next_frame(self):
             del self.frames[0]
             return rtn
         else:
-            return ''
+            return None
 
     def add_frame(self, frame):
         self.frames.append(frame)