github
/
mosquitto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

'mosquitto_pub -f' and 'mosquitto_pub -L' tests.

Browse Source
pull/2345/head
Roger A. Light 4 years ago
parent 57ef2be8d3
commit 5006943d9a
6 changed files with 274 additions and 5 deletions
Show Stats Download Patch File Download Diff File

69
test/client/03-publish-file-empty.py
Unescape Escape View File

@ -0,0 +1,69 @@
#!/usr/bin/env python3
#
from mosq_test_helper import *
def write_file(filename):
with open(filename, 'w') as f:
pass
def do_test(proto_ver):
rc = 1
port = mosq_test.get_port()
data_file = os.path.basename(__file__).replace('.py', '.data')
if proto_ver == 5:
V = 'mqttv5'
elif proto_ver == 4:
V = 'mqttv311'
else:
V = 'mqttv31'
env = {
'LD_LIBRARY_PATH':'../../lib',
'XDG_CONFIG_HOME':'/tmp/missing'
}
cmd = ['../../client/mosquitto_pub',
'-p', str(port),
'-q', '1',
'-t', '03/pub/file/empty/test',
'-f', data_file,
'-V', V
]
publish_packet = mosq_test.gen_publish("03/pub/file/empty/test", qos=0, payload="", proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
write_file(data_file)
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.wait()
mosq_test.expect_packet(sock, "publish", publish_packet)
rc = 0
sock.close()
except mosq_test.TestError:
pass
except Exception as e:
print(e)
finally:
os.remove(data_file)
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if rc:
print(stde.decode('utf-8'))
print("proto_ver=%d" % (proto_ver))
exit(rc)
do_test(proto_ver=3)
do_test(proto_ver=4)
do_test(proto_ver=5)

70
test/client/03-publish-file.py
Unescape Escape View File

@ -0,0 +1,70 @@
#!/usr/bin/env python3
#
from mosq_test_helper import *
def write_file(filename):
with open(filename, 'w') as f:
f.write("line1\n")
f.write("line2\n")
def do_test(proto_ver):
rc = 1
port = mosq_test.get_port()
data_file = os.path.basename(__file__).replace('.py', '.data')
if proto_ver == 5:
V = 'mqttv5'
elif proto_ver == 4:
V = 'mqttv311'
else:
V = 'mqttv31'
env = {
'LD_LIBRARY_PATH':'../../lib',
'XDG_CONFIG_HOME':'/tmp/missing'
}
cmd = ['../../client/mosquitto_pub',
'-p', str(port),
'-q', '1',
'-t', '03/pub/file/test',
'-f', data_file,
'-V', V
]
publish_packet = mosq_test.gen_publish("03/pub/file/test", qos=0, payload="line1\nline2\n", proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
write_file(data_file)
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.wait()
mosq_test.expect_packet(sock, "publish", publish_packet)
rc = 0
sock.close()
except mosq_test.TestError:
pass
except Exception as e:
print(e)
finally:
os.remove(data_file)
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if rc:
print(stde.decode('utf-8'))
print("proto_ver=%d" % (proto_ver))
exit(rc)
do_test(proto_ver=3)
do_test(proto_ver=4)
do_test(proto_ver=5)

64
test/client/03-publish-repeat.py
Unescape Escape View File

@ -0,0 +1,64 @@
#!/usr/bin/env python3
#
from mosq_test_helper import *
def do_test(proto_ver):
rc = 1
port = mosq_test.get_port()
if proto_ver == 5:
V = 'mqttv5'
elif proto_ver == 4:
V = 'mqttv311'
else:
V = 'mqttv31'
env = {
'LD_LIBRARY_PATH':'../../lib',
'XDG_CONFIG_HOME':'/tmp/missing'
}
cmd = ['../../client/mosquitto_pub',
'-p', str(port),
'-q', '1',
'-t', '03/pub/repeat/test',
'-m', 'message',
'-V', V,
'--repeat', '2',
'--repeat-delay', '0.1',
]
mid = 1
publish_packet = mosq_test.gen_publish("03/pub/repeat/test", qos=0, mid=mid, payload="message", proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
try:
sock = mosq_test.sub_helper(port=port, topic="#", qos=0, proto_ver=proto_ver)
pub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
pub.wait()
mosq_test.expect_packet(sock, "publish 1", publish_packet)
mosq_test.expect_packet(sock, "publish 2", publish_packet)
rc = 0
sock.close()
except mosq_test.TestError:
pass
except Exception as e:
print(e)
finally:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if rc:
print(stde.decode('utf-8'))
print("proto_ver=%d" % (proto_ver))
exit(rc)
do_test(proto_ver=3)
do_test(proto_ver=4)
do_test(proto_ver=5)

59
test/client/03-publish-url.py
Unescape Escape View File

@ -0,0 +1,59 @@
#!/usr/bin/env python3
#
from mosq_test_helper import *
def do_test(proto_ver):
rc = 1
port = mosq_test.get_port()
if proto_ver == 5:
V = 'mqttv5'
elif proto_ver == 4:
V = 'mqttv311'
else:
V = 'mqttv31'
env = {
'LD_LIBRARY_PATH':'../../lib',
'XDG_CONFIG_HOME':'/tmp/missing'
}
cmd = ['../../client/mosquitto_pub',
'-L', f'mqtt://localhost:{port}/03/pub/url/test',
'-m', 'message',
'-V', V
]
publish_packet = mosq_test.gen_publish("03/pub/url/test", qos=0, payload="message", proto_ver=proto_ver)
broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
try:
sock = mosq_test.sub_helper(port=port, topic="#", qos=0, proto_ver=proto_ver)
pub = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
pub.wait()
(stdo, stde) = pub.communicate()
mosq_test.expect_packet(sock, "publish", publish_packet)
rc = 0
sock.close()
except mosq_test.TestError:
pass
except Exception as e:
print(e)
finally:
broker.terminate()
broker.wait()
(stdo, stde) = broker.communicate()
if rc:
print(stde.decode('utf-8'))
print("proto_ver=%d" % (proto_ver))
exit(rc)
do_test(proto_ver=3)
do_test(proto_ver=4)
do_test(proto_ver=5)

3
test/client/Makefile
Unescape Escape View File

@ -17,6 +17,8 @@ test : 02 03
03 :
./03-publish-argv-errors.py
./03-publish-file-empty.py
./03-publish-file.py
./03-publish-options-file.py
./03-publish-qos0-empty.py
./03-publish-qos1-properties.py
@ -25,6 +27,7 @@ test : 02 03
./03-publish-socks.py
./03-publish-stdin-file.py
./03-publish-stdin-line.py
./03-publish-url.py,
ptest :
./test.sh

14
test/client/test.py
Unescape Escape View File

@ -6,21 +6,25 @@ import ptest
tests = [
#(ports required, 'path'),
(1, './02-subscribe-argv-errors.py'),
(1, './02-subscribe-filter-out.py'),
(1, './02-subscribe-filter-out.py'),
(1, './02-subscribe-format.py'),
(1, './02-subscribe-null.py'),
(1, './02-subscribe-qos1.py'),
(1, './02-subscribe-verbose.py'),
(1, './03-publish-argv-errors.py'),
(1, './03-publish-options-file.py'),
(1, './03-publish-file-empty.py'),
(1, './03-publish-file.py'),
(1, './03-publish-options-file.py'),
(1, './03-publish-qos0-empty.py'),
(1, './03-publish-qos1-properties.py'),
(1, './03-publish-qos1.py'),
(1, './03-publish-repeat.py'),
(1, './03-publish-repeat.py'),
(1, './03-publish-url.py'),
(2, './03-publish-socks.py'),
(1, './03-publish-stdin-file.py'),
(1, './03-publish-stdin-line.py'),
(1, './03-publish-stdin-file.py'),
(1, './03-publish-stdin-line.py'),
]
ptest.run_tests(tests)

Write Preview
Loading…
Cancel
Save