nicolabs/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-05-21 12:24:29 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

throw BrokenPipeError if nothing received on sock

Browse source 
instead of ignoring this error, we throw an exception
the previous way might cause hard to find issues

Signed-off-by: Kai Buschulte <kai.buschulte@cedalo.com>
This commit is contained in:
Kai Buschulte 2022-06-20 11:18:36 +02:00
parent 546df9db85
commit a7304083f8
15 changed files with 120 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
test/broker/01-connect-global-max-clients.py
View file

@ -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())

9
test/broker/01-connect-global-max-connections.py
View file

@ -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())

9
test/broker/01-connect-max-connections.py
View file

@ -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())

27
test/broker/02-subpub-qos0-long-topic.py
View file

@ -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))

25
test/broker/02-subscribe-invalid-utf8.py
View file

@ -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))

24
test/broker/02-subscribe-long-topic.py
View file

@ -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))

25
test/broker/03-publish-invalid-utf8.py
View file

@ -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))

24
test/broker/03-publish-long-topic.py
View file

@ -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))

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

@ -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())

6
test/broker/07-will-delay-invalid-573191.py
View file

@ -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())

22
test/broker/07-will-invalid-utf8.py
View file

@ -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))

19
test/broker/07-will-no-flag.py
View file

@ -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))

23
test/broker/07-will-null-topic.py
View file

@ -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))

15
test/broker/09-extended-auth-single.py
View file

@ -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)

2
test/mosq_test.py
View file

@ -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