mirror of
https://github.com/eclipse-mosquitto/mosquitto.git
synced 2026-05-21 12:24:29 +02:00
Tests: Refactor client argv tests
This commit is contained in:
Roger Light
parent
517ea6072f
commit
f941f83f73
|
|
@ -7,26 +7,7 @@ from mosq_test_helper import *
|
|||
mosq_test.require_features(["WITH_CLIENTS", "WITH_TLS", "WITH_TLS_PSK"])
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_sub] + args
|
||||
|
||||
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(sub):
|
||||
print("sub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
(stdo, stde) = sub.communicate()
|
||||
if sub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(sub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
|
||||
client_run(mosq_paths.mosquitto_sub, args, stderr_expected, rc_expected)
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Missing args for TLS-PSK related options
|
||||
|
|
|
|||
|
|
@ -7,25 +7,7 @@ from mosq_test_helper import *
|
|||
mosq_test.require_features(["WITH_TLS"])
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_sub] + args
|
||||
|
||||
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(sub):
|
||||
print("sub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
(stdo, stde) = sub.communicate()
|
||||
if sub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(sub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_sub, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -5,25 +5,7 @@
|
|||
from mosq_test_helper import *
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_sub] + args
|
||||
|
||||
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(sub):
|
||||
print("sub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
(stdo, stde) = sub.communicate()
|
||||
if sub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(sub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_sub, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -7,25 +7,7 @@ from mosq_test_helper import *
|
|||
mosq_test.require_features(["WITH_CLIENTS", "WITH_TLS", "WITH_TLS_PSK"])
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_pub] + args
|
||||
|
||||
pub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(pub):
|
||||
print("pub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
(stdo, stde) = pub.communicate()
|
||||
if pub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(pub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_pub, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -7,25 +7,7 @@ from mosq_test_helper import *
|
|||
mosq_test.require_features(["WITH_TLS"])
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_pub] + args
|
||||
|
||||
pub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(pub):
|
||||
print("pub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
(stdo, stde) = pub.communicate()
|
||||
if pub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(pub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde.decode('utf-8'))
|
||||
client_run(mosq_paths.mosquitto_pub, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -5,25 +5,7 @@
|
|||
from mosq_test_helper import *
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_pub] + args
|
||||
|
||||
pub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(pub):
|
||||
print("pub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
(stdo, stde) = pub.communicate()
|
||||
if pub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(pub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_pub, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -7,24 +7,7 @@ from mosq_test_helper import *
|
|||
mosq_test.require_features(["WITH_CLIENTS", "WITH_TLS", "WITH_TLS_PSK"])
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
|
||||
cmd = [mosq_paths.mosquitto_rr] + args
|
||||
|
||||
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
sub.wait()
|
||||
(stdo, stde) = sub.communicate()
|
||||
if sub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(sub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_rr, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -7,23 +7,7 @@ from mosq_test_helper import *
|
|||
mosq_test.require_features(["WITH_TLS"])
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_rr] + args
|
||||
|
||||
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
sub.wait()
|
||||
(stdo, stde) = sub.communicate()
|
||||
if sub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(sub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_rr, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -5,24 +5,7 @@
|
|||
from mosq_test_helper import *
|
||||
|
||||
def do_test(args, stderr_expected, rc_expected):
|
||||
rc = 1
|
||||
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [mosq_paths.mosquitto_rr] + args
|
||||
|
||||
sub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
if mosq_test.wait_for_subprocess(sub):
|
||||
print("sub not terminated")
|
||||
raise mosq_test.TestError(1)
|
||||
if sub.returncode != rc_expected:
|
||||
raise mosq_test.TestError(sub.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in stde.decode('utf-8'):
|
||||
raise mosq_test.TestError(stde)
|
||||
client_run(mosq_paths.mosquitto_rr, args, stderr_expected, rc_expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -21,3 +21,18 @@ import errno
|
|||
from pathlib import Path
|
||||
|
||||
source_dir = Path(__file__).resolve().parent
|
||||
|
||||
def client_run(cmd_path, args, stderr_expected, rc_expected):
|
||||
port = mosq_test.get_port()
|
||||
|
||||
env = {
|
||||
'XDG_CONFIG_HOME':'/tmp/missing'
|
||||
}
|
||||
env = mosq_test.env_add_ld_library_path(env)
|
||||
cmd = [cmd_path] + args
|
||||
|
||||
client = subprocess.run(cmd, capture_output=True, text=True, env=env)
|
||||
if client.returncode != rc_expected:
|
||||
raise mosq_test.TestError(client.returncode)
|
||||
if stderr_expected is not None and stderr_expected not in client.stderr:
|
||||
raise mosq_test.TestError(f"Got:\n{client.stderr}\nExpected:\n{stderr_expected}")
|
||||
Loading…
Reference in a new issue