1#!/usr/bin/env python3 2 3# Test whether message size limits apply. 4 5from mosq_test_helper import * 6 7def write_config(filename, port): 8 with open(filename, 'w') as f: 9 f.write("listener %d\n" % (port)) 10 f.write("allow_anonymous true\n") 11 f.write("message_size_limit 1\n") 12 13def do_test(proto_ver): 14 rc = 1 15 mid = 53 16 keepalive = 60 17 connect_packet = mosq_test.gen_connect("subpub-qos0-test", keepalive=keepalive, proto_ver=proto_ver) 18 connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) 19 20 subscribe_packet = mosq_test.gen_subscribe(mid, "subpub/qos0", 0, proto_ver=proto_ver) 21 suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=proto_ver) 22 23 connect2_packet = mosq_test.gen_connect("subpub-qos0-helper", keepalive=keepalive, proto_ver=proto_ver) 24 connack2_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) 25 26 publish_packet_ok = mosq_test.gen_publish("subpub/qos0", qos=0, payload="A", proto_ver=proto_ver) 27 publish_packet_bad = mosq_test.gen_publish("subpub/qos0", qos=0, payload="AB", proto_ver=proto_ver) 28 29 port = mosq_test.get_port() 30 conf_file = os.path.basename(__file__).replace('.py', '.conf') 31 write_config(conf_file, port) 32 33 broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port) 34 35 try: 36 sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port) 37 mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") 38 39 sock2 = mosq_test.do_client_connect(connect2_packet, connack2_packet, timeout=20, port=port) 40 sock2.send(publish_packet_ok) 41 mosq_test.expect_packet(sock, "publish 1", publish_packet_ok) 42 43 # Check all is still well on the publishing client 44 mosq_test.do_ping(sock2) 45 46 sock2.send(publish_packet_bad) 47 48 # Check all is still well on the publishing client 49 mosq_test.do_ping(sock2) 50 51 # The subscribing client shouldn't have received a PUBLISH 52 mosq_test.do_ping(sock) 53 rc = 0 54 55 sock.close() 56 except SyntaxError: 57 raise 58 except TypeError: 59 raise 60 except mosq_test.TestError: 61 pass 62 finally: 63 os.remove(conf_file) 64 broker.terminate() 65 broker.wait() 66 (stdo, stde) = broker.communicate() 67 if rc: 68 print(stde.decode('utf-8')) 69 print("proto_ver=%d" % (proto_ver)) 70 exit(rc) 71 72 73do_test(proto_ver=4) 74do_test(proto_ver=5) 75exit(0) 76