Merge pull request #2573 from buschulte/throw-broken-pipe-error-when-loosing-connection

throw BrokenPipeError if nothing received on socket

test/broker/01-connect-global-max-clients.py

@@ -37,7 +37,7 @@ def do_test():
         # Try to open an 11th connection
         try:
             sock_bad = mosq_test.do_client_connect(connect_packet_bad, connack_packet_bad, port=port)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # Expected behaviour
             pass
 
@@ -50,7 +50,7 @@ def do_test():
         # Try to open an 11th connection
         try:
             sock_bad = mosq_test.do_client_connect(connect_packet_bad, connack_packet_bad, port=port)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # Expected behaviour
             pass
 
@@ -66,7 +66,6 @@ def do_test():
         (stdo, stde) = broker.communicate()
         if rc:
             print(stde.decode('utf-8'))
-            exit(rc)
+    return rc
 
-do_test()
-exit(0)
+sys.exit(do_test())

test/broker/01-connect-global-max-connections.py

@@ -36,7 +36,7 @@ def do_test():
         # Try to open an 11th connection
         try:
             sock_bad = mosq_test.do_client_connect(connect_packet_bad, connack_packet_bad, port=port)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # Expected behaviour
             pass
 
@@ -53,7 +53,7 @@ def do_test():
         # Try to open an 11th connection
         try:
             sock_bad = mosq_test.do_client_connect(connect_packet_bad, connack_packet_bad, port=port)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # Expected behaviour
             pass
 
@@ -73,7 +73,6 @@ def do_test():
         (stdo, stde) = broker.communicate()
         if rc:
             print(stde.decode('utf-8'))
-            exit(rc)
+    return rc
 
-do_test()
-exit(0)
+sys.exit(do_test())

test/broker/01-connect-max-connections.py

@@ -36,7 +36,7 @@ def do_test():
         # Try to open an 11th connection
         try:
             sock_bad = mosq_test.do_client_connect(connect_packet_bad, connack_packet_bad, port=port)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # Expected behaviour
             pass
 
@@ -53,7 +53,7 @@ def do_test():
         # Try to open an 11th connection
         try:
             sock_bad = mosq_test.do_client_connect(connect_packet_bad, connack_packet_bad, port=port)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # Expected behaviour
             pass
 
@@ -73,7 +73,6 @@ def do_test():
         (stdo, stde) = broker.communicate()
         if rc:
             print(stde.decode('utf-8'))
-            exit(rc)
+    return rc
 
-do_test()
-exit(0)
+sys.exit(do_test())

test/broker/02-subpub-qos0-long-topic.py

@@ -16,50 +16,51 @@ def do_test(start_broker, topic, succeeds):
     publish_packet = mosq_test.gen_publish(topic, qos=0, payload="message")
 
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         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)
 
-        if succeeds == True:
+        if succeeds:
             mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback")
             mosq_test.do_send_receive(sock, publish_packet, publish_packet, "publish")
         else:
-            mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+            try:
+                mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+                return 1
+            except BrokenPipeError:
+                pass
 
         rc = 0
 
         sock.close()
-    except mosq_test.TestError:
-        pass
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, "/"*200, True) # 200 max hierarchy limit
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, "abc/"*199+"d", True) # 200 max hierarchy limit, longer overall string than 200
     if rc:
-        return rc;
+        return rc
 
     rc = do_test(start_broker, "/"*201, False) # Exceeds 200 max hierarchy limit
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, "abc/"*201+"d", False) # Exceeds 200 max hierarchy limit, longer overall string than 200
     if rc:
-        return rc;
+        return rc
     return 0
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

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

@@ -15,46 +15,43 @@ def do_test(start_broker, proto_ver):
     b[13] = 0 # Topic should never have a 0x0000
     subscribe_packet = struct.pack("B"*len(b), *b)
 
-    suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
-
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         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)
         if proto_ver == 4:
-            mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+            try:
+                mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+            except BrokenPipeError:
+                rc = 0
         else:
             disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code = mqtt5_rc.MQTT_RC_MALFORMED_PACKET)
             mosq_test.do_send_receive(sock, subscribe_packet, disconnect_packet, "suback")
-
-        rc = 0
+            rc = 0
 
         sock.close()
-    except mosq_test.TestError:
-        pass
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, proto_ver=5)
     if rc:
-        return rc;
+        return rc
     return 0
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

test/broker/02-subscribe-long-topic.py

@@ -13,46 +13,44 @@ def do_test(start_broker, proto_ver):
     connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
     subscribe_packet = mosq_test.gen_subscribe(mid, "/"*65535, 0, proto_ver=proto_ver)
-    suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver)
 
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         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)
         if proto_ver == 4:
-            mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+            try:
+                mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+            except BrokenPipeError:
+                rc = 0
         else:
             disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code = mqtt5_rc.MQTT_RC_MALFORMED_PACKET)
             mosq_test.do_send_receive(sock, subscribe_packet, disconnect_packet, "suback")
-
-        rc = 0
+            rc = 0
 
         sock.close()
-    except mosq_test.TestError:
-        pass
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, proto_ver=5)
     if rc:
-        return rc;
+        return rc
     return 0
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

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

@@ -15,46 +15,43 @@ def do_test(start_broker, proto_ver):
     b[11] = 0 # Topic should never have a 0x0000
     publish_packet = struct.pack("B"*len(b), *b)
 
-    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
-
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         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)
         if proto_ver == 4:
-            mosq_test.do_send_receive(sock, publish_packet, b"", "puback")
+            try:
+                mosq_test.do_send_receive(sock, publish_packet, b"", "puback")
+            except BrokenPipeError:
+                rc = 0
         else:
             disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code=mqtt5_rc.MQTT_RC_MALFORMED_PACKET)
             mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "puback")
-
-        rc = 0
+            rc = 0
 
         sock.close()
-    except mosq_test.TestError:
-        pass
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, proto_ver=5)
     if rc:
-        return rc;
+        return rc
     return 0
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

test/broker/03-publish-long-topic.py

@@ -14,46 +14,44 @@ def do_test(start_broker, proto_ver):
     connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
 
     publish_packet = mosq_test.gen_publish("/"*65535, qos=1, mid=mid, payload="message", proto_ver=proto_ver)
-    puback_packet = mosq_test.gen_puback(mid, proto_ver=proto_ver)
 
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         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)
         if proto_ver == 4:
-            mosq_test.do_send_receive(sock, publish_packet, b"", "puback")
+            try:
+                mosq_test.do_send_receive(sock, publish_packet, b"", "puback")
+            except BrokenPipeError:
+                rc = 0
         else:
             disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code=mqtt5_rc.MQTT_RC_MALFORMED_PACKET)
             mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "puback")
-
-        rc = 0
+            rc = 0
 
         sock.close()
-    except mosq_test.TestError:
-        pass
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, proto_ver=5)
     if rc:
-        return rc;
+        return rc
     return 0
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

test/broker/03-publish-qos2-max-inflight-exceeded.py

@@ -27,13 +27,14 @@ def do_test(proto_ver):
             disconnect_packet = mosq_test.gen_disconnect(reason_code=mqtt5_rc.MQTT_RC_RECEIVE_MAXIMUM_EXCEEDED, proto_ver=proto_ver)
         else:
             disconnect_packet = b""
-        mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "disconnect")
+        try:
+            mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "disconnect")
+        except BrokenPipeError:
+            pass
 
         rc = 0
 
         sock.close()
-    except mosq_test.TestError:
-        pass
     finally:
         broker.terminate()
         broker.wait()
@@ -41,9 +42,15 @@ def do_test(proto_ver):
         if rc:
             print(stde.decode('utf-8'))
             print("proto_ver=%d" % (proto_ver))
-            exit(rc)
+    return rc
 
 
-do_test(proto_ver=4)
-do_test(proto_ver=5)
-exit(0)
+def all_test():
+    rc = do_test(proto_ver=4)
+    if rc:
+        return rc
+    rc = do_test(proto_ver=5)
+    return rc
+
+if __name__ == "__main__":
+    sys.exit(all_test())

test/broker/07-will-delay-invalid-573191.py

@@ -8,7 +8,6 @@ from mosq_test_helper import *
 def do_test():
     rc = 1
 
-    mid = 1
     props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_WILL_DELAY_INTERVAL, 3)
     connect_packet = mosq_test.gen_connect("will-573191-test", proto_ver=5, will_topic="", will_properties=props)
     connack_packet = b""
@@ -19,6 +18,7 @@ def do_test():
     try:
         sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=30, port=port)
         sock.close()
+    except BrokenPipeError:
         rc = 0
     finally:
         broker.terminate()
@@ -26,6 +26,6 @@ def do_test():
         (stdo, stde) = broker.communicate()
         if rc:
             print(stde.decode('utf-8'))
-            exit(rc)
+    return rc
 
-do_test()
+sys.exit(do_test())

test/broker/07-will-invalid-utf8.py

@@ -6,7 +6,6 @@ from mosq_test_helper import *
 
 def do_test(start_broker, proto_ver):
     rc = 1
-    mid = 53
     connect_packet = mosq_test.gen_connect("will-invalid-utf8", will_topic="will/invalid/utf8", proto_ver=proto_ver)
 
     b = list(struct.unpack("B"*len(connect_packet), connect_packet))
@@ -14,36 +13,31 @@ def do_test(start_broker, proto_ver):
     connect_packet = struct.pack("B"*len(b), *b)
 
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
     try:
         sock = mosq_test.do_client_connect(connect_packet, b"", timeout=30, port=port)
-        rc = 0
         sock.close()
-    except mosq_test.TestError:
-        pass
+    except BrokenPipeError:
+        rc = 0
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
-    rc = do_test(start_broker, proto_ver=5)
-    if rc:
-        return rc;
-    return 0
+        return rc
+    return do_test(start_broker, proto_ver=5)
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

test/broker/07-will-no-flag.py

@@ -16,36 +16,31 @@ def do_test(start_broker, proto_ver):
     connect_packet = struct.pack("B"*len(bmod), *bmod)
 
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
     try:
         sock = mosq_test.do_client_connect(connect_packet, b"", port=port)
         sock.close()
+    except BrokenPipeError:
         rc = 0
-    except mosq_test.TestError:
-        pass
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
-    rc = do_test(start_broker, proto_ver=5)
-    if rc:
-        return rc;
-    return 0
+        return rc
+    return do_test(start_broker, proto_ver=5)
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

test/broker/07-will-null-topic.py

@@ -7,43 +7,36 @@ from mosq_test_helper import *
 def do_test(start_broker, proto_ver):
     rc = 1
     connect_packet = mosq_test.gen_connect("will-null-topic", will_topic="", will_payload=struct.pack("!4sB7s", b"will", 0, b"message"), proto_ver=proto_ver)
-    connack_packet = mosq_test.gen_connack(rc=2, proto_ver=proto_ver)
 
     port = mosq_test.get_port()
+    broker = None
     if start_broker:
         broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
 
     try:
         sock = mosq_test.do_client_connect(connect_packet, b"", timeout=30, port=port)
-        rc = 0
         sock.close()
-    except socket.error as e:
-        if e.errno == errno.ECONNRESET:
-            # Connection has been closed by peer, this is the expected behaviour
-            rc = 0
-    except mosq_test.TestError:
-        pass
+    except BrokenPipeError:
+        rc = 0
     finally:
-        if start_broker:
+        if broker:
             broker.terminate()
             broker.wait()
             (stdo, stde) = broker.communicate()
             if rc:
                 print(stde.decode('utf-8'))
                 print("proto_ver=%d" % (proto_ver))
-                exit(rc)
-        else:
-            return rc
+    return rc
 
 
 def all_tests(start_broker=False):
     rc = do_test(start_broker, proto_ver=4)
     if rc:
-        return rc;
+        return rc
     rc = do_test(start_broker, proto_ver=5)
     if rc:
-        return rc;
+        return rc
     return 0
 
 if __name__ == '__main__':
-    all_tests(True)
+    sys.exit(all_tests(True))

test/broker/09-extended-auth-single.py

@@ -54,9 +54,15 @@ connack5_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
 
 broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
 
+
 try:
-    sock = mosq_test.do_client_connect(connect1_packet, b"", timeout=20, port=port)
-    sock.close()
+    sock = None
+    try:
+        sock = mosq_test.do_client_connect(connect1_packet, b"", timeout=20, port=port)
+        sock.close()
+        rc = 2
+    except BrokenPipeError:
+        pass
 
     sock = mosq_test.do_client_connect(connect2_packet, connack2_packet, timeout=20, port=port)
     sock.close()
@@ -71,8 +77,6 @@ try:
     sock.close()
 
     rc = 0
-except mosq_test.TestError:
-    pass
 finally:
     os.remove(conf_file)
     broker.terminate()
@@ -82,5 +86,4 @@ finally:
         print(stde.decode('utf-8'))
 
 
-exit(rc)
-
+sys.exit(rc)

test/mosq_test.py

@@ -119,7 +119,7 @@ def expect_packet(sock, name, expected):
         while len(packet_recvd) < rlen:
             data = sock.recv(rlen-len(packet_recvd))
             if len(data) == 0:
-                break
+                raise BrokenPipeError(f"when reading {name} from {sock.getpeername()}")
             packet_recvd += data
     except socket.timeout:
         pass