Simplify broker tests with helper function.

test/broker/02-subpub-qos0.py

@@ -26,13 +26,11 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
-    sock.send(subscribe_packet)
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+    mosq_test.do_send_receive(sock, publish_packet, publish_packet, "publish")
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-subpub-qos1.py

@@ -32,15 +32,13 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
-    sock.send(subscribe_packet)
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-        if mosq_test.expect_packet(sock, "puback", puback_packet):
+    mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
 
-            if mosq_test.expect_packet(sock, "publish2", publish_packet2):
-                rc = 0
+    if mosq_test.expect_packet(sock, "publish2", publish_packet2):
+        rc = 0
 
     sock.close()
 finally:

test/broker/02-subpub-qos2.py

@@ -37,21 +37,15 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
-    sock.send(subscribe_packet)
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+    mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec")
+    mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "pubcomp")
 
-        if mosq_test.expect_packet(sock, "pubrec", pubrec_packet):
-            sock.send(pubrel_packet)
-
-            if mosq_test.expect_packet(sock, "pubcomp", pubcomp_packet):
-                if mosq_test.expect_packet(sock, "publish2", publish_packet2):
-                    sock.send(pubrec_packet2)
-
-                    if mosq_test.expect_packet(sock, "pubrel2", pubrel_packet2):
-                        # Broker side of flow complete so can quit here.
-                        rc = 0
+    if mosq_test.expect_packet(sock, "publish2", publish_packet2):
+        mosq_test.do_send_receive(sock, pubrec_packet2, pubrel_packet2, "pubrel2")
+        # Broker side of flow complete so can quit here.
+        rc = 0
 
     sock.close()
 finally:

test/broker/02-subscribe-invalid-utf8.py

@@ -32,10 +32,9 @@ try:
     time.sleep(0.5)
 
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, "", "suback")
 
-    if mosq_test.expect_packet(sock, "suback", ""):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-subscribe-qos0.py

@@ -27,10 +27,9 @@ try:
     time.sleep(0.5)
 
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-subscribe-qos1.py

@@ -24,10 +24,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-subscribe-qos2.py

@@ -24,10 +24,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-unsubscribe-invalid-no-topic.py

@@ -29,10 +29,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(unsubscribe_packet)
+    mosq_test.do_send_receive(sock, unsubscribe_packet, "", "disconnect")
 
-    if mosq_test.expect_packet(sock, "disconnect", ""):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-unsubscribe-qos0.py

@@ -25,10 +25,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(unsubscribe_packet)
+    mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
 
-    if mosq_test.expect_packet(sock, "unsuback", unsuback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-unsubscribe-qos1.py

@@ -24,10 +24,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(unsubscribe_packet)
+    mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
 
-    if mosq_test.expect_packet(sock, "unsuback", unsuback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/02-unsubscribe-qos2.py

@@ -24,10 +24,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(unsubscribe_packet)
+    mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
 
-    if mosq_test.expect_packet(sock, "unsuback", unsuback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/03-pattern-matching.py

@@ -35,22 +35,19 @@ def pattern_test(sub_topic, pub_topic):
         time.sleep(0.5)
 
         sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
-        sock.send(subscribe_packet)
+        mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-        if mosq_test.expect_packet(sock, "suback", suback_packet):
-            pub = subprocess.Popen(['./03-pattern-matching-helper.py', pub_topic, str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            pub.wait()
-            (stdo, stde) = pub.communicate()
+        pub = subprocess.Popen(['./03-pattern-matching-helper.py', pub_topic, str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        pub.wait()
+        (stdo, stde) = pub.communicate()
 
-            if mosq_test.expect_packet(sock, "publish", publish_packet):
-                sock.send(unsubscribe_packet)
+        if mosq_test.expect_packet(sock, "publish", publish_packet):
+            mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
 
-                if mosq_test.expect_packet(sock, "unsuback", unsuback_packet):
-                    sock.send(subscribe_packet)
+            mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-                    if mosq_test.expect_packet(sock, "suback", suback_packet):
-                        if mosq_test.expect_packet(sock, "publish retained", publish_retained_packet):
-                            rc = 0
+            if mosq_test.expect_packet(sock, "publish retained", publish_retained_packet):
+                rc = 0
 
         sock.close()
     finally:

test/broker/03-publish-b2c-disconnect-qos1-helper.py

@@ -20,10 +20,10 @@ publish_packet = mosq_test.gen_publish("qos1/disconnect/test", qos=1, mid=mid, p
 puback_packet = mosq_test.gen_puback(mid)
 
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, connack_error="helper connack", port=port)
-sock.send(publish_packet)
 
-if mosq_test.expect_packet(sock, "helper puback", puback_packet):
-    rc = 0
+mosq_test.do_send_receive(sock, publish_packet, puback_packet, "helper puback")
+
+rc = 0
 
 sock.close()
 

test/broker/03-publish-b2c-disconnect-qos1.py

@@ -36,30 +36,29 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack1_packet, port=port)
-    sock.send(subscribe_packet)
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./03-publish-b2c-disconnect-qos1-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        pub.wait()
-        (stdo, stde) = pub.communicate()
-        # Should have now received a publish command
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            # Send our outgoing message. When we disconnect the broker
-            # should get rid of it and assume we're going to retry.
-            sock.send(publish2_packet)
-            sock.close()
+    pub = subprocess.Popen(['./03-publish-b2c-disconnect-qos1-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    pub.wait()
+    (stdo, stde) = pub.communicate()
+    # Should have now received a publish command
 
-            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            sock.settimeout(60) # 60 seconds timeout is much longer than 5 seconds message retry.
-            sock.connect(("localhost", port))
-            sock.send(connect_packet)
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        # Send our outgoing message. When we disconnect the broker
+        # should get rid of it and assume we're going to retry.
+        sock.send(publish2_packet)
+        sock.close()
 
-            if mosq_test.expect_packet(sock, "connack", connack2_packet):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.settimeout(60) # 60 seconds timeout is much longer than 5 seconds message retry.
+        sock.connect(("localhost", port))
 
-                if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
-                    sock.send(puback_packet)
-                    rc = 0
+        mosq_test.do_send_receive(sock, connect_packet, connack2_packet, "connack")
+
+        if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
+            sock.send(puback_packet)
+            rc = 0
 
     sock.close()
 finally:

test/broker/03-publish-b2c-disconnect-qos2-helper.py

@@ -24,13 +24,11 @@ pubrel_packet = mosq_test.gen_pubrel(mid)
 pubcomp_packet = mosq_test.gen_pubcomp(mid)
 
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, connack_error="helper connack", port=port)
-sock.send(publish_packet)
 
-if mosq_test.expect_packet(sock, "helper pubrec", pubrec_packet):
-    sock.send(pubrel_packet)
+mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "helper pubrec")
+mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "helper pubcomp")
 
-    if mosq_test.expect_packet(sock, "helper pubcomp", pubcomp_packet):
-        rc = 0
+rc = 0
 
 sock.close()
 

test/broker/03-publish-b2c-disconnect-qos2.py

@@ -36,34 +36,33 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack1_packet, port=port)
-    sock.send(subscribe_packet)
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./03-publish-b2c-disconnect-qos2-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        hrc = pub.wait()
-        (stdo, stde) = pub.communicate()
-        if hrc:
-            exit(hrc)
-        # Should have now received a publish command
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+    pub = subprocess.Popen(['./03-publish-b2c-disconnect-qos2-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    hrc = pub.wait()
+    (stdo, stde) = pub.communicate()
+    if hrc:
+        exit(hrc)
+    # Should have now received a publish command
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        # Send our outgoing message. When we disconnect the broker
+        # should get rid of it and assume we're going to retry.
+        sock.send(publish2_packet)
+        sock.close()
+
+        sock = mosq_test.do_client_connect(connect_packet, connack2_packet, port=port)
+        if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
+            mosq_test.do_send_receive(sock, pubrec_packet, pubrel_packet, "pubrel")
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            # Send our outgoing message. When we disconnect the broker
-            # should get rid of it and assume we're going to retry.
-            sock.send(publish2_packet)
             sock.close()
 
             sock = mosq_test.do_client_connect(connect_packet, connack2_packet, port=port)
-            if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
-                sock.send(pubrec_packet)
-
-                if mosq_test.expect_packet(sock, "pubrel", pubrel_packet):
-                    sock.close()
-
-                    sock = mosq_test.do_client_connect(connect_packet, connack2_packet, port=port)
-                    if mosq_test.expect_packet(sock, "dup pubrel", pubrel_packet):
-                        sock.send(pubcomp_packet)
-                        rc = 0
-                    sock.close()
+            if mosq_test.expect_packet(sock, "dup pubrel", pubrel_packet):
+                sock.send(pubcomp_packet)
+                rc = 0
+            sock.close()
 
 finally:
     broker.terminate()

test/broker/03-publish-b2c-timeout-qos1-helper.py

@@ -20,10 +20,9 @@ publish_packet = mosq_test.gen_publish("qos1/timeout/test", qos=1, mid=mid, payl
 puback_packet = mosq_test.gen_puback(mid)
 
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, connack_error="helper connack")
-sock.send(publish_packet)
+mosq_test.do_send_receive(sock, publish_packet, puback_packet, "helper puback")
 
-if mosq_test.expect_packet(sock, "helper puback", puback_packet):
-    rc = 0
+rc = 0
 
 sock.close()
 

test/broker/03-publish-b2c-timeout-qos1.py

@@ -30,21 +30,20 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__))
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./03-publish-b2c-timeout-qos1-helper.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE))
-        pub.wait()
-        (stdo, stde) = pub.communicate()
-        # Should have now received a publish command
+    pub = subprocess.Popen(['./03-publish-b2c-timeout-qos1-helper.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE))
+    pub.wait()
+    (stdo, stde) = pub.communicate()
+    # Should have now received a publish command
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            # Wait for longer than 5 seconds to get republish with dup set
-            # This is covered by the 8 second timeout
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        # Wait for longer than 5 seconds to get republish with dup set
+        # This is covered by the 8 second timeout
 
-            if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
-                sock.send(puback_packet)
-                rc = 0
+        if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
+            sock.send(puback_packet)
+            rc = 0
 
     sock.close()
 finally:

test/broker/03-publish-b2c-timeout-qos2-helper.py

@@ -22,13 +22,10 @@ pubrel_packet = mosq_test.gen_pubrel(mid)
 pubcomp_packet = mosq_test.gen_pubcomp(mid)
 
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, connack_error="helper connack")
-sock.send(publish_packet)
+mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "helper pubrec")
+mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "helper pubcomp")
 
-if mosq_test.expect_packet(sock, "helper pubrec", pubrec_packet):
-    sock.send(pubrel_packet)
-
-    if mosq_test.expect_packet(sock, "helper pubcomp", pubcomp_packet):
-        rc = 0
+rc = 0
 
 sock.close()
 

test/broker/03-publish-b2c-timeout-qos2.py

@@ -32,28 +32,26 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__))
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./03-publish-b2c-timeout-qos2-helper.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE))
-        pub.wait()
-        (stdo, stde) = pub.communicate()
-        # Should have now received a publish command
+    pub = subprocess.Popen(['./03-publish-b2c-timeout-qos2-helper.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE))
+    pub.wait()
+    (stdo, stde) = pub.communicate()
+    # Should have now received a publish command
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        # Wait for longer than 5 seconds to get republish with dup set
+        # This is covered by the 8 second timeout
+
+        if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
+            mosq_test.do_send_receive(sock, pubrec_packet, pubrel_packet, "pubrel")
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
             # Wait for longer than 5 seconds to get republish with dup set
             # This is covered by the 8 second timeout
 
-            if mosq_test.expect_packet(sock, "dup publish", publish_dup_packet):
-                sock.send(pubrec_packet)
-
-                if mosq_test.expect_packet(sock, "pubrel", pubrel_packet):
-                    # Wait for longer than 5 seconds to get republish with dup set
-                    # This is covered by the 8 second timeout
-
-                    if mosq_test.expect_packet(sock, "dup pubrel", pubrel_packet):
-                        sock.send(pubcomp_packet)
-                        rc = 0
+            if mosq_test.expect_packet(sock, "dup pubrel", pubrel_packet):
+                sock.send(pubcomp_packet)
+                rc = 0
 
     sock.close()
 finally:

test/broker/03-publish-c2b-disconnect-qos2.py

@@ -31,28 +31,26 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
 
-    sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec")
+
+    # We're now going to disconnect and pretend we didn't receive the pubrec.
+    sock.close()
+
+    sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+    sock.send(publish_dup_packet)
+
     if mosq_test.expect_packet(sock, "pubrec", pubrec_packet):
-        # We're now going to disconnect and pretend we didn't receive the pubrec.
+        mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "pubcomp")
+
+        # Again, pretend we didn't receive this pubcomp
         sock.close()
 
         sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-        sock.send(publish_dup_packet)
+        mosq_test.do_send_receive(sock, pubrel_dup_packet, pubcomp_packet, "pubcomp")
 
-        if mosq_test.expect_packet(sock, "pubrec", pubrec_packet):
-            sock.send(pubrel_packet)
+        rc = 0
 
-            if mosq_test.expect_packet(sock, "pubcomp", pubcomp_packet):
-                # Again, pretend we didn't receive this pubcomp
-                sock.close()
-
-                sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-                sock.send(pubrel_dup_packet)
-
-                if mosq_test.expect_packet(sock, "pubcomp", pubcomp_packet):
-                    rc = 0
-
-                sock.close()
+        sock.close()
 finally:
     broker.terminate()
     broker.wait()

test/broker/03-publish-c2b-timeout-qos2.py

@@ -27,16 +27,14 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__))
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet)
-    sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec")
+
+    # Timeout is 8 seconds which means the broker should repeat the PUBREC.
 
     if mosq_test.expect_packet(sock, "pubrec", pubrec_packet):
-        # Timeout is 8 seconds which means the broker should repeat the PUBREC.
+        mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "pubcomp")
 
-        if mosq_test.expect_packet(sock, "pubrec", pubrec_packet):
-            sock.send(pubrel_packet)
-
-            if mosq_test.expect_packet(sock, "pubcomp", pubcomp_packet):
-                rc = 0
+        rc = 0
 
     sock.close()
 finally:

test/broker/03-publish-invalid-utf8.py

@@ -32,10 +32,9 @@ try:
     time.sleep(0.5)
 
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, "", "puback")
 
-    if mosq_test.expect_packet(sock, "puback", ""):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/03-publish-qos1.py

@@ -24,10 +24,9 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
 
-    if mosq_test.expect_packet(sock, "puback", puback_packet):
-        rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/03-publish-qos2.py

@@ -26,13 +26,10 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec")
+    mosq_test.do_send_receive(sock, pubrel_packet, pubcomp_packet, "pubcomp")
 
-    if mosq_test.expect_packet(sock, "pubrec", pubrec_packet):
-        sock.send(pubrel_packet)
-
-        if mosq_test.expect_packet(sock, "pubcomp", pubcomp_packet):
-            rc = 0
+    rc = 0
 
     sock.close()
 finally:

test/broker/04-retain-qos0-clear.py

@@ -36,29 +36,26 @@ try:
     # Send retained message
     sock.send(publish_packet)
     # Subscribe to topic, we should get the retained message back.
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            # Now unsubscribe from the topic before we clear the retained
-            # message.
-            sock.send(unsubscribe_packet)
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        # Now unsubscribe from the topic before we clear the retained
+        # message.
+        mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
 
-            if mosq_test.expect_packet(sock, "unsuback", unsuback_packet):
-                # Now clear the retained message.
-                sock.send(retain_clear_packet)
+        # Now clear the retained message.
+        sock.send(retain_clear_packet)
 
-                # Subscribe to topic, we shouldn't get anything back apart
-                # from the SUBACK.
-                sock.send(subscribe_packet)
-                if mosq_test.expect_packet(sock, "suback", suback_packet):
-                    try:
-                        retain_clear = sock.recv(256)
-                    except socket.timeout:
-                        # This is the expected event
-                        rc = 0
-                    else:
-                        print("FAIL: Received unexpected message.")
+        # Subscribe to topic, we shouldn't get anything back apart
+        # from the SUBACK.
+        mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+        try:
+            retain_clear = sock.recv(256)
+        except socket.timeout:
+            # This is the expected event
+            rc = 0
+        else:
+            print("FAIL: Received unexpected message.")
 
     sock.close()
 finally:

test/broker/04-retain-qos0-fresh.py

@@ -27,13 +27,11 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+    mosq_test.do_send_receive(sock, publish_packet, publish_fresh_packet, "publish")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(publish_packet)
+    rc = 0
 
-        if mosq_test.expect_packet(sock, "publish", publish_fresh_packet):
-            rc = 0
     sock.close()
 finally:
     broker.terminate()

test/broker/04-retain-qos0-repeated.py

@@ -31,18 +31,15 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
     sock.send(publish_packet)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        mosq_test.do_send_receive(sock, unsubscribe_packet, unsuback_packet, "unsuback")
+        mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
         if mosq_test.expect_packet(sock, "publish", publish_packet):
-            sock.send(unsubscribe_packet)
+            rc = 0
 
-            if mosq_test.expect_packet(sock, "unsuback", unsuback_packet):
-                sock.send(subscribe_packet)
-
-                if mosq_test.expect_packet(sock, "suback", suback_packet):
-                    if mosq_test.expect_packet(sock, "publish", publish_packet):
-                        rc = 0
     sock.close()
 finally:
     broker.terminate()

test/broker/04-retain-qos0.py

@@ -26,11 +26,11 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
     sock.send(publish_packet)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
     sock.close()
 finally:
     broker.terminate()

test/broker/04-retain-qos1-qos0.py

@@ -30,14 +30,12 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "puback", puback_packet):
-        sock.send(subscribe_packet)
+    if mosq_test.expect_packet(sock, "publish0", publish0_packet):
+        rc = 0
 
-        if mosq_test.expect_packet(sock, "suback", suback_packet):
-            if mosq_test.expect_packet(sock, "publish0", publish0_packet):
-                rc = 0
     sock.close()
 finally:
     broker.terminate()

test/broker/04-retain-upgrade-outgoing-qos.py

@@ -38,13 +38,11 @@ try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
     sock.send(publish_packet)
 
-    #sock.close()
-    #sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet2):
+        rc = 0
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        if mosq_test.expect_packet(sock, "publish", publish_packet2):
-            rc = 0
     sock.close()
 finally:
     os.remove(conf_file)

test/broker/05-clean-session-qos1-helper.py

@@ -21,10 +21,9 @@ puback_packet = mosq_test.gen_puback(mid)
 
 port = mosq_test.get_port()
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
-sock.send(publish_packet)
+mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
 
-if mosq_test.expect_packet(sock, "puback", puback_packet):
-    rc = 0
+rc = 0
 
 sock.close()
 

test/broker/05-clean-session-qos1.py

@@ -33,23 +33,22 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack1_packet, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(disconnect_packet)
-        sock.close()
+    sock.send(disconnect_packet)
+    sock.close()
 
-        pub = subprocess.Popen(['./05-clean-session-qos1-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        pub.wait()
-        (stdo, stde) = pub.communicate()
+    pub = subprocess.Popen(['./05-clean-session-qos1-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    pub.wait()
+    (stdo, stde) = pub.communicate()
 
-        # Now reconnect and expect a publish message.
-        sock = mosq_test.do_client_connect(connect_packet, connack2_packet, timeout=30, port=port)
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            sock.send(puback_packet)
-            rc = 0
+    # Now reconnect and expect a publish message.
+    sock = mosq_test.do_client_connect(connect_packet, connack2_packet, timeout=30, port=port)
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        sock.send(puback_packet)
+        rc = 0
 
-        sock.close()
+    sock.close()
 finally:
     broker.terminate()
     broker.wait()

test/broker/06-bridge-br2b-disconnect-qos1-helper.py

@@ -20,10 +20,9 @@ publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid,
 puback_packet = mosq_test.gen_puback(mid)
 
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, connack_error="helper connack")
-sock.send(publish_packet)
+mosq_test.do_send_receive(sock, publish_packet, puback_packet, "helper puback")
 
-if mosq_test.expect_packet(sock, "helper puback", puback_packet):
-    rc = 0
+rc = 0
 
 sock.close()
 

test/broker/06-bridge-reconnect-local-out-helper.py

@@ -21,11 +21,10 @@ puback_packet = mosq_test.gen_puback(mid=1)
 disconnect_packet = mosq_test.gen_disconnect()
 
 sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port, connack_error="helper connack")
-sock.send(publish_packet)
+mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
 
-if mosq_test.expect_packet(sock, "puback", puback_packet):
-    sock.send(disconnect_packet)
-    rc = 0
+sock.send(disconnect_packet)
+rc = 0
 
 sock.close()
 

test/broker/06-bridge-reconnect-local-out.py

@@ -67,19 +67,17 @@ else:
 pub = None
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port1)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(subscribe_packet)
+    pub = subprocess.Popen(['./06-bridge-reconnect-local-out-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    pub.wait()
+    (stdo, stde) = pub.communicate()
+    # Should have now received a publish command
 
-        if mosq_test.expect_packet(sock, "suback", suback_packet):
-            pub = subprocess.Popen(['./06-bridge-reconnect-local-out-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            pub.wait()
-            (stdo, stde) = pub.communicate()
-            # Should have now received a publish command
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
-            if mosq_test.expect_packet(sock, "publish", publish_packet):
-                rc = 0
     sock.close()
 finally:
     os.remove(conf_file)

test/broker/07-will-null.py

@@ -27,15 +27,14 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=30, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        will = subprocess.Popen(['./07-will-null-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        will.wait()
-        (stdo, stde) = will.communicate()
+    will = subprocess.Popen(['./07-will-null-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    will.wait()
+    (stdo, stde) = will.communicate()
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
     sock.close()
 finally:

test/broker/07-will-qos0.py

@@ -28,15 +28,14 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=30, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        will = subprocess.Popen(['./07-will-qos0-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        will.wait()
-        (stdo, stde) = will.communicate()
+    will = subprocess.Popen(['./07-will-qos0-helper.py', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    will.wait()
+    (stdo, stde) = will.communicate()
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
     sock.close()
 finally:

test/broker/08-ssl-connect-cert-auth-crl.py

@@ -42,10 +42,10 @@ try:
     ssock = ssl.wrap_socket(sock, ca_certs="../ssl/test-root-ca.crt", certfile="../ssl/client.crt", keyfile="../ssl/client.key", cert_reqs=ssl.CERT_REQUIRED)
     ssock.settimeout(20)
     ssock.connect(("localhost", port1))
-    ssock.send(connect_packet)
 
-    if mosq_test.expect_packet(ssock, "connack", connack_packet):
-        rc = 0
+    mosq_test.do_send_receive(ssock, connect_packet, connack_packet, "connack")
+
+    rc = 0
 
     ssock.close()
 finally:

test/broker/08-ssl-connect-cert-auth.py

@@ -43,10 +43,10 @@ try:
     ssock = ssl.wrap_socket(sock, ca_certs="../ssl/test-root-ca.crt", certfile="../ssl/client.crt", keyfile="../ssl/client.key", cert_reqs=ssl.CERT_REQUIRED)
     ssock.settimeout(20)
     ssock.connect(("localhost", port1))
-    ssock.send(connect_packet)
 
-    if mosq_test.expect_packet(ssock, "connack", connack_packet):
-        rc = 0
+    mosq_test.do_send_receive(ssock, connect_packet, connack_packet, "connack")
+
+    rc = 0
 
     ssock.close()
 finally:

test/broker/08-ssl-connect-identity.py

@@ -46,10 +46,10 @@ try:
     ssock = ssl.wrap_socket(sock, ca_certs="../ssl/test-root-ca.crt", certfile="../ssl/client.crt", keyfile="../ssl/client.key", cert_reqs=ssl.CERT_REQUIRED)
     ssock.settimeout(20)
     ssock.connect(("localhost", port1))
-    ssock.send(connect_packet)
 
-    if mosq_test.expect_packet(ssock, "connack", connack_packet):
-        rc = 0
+    mosq_test.do_send_receive(ssock, connect_packet, connack_packet, "connack")
+
+    rc = 0
 
     ssock.close()
 finally:

test/broker/08-ssl-connect-no-auth.py

@@ -43,10 +43,10 @@ try:
     ssock = ssl.wrap_socket(sock, ca_certs="../ssl/test-root-ca.crt", cert_reqs=ssl.CERT_REQUIRED, ssl_version=ssl.PROTOCOL_TLSv1)
     ssock.settimeout(20)
     ssock.connect(("localhost", port1))
-    ssock.send(connect_packet)
 
-    if mosq_test.expect_packet(ssock, "connack", connack_packet):
-        rc = 0
+    mosq_test.do_send_receive(ssock, connect_packet, connack_packet, "connack")
+
+    rc = 0
 
     ssock.close()
 finally:

test/broker/08-ssl-connect-no-identity.py

@@ -45,10 +45,10 @@ try:
     ssock = ssl.wrap_socket(sock, ca_certs="../ssl/test-root-ca.crt", cert_reqs=ssl.CERT_REQUIRED)
     ssock.settimeout(20)
     ssock.connect(("localhost", port1))
-    ssock.send(connect_packet)
+    
+    mosq_test.do_send_receive(ssock, connect_packet, connack_packet, "connack")
 
-    if mosq_test.expect_packet(ssock, "connack", connack_packet):
-        rc = 0
+    rc = 0
 
     ssock.close()
 finally:

test/broker/08-tls-psk-bridge.py

@@ -72,16 +72,17 @@ bridge = mosq_test.start_broker(filename=os.path.basename(__file__)+'_bridge', c
 pub = None
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=30, port=port1)
-    sock.send(subscribe_packet)
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./c/08-tls-psk-bridge.test', str(port3)], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        if pub.wait():
-            raise ValueError
-        (stdo, stde) = pub.communicate()
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
+
+    pub = subprocess.Popen(['./c/08-tls-psk-bridge.test', str(port3)], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if pub.wait():
+        raise ValueError
+    (stdo, stde) = pub.communicate()
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
     sock.close()
 finally:
     os.remove(conf_file1)

test/broker/08-tls-psk-pub.py

@@ -56,16 +56,16 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=Tr
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port2)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./c/08-tls-psk-pub.test', str(port1)], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        if pub.wait():
-            raise ValueError
-        (stdo, stde) = pub.communicate()
+    pub = subprocess.Popen(['./c/08-tls-psk-pub.test', str(port1)], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if pub.wait():
+        raise ValueError
+    (stdo, stde) = pub.communicate()
+
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
     sock.close()
 finally:
     os.remove(conf_file)

test/broker/09-plugin-auth-acl-sub.py

@@ -37,12 +37,11 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=Tr
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        sock.send(subscribe_packet_fail)
-        if mosq_test.expect_packet(sock, "suback", suback_packet_fail):
-            rc = 0
+    mosq_test.do_send_receive(sock, subscribe_packet_fail, suback_packet_fail, "suback")
+
+    rc = 0
 
     sock.close()
 finally:

test/broker/10-listener-mount-point.py

@@ -38,16 +38,15 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=Tr
 
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port1)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        pub = subprocess.Popen(['./10-listener-mount-point-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        pub.wait()
-        (stdo, stde) = pub.communicate()
-        # Should have now received a publish command
+    pub = subprocess.Popen(['./10-listener-mount-point-helper.py', str(port2)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    pub.wait()
+    (stdo, stde) = pub.communicate()
+    # Should have now received a publish command
 
-        if mosq_test.expect_packet(sock, "publish", publish_packet):
-            rc = 0
+    if mosq_test.expect_packet(sock, "publish", publish_packet):
+        rc = 0
 
     sock.close()
 finally:

test/broker/11-persistent-subscription.py

@@ -47,23 +47,20 @@ broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=Tr
 (stdo1, stde1) = ("", "")
 try:
     sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
-    sock.send(subscribe_packet)
+    mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
 
-    if mosq_test.expect_packet(sock, "suback", suback_packet):
-        broker.terminate()
-        broker.wait()
-        (stdo1, stde1) = broker.communicate()
-        sock.close()
-        broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
+    broker.terminate()
+    broker.wait()
+    (stdo1, stde1) = broker.communicate()
+    sock.close()
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
 
-        sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=20, port=port)
+    sock = mosq_test.do_client_connect(connect_packet, connack_packet2, timeout=20, port=port)
 
-        sock.send(publish_packet)
+    mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback")
 
-        if mosq_test.expect_packet(sock, "puback", puback_packet):
-
-            if mosq_test.expect_packet(sock, "publish2", publish_packet2):
-                rc = 0
+    if mosq_test.expect_packet(sock, "publish2", publish_packet2):
+        rc = 0
 
     sock.close()
 finally: