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 5

Tests: Use subprocess.run instead of subprocess.Popen where possible

Browse source
This commit is contained in:
Roger Light 2026-04-27 09:28:55 +01:00
parent 76c6dd3fe3
commit 81fc50e2be
4 changed files with 26 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
test/client/02-subscribe-env.py
View file

@ -27,8 +27,8 @@ def do_test(proto_ver, env):
]
payload = "message"
publish_packet_s = mosq_test.gen_publish("env/config/file/sub", qos=1, mid=1, payload=payload, proto_ver=proto_ver)
publish_packet_r = mosq_test.gen_publish("env/config/file/sub", qos=1, mid=2, payload=payload, proto_ver=proto_ver)
publish_packet_s = mosq_test.gen_publish("env/config/file/sub", qos=1, mid=1, payload=payload, proto_ver=proto_ver, retain=True)
publish_packet_r = mosq_test.gen_publish("env/config/file/sub", qos=1, mid=2, payload=payload, proto_ver=proto_ver, retain=True)
puback_packet_s = mosq_test.gen_puback(1, proto_ver=proto_ver)
puback_packet_r = mosq_test.gen_puback(2, proto_ver=proto_ver)
@ -36,18 +36,12 @@ def do_test(proto_ver, env):
try:
sock = mosq_test.pub_helper(port=port, proto_ver=proto_ver)
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
time.sleep(0.1)
sock.send(publish_packet_s)
mosq_test.expect_packet(sock, "puback", puback_packet_s)
sub_terminate_rc = 0
if mosq_test.wait_for_subprocess(sub):
print("sub not terminated")
sub_terminate_rc = 1
(stdo, stde) = sub.communicate()
if stdo.decode('utf-8') == payload + '\n':
rc = sub_terminate_rc
sub = subprocess.run(cmd, capture_output=True, text=True, env=env)
if payload in sub.stdout:
rc = sub.returncode
sock.close()
except mosq_test.TestError:
pass

30
test/client/02-subscribe-format.py
View file

@ -45,28 +45,28 @@ def do_test(format_str, expected_output, proto_ver=4, payload="message"):
props += mqtt5_props.gen_string_pair_prop(mqtt5_props.USER_PROPERTY, "name3", "value3")
props += mqtt5_props.gen_string_pair_prop(mqtt5_props.USER_PROPERTY, "name4", "value4")
if proto_ver == 5:
publish_packet = mosq_test.gen_publish("02/sub/format/test", qos=0, payload=payload, properties=props, proto_ver=proto_ver)
publish_packet = mosq_test.gen_publish("02/sub/format/test", qos=0, payload=payload, properties=props, proto_ver=proto_ver, retain=True)
else:
publish_packet = mosq_test.gen_publish("02/sub/format/test", qos=0, payload=payload, proto_ver=proto_ver)
publish_packet = mosq_test.gen_publish("02/sub/format/test", qos=0, payload=payload, proto_ver=proto_ver, retain=True)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
try:
sock = mosq_test.pub_helper(port=port, proto_ver=proto_ver)
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
time.sleep(0.1)
sock.send(publish_packet)
sub_terminate_rc = 0
if mosq_test.wait_for_subprocess(sub):
print("sub not terminated")
sub_terminate_rc = 1
(stdo, stde) = sub.communicate()
if stdo.decode('utf-8') == expected_output:
rc = sub_terminate_rc
else:
print("expected: (%d) %s" % (len(expected_output), expected_output))
print("actual: (%d) %s" % (len(stdo.decode('utf-8')), stdo.decode('utf-8')))
sub = subprocess.run(cmd, capture_output=True, text=True, env=env)
have_match = False
for expected_output in expected_outputs:
if sub.stdout.startswith(expected_output):
rc = sub.returncode
have_match = True
break
if have_match == False:
print(f"input: {format_str}")
print("expected: (%d) %s" % (len(expected_outputs), expected_outputs))
print("actual: (%d) %s" % (len(sub.stdout), sub.stdout))
sock.close()
except mosq_test.TestError:
pass

9
test/client/03-publish-env.py
View file

@ -37,15 +37,10 @@ def do_test(proto_ver, env):
try:
sock = mosq_test.sub_helper(port=port, topic="#", qos=1, proto_ver=proto_ver)
pub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
pub_terminate_rc = 0
if mosq_test.wait_for_subprocess(pub):
print("pub not terminated")
pub_terminate_rc = 1
(stdo, stde) = pub.communicate()
pub = subprocess.run(cmd, capture_output=True, env=env)
mosq_test.expect_packet(sock, "publish", publish_packet)
rc = pub_terminate_rc
rc = pub.returncode
sock.close()
except mosq_test.TestError:
pass

11
test/client/03-publish-file.py
View file

@ -47,14 +47,9 @@ def do_test(proto_ver):
try:
sock = mosq_test.sub_helper(port=port, topic="#", qos=0, proto_ver=proto_ver)
pub = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env)
pub_terminate_rc = 0
if mosq_test.wait_for_subprocess(pub):
print("pub not terminated")
pub_terminate_rc = 1
pub = subprocess.run(cmd, env=env)
mosq_test.expect_packet(sock, "publish", publish_packet)
rc = pub_terminate_rc
rc = pub.returncode
sock.close()
except mosq_test.TestError:
pass
@ -74,4 +69,4 @@ def do_test(proto_ver):
do_test(proto_ver=3)
do_test(proto_ver=4)
do_test(proto_ver=5)
do_test(proto_ver=5)