Mirrors_Framework/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-02-05 18:50:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Big rejig of library tests

Browse Source 
Includes full consistency between C/C++ library tests
This commit is contained in:
Roger A. Light
2023-01-02 11:42:57 +00:00
parent 4199e7b2d3
commit 343a984083
163 changed files with 4518 additions and 3002 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
config.mk
View File

@@ -165,6 +165,7 @@ ifeq ($(UNAME),SunOS)
endif
else
CFLAGS?=-Wall -ggdb -O3 -Wconversion -Wextra -std=gnu99
CXXFLAGS?=-Wall -ggdb -O3 -Wconversion -Wextra
endif
STATIC_LIB_DEPS:=
@@ -237,10 +238,6 @@ else
LIB_CXXFLAGS:=$(LIB_CXXFLAGS) -fPIC
endif
ifneq ($(UNAME),SunOS)
LIB_LDFLAGS:=$(LIB_LDFLAGS) -Wl,--version-script=linker.version -Wl,-soname,libmosquitto.so.$(SOVERSION)
endif
ifeq ($(UNAME),QNX)
BROKER_LDADD:=$(BROKER_LDADD) -lsocket
LIB_LIBADD:=$(LIB_LIBADD) -lsocket
@@ -404,6 +401,7 @@ ifeq ($(WITH_COVERAGE),yes)
PLUGIN_CFLAGS:=$(PLUGIN_CFLAGS) -coverage
PLUGIN_LDFLAGS:=$(PLUGIN_LDFLAGS) -coverage
LIB_CFLAGS:=$(LIB_CFLAGS) -coverage
LIB_CXXFLAGS:=$(LIB_CXXFLAGS) -coverage
LIB_LDFLAGS:=$(LIB_LDFLAGS) -coverage
CLIENT_CFLAGS:=$(CLIENT_CFLAGS) -coverage
CLIENT_LDFLAGS:=$(CLIENT_LDFLAGS) -coverage

6
lib/Makefile
View File

@@ -1,6 +1,10 @@
R=..
include ${R}/config.mk
ifneq ($(UNAME),SunOS)
LOCAL_LDFLAGS:=-Wl,--version-script=linker.version -Wl,-soname,libmosquitto.so.$(SOVERSION)
endif
.PHONY : really clean install
MOSQ_OBJS=mosquitto.o \
@@ -105,7 +109,7 @@ clean :
$(MAKE) -C cpp clean
libmosquitto.so.${SOVERSION} : ${MOSQ_OBJS}
${CROSS_COMPILE}$(CC) -shared $(LIB_LDFLAGS) $^ -o $@ ${LIB_LIBADD}
${CROSS_COMPILE}$(CC) -shared $(LIB_LDFLAGS) $(LOCAL_LDFLAGS) $^ -o $@ ${LIB_LIBADD}
libmosquitto.a : ${MOSQ_OBJS}
${CROSS_COMPILE}$(AR) cr $@ $^

5
lib/cpp/Makefile
View File

@@ -2,7 +2,7 @@ R=../..
include ${R}/config.mk
ifneq ($(UNAME),SunOS)
LIB_LDFLAGS:=$(LDFLAGS) -Wl,-soname,libmosquittopp.so.${SOVERSION}
LOCAL_LDFLAGS:=-Wl,-soname,libmosquittopp.so.${SOVERSION}
endif
.PHONY : clean install
@@ -39,11 +39,10 @@ clean :
-rm -f *.o libmosquittopp.so.${SOVERSION} libmosquittopp.a
libmosquittopp.so.${SOVERSION} : mosquittopp.o
${CROSS_COMPILE}$(CXX) -shared $(LIB_LDFLAGS) $< -o $@ ../libmosquitto.so.${SOVERSION} $(LIB_LIDADD)
${CROSS_COMPILE}$(CXX) -shared $(LOCAL_LDFLAGS) $(LIB_LDFLAGS) $< -o $@ ../libmosquitto.so.${SOVERSION} $(LIB_LIDADD)
libmosquittopp.a : mosquittopp.o
${CROSS_COMPILE}$(AR) cr $@ $^
mosquittopp.o : mosquittopp.cpp ${R}/include/mosquittopp.h
${CROSS_COMPILE}$(CXX) $(LIB_CPPFLAGS) $(LIB_CXXFLAGS) -c $< -o $@

12
test/broker/09-plugin-evt-psk-key.py
View File

@@ -28,8 +28,8 @@ def do_test(per_listener_settings):
'-m', 'psk-test',
'--psk', '159445',
'--psk-identity', 'badidentity',
'-p']
bad_client = mosq_test.start_client(filename=sys.argv[0].replace('/', '-'), cmd=bad_client_args, port=port)
'-p', str(port)]
bad_client = mosq_test.start_client(filename=sys.argv[0].replace('/', '-'), cmd=bad_client_args)
bad_client.wait()
if bad_client.returncode == 0:
raise ValueError('bad client should have failed')
@@ -41,16 +41,16 @@ def do_test(per_listener_settings):
'--psk', '159445',
'--psk-identity', 'subidentity',
'-v', '-N',
'-p']
sub_client = mosq_test.start_client(filename=sys.argv[0].replace('/', '-'), cmd=sub_client_args, port=port)
'-p', str(port)]
sub_client = mosq_test.start_client(filename=sys.argv[0].replace('/', '-'), cmd=sub_client_args)
pub_client_args = [mosq_test.get_build_root() + '/client/mosquitto_pub',
'-t', 'plugin/psk/test',
'-m', 'psk-test',
'--psk', '297A49',
'--psk-identity', 'pubidentity',
'-p']
pub_client = mosq_test.start_client(filename=sys.argv[0].replace('/', '-'), cmd=pub_client_args, port=port)
'-p', str(port)]
pub_client = mosq_test.start_client(filename=sys.argv[0].replace('/', '-'), cmd=pub_client_args)
pub_client.wait()
sub_client.wait()

1
test/broker/CMakeLists.txt
View File

@@ -1,6 +1,5 @@
add_subdirectory(c)
file(GLOB PY_TEST_FILES [0-9][0-9]-*.py)
file(GLOB PY_PERSIST_TEST_FILES 15-*.py)

6
test/client/02-subscribe-env.py
View File

@@ -17,7 +17,7 @@ def do_test(proto_ver, env):
V = 'mqttv31'
env['LD_LIBRARY_PATH'] = mosq_test.get_build_root() + '/lib'
cmd = ['../../client/mosquitto_sub',
cmd = [mosq_test.get_build_root() + '/client/mosquitto_sub',
'-p', str(port),
'-q', '1',
'-V', V,
@@ -63,12 +63,12 @@ def do_test(proto_ver, env):
exit(rc)
env = {'HOME': mosq_test.get_build_root() + '/test/client/data'}
env = {'HOME': str(source_dir / 'data')}
do_test(proto_ver=3, env=env)
do_test(proto_ver=4, env=env)
do_test(proto_ver=5, env=env)
env = {'XDG_CONFIG_HOME': mosq_test.get_build_root() + '/test/client/data/.config'}
env = {'XDG_CONFIG_HOME': str(source_dir / 'data/.config')}
do_test(proto_ver=3, env=env)
do_test(proto_ver=4, env=env)
do_test(proto_ver=5, env=env)

6
test/client/03-publish-env.py
View File

@@ -18,7 +18,7 @@ def do_test(proto_ver, env):
env['LD_LIBRARY_PATH'] = mosq_test.get_build_root() + '/lib'
cmd = ['../../client/mosquitto_pub',
cmd = [mosq_test.get_build_root() + '/client/mosquitto_pub',
'-p', str(port),
'-q', '1',
'-V', V
@@ -62,12 +62,12 @@ def do_test(proto_ver, env):
exit(rc)
env = {'HOME': mosq_test.get_build_root() + '/test/client/data'}
env = {'HOME': str(source_dir / 'data')}
do_test(proto_ver=3, env=env)
do_test(proto_ver=4, env=env)
do_test(proto_ver=5, env=env)
env = {'XDG_CONFIG_HOME': mosq_test.get_build_root() + '/test/client/data/.config'}
env = {'XDG_CONFIG_HOME': str(source_dir / 'data/.config')}
do_test(proto_ver=3, env=env)
do_test(proto_ver=4, env=env)
do_test(proto_ver=5, env=env)

6
test/client/04-rr-env.py
View File

@@ -19,7 +19,7 @@ def do_test(proto_ver, env):
env['LD_LIBRARY_PATH'] = mosq_test.get_build_root() + '/lib'
payload = "message"
cmd = ['../../client/mosquitto_rr',
cmd = [mosq_test.get_build_root() + '/client/mosquitto_rr',
'-p', str(port),
'-q', '1',
'-t', '04/rr/qos1/test/request',
@@ -76,12 +76,12 @@ def do_test(proto_ver, env):
exit(rc)
env = {'HOME': mosq_test.get_build_root() + '/test/client/data'}
env = {'HOME': str(source_dir / 'data')}
do_test(proto_ver=3, env=env)
do_test(proto_ver=4, env=env)
do_test(proto_ver=5, env=env)
env = {'XDG_CONFIG_HOME': mosq_test.get_build_root() + '/test/client/data/.config'}
env = {'XDG_CONFIG_HOME': str(source_dir / 'data/.config')}
do_test(proto_ver=3, env=env)
do_test(proto_ver=4, env=env)
do_test(proto_ver=5, env=env)

4
test/client/mosq_test_helper.py
View File

@@ -16,3 +16,7 @@ import struct
import subprocess
import time
import errno
from pathlib import Path
source_dir = Path(__file__).resolve().parent

41
test/lib/01-con-discon-success-v5.py
View File

@@ -10,42 +10,17 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MAXIMUM_PACKET_SIZE, 1000)
props += mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 20)
connect_packet = mosq_test.gen_connect("01-con-discon-success-v5", proto_ver=5, properties=props)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
rc = 1
keepalive = 60
props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MAXIMUM_PACKET_SIZE, 1000)
props += mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 20)
connect_packet = mosq_test.gen_connect("01-con-discon-success-v5", keepalive=keepalive, proto_ver=5, properties=props)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-con-discon-success-v5.test", [], do_test, None)
mosq_test.client_test("cpp/01-con-discon-success-v5.test", [], do_test, None)

39
test/lib/01-con-discon-success.py
View File

@@ -10,40 +10,13 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-con-discon-success", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-con-discon-success")
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-con-discon-success.test", [], do_test, None)
mosq_test.client_test("cpp/01-con-discon-success.test", [], do_test, None)

37
test/lib/01-con-discon-will-clear.py
View File

@@ -4,40 +4,15 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-con-discon-will")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-con-discon-will", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
disconnect_packet = mosq_test.gen_disconnect()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-con-discon-will-clear.test", [], do_test, None)
mosq_test.client_test("cpp/01-con-discon-will-clear.test", [], do_test, None)

40
test/lib/01-con-discon-will-v5.py
View File

@@ -4,41 +4,15 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_PAYLOAD_FORMAT_INDICATOR, 0x01)
connect_packet = mosq_test.gen_connect("01-con-discon-will", keepalive=keepalive, will_topic="will/topic", will_payload=b"will-payload", will_qos=1, will_retain=True, will_properties=props, proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_PAYLOAD_FORMAT_INDICATOR, 0x01)
connect_packet = mosq_test.gen_connect("01-con-discon-will", will_topic="will/topic", will_payload=b"will-payload", will_qos=1, will_retain=True, will_properties=props, proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-con-discon-will-v5.test", [], do_test, None)
mosq_test.client_test("cpp/01-con-discon-will-v5.test", [], do_test, None)

38
test/lib/01-con-discon-will.py
View File

@@ -4,40 +4,14 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-con-discon-will", keepalive=keepalive, will_topic="will/topic", will_payload=b"will-payload", will_qos=1, will_retain=True)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-con-discon-will", will_topic="will/topic", will_payload=b"will-payload", will_qos=1, will_retain=True)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-con-discon-will.test", [], do_test, None)
mosq_test.client_test("cpp/01-con-discon-will.test", [], do_test, None)

41
test/lib/01-keepalive-pingreq.py
View File

@@ -9,28 +9,13 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
keepalive = 5
connect_packet = mosq_test.gen_connect("01-keepalive-pingreq", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 5
connect_packet = mosq_test.gen_connect("01-keepalive-pingreq", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
pingreq_packet = mosq_test.gen_pingreq()
pingresp_packet = mosq_test.gen_pingresp()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(keepalive+10)
pingreq_packet = mosq_test.gen_pingreq()
pingresp_packet = mosq_test.gen_pingresp()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
@@ -39,17 +24,7 @@ try:
conn.send(pingresp_packet)
mosq_test.expect_packet(conn, "pingreq", pingreq_packet)
conn.close()
rc = 0
except mosq_test.TestError:
pass
finally:
#client.terminate()
if mosq_test.wait_for_subprocess(client, timeout=20):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-keepalive-pingreq.test", [], do_test, None)
mosq_test.client_test("cpp/01-keepalive-pingreq.test", [], do_test, None)

33
test/lib/01-no-clean-session.py
View File

@@ -7,36 +7,11 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-no-clean-session", clean_session=False, keepalive=keepalive)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-no-clean-session", clean_session=False)
mosq_test.expect_packet(conn, "connect", connect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-no-clean-session.test", [], do_test, None)
mosq_test.client_test("cpp/01-no-clean-session.test", [], do_test, None)

33
test/lib/01-pre-connect-callback.py
View File

@@ -7,36 +7,11 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-pre-connect", keepalive=keepalive, username="uname", password=";'[08gn=#")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-pre-connect", username="uname", password=";'[08gn=#")
mosq_test.expect_packet(conn, "connect", connect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-pre-connect-callback.test", [], do_test, None)
mosq_test.client_test("cpp/01-pre-connect-callback.test", [], do_test, None)

44
test/lib/01-server-keepalive-pingreq.py
View File

@@ -6,31 +6,16 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
keepalive = 60
server_keepalive = 4
connect_packet = mosq_test.gen_connect("01-server-keepalive-pingreq", keepalive=keepalive, proto_ver=5)
rc = 1
keepalive = 60
server_keepalive = 4
connect_packet = mosq_test.gen_connect("01-server-keepalive-pingreq", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_SERVER_KEEP_ALIVE, server_keepalive)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_SERVER_KEEP_ALIVE, server_keepalive)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
pingreq_packet = mosq_test.gen_pingreq()
pingresp_packet = mosq_test.gen_pingresp()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(server_keepalive+10)
pingreq_packet = mosq_test.gen_pingreq()
pingresp_packet = mosq_test.gen_pingresp()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
@@ -39,16 +24,7 @@ try:
conn.send(pingresp_packet)
mosq_test.expect_packet(conn, "pingreq", pingreq_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-server-keepalive-pingreq.test", [], do_test, None)
mosq_test.client_test("cpp/01-server-keepalive-pingreq.test", [], do_test, None)

33
test/lib/01-unpwd-set.py
View File

@@ -7,36 +7,11 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-unpwd-set", keepalive=keepalive, username="uname", password=";'[08gn=#")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-unpwd-set", username="uname", password=";'[08gn=#")
mosq_test.expect_packet(conn, "connect", connect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-unpwd-set.test", [], do_test, None)
mosq_test.client_test("cpp/01-unpwd-set.test", [], do_test, None)

33
test/lib/01-will-set.py
View File

@@ -9,36 +9,11 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-will-set", keepalive=keepalive, will_topic="topic/on/unexpected/disconnect", will_qos=1, will_retain=True, will_payload=b"will message")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-will-set", will_topic="topic/on/unexpected/disconnect", will_qos=1, will_retain=True, will_payload=b"will message")
mosq_test.expect_packet(conn, "connect", connect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-will-set.test", [], do_test, None)
mosq_test.client_test("cpp/01-will-set.test", [], do_test, None)

37
test/lib/01-will-unpwd-set.py
View File

@@ -9,38 +9,13 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("01-will-unpwd-set",
keepalive=keepalive, username="oibvvwqw", password="#'^2hg9a&nm38*us",
will_topic="will-topic", will_qos=2, will_payload=b"will message")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("01-will-unpwd-set",
username="oibvvwqw", password="#'^2hg9a&nm38*us",
will_topic="will-topic", will_qos=2, will_payload=b"will message")
mosq_test.expect_packet(conn, "connect", connect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/01-will-unpwd-set.test", [], do_test, None)
mosq_test.client_test("cpp/01-will-unpwd-set.test", [], do_test, None)

46
test/lib/02-subscribe-helper-qos2.py
View File

@@ -14,47 +14,25 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("subscribe-qos2-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("subscribe-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos2/test", 2)
suback_packet = mosq_test.gen_suback(mid, 2)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos2/test", 2)
suback_packet = mosq_test.gen_suback(mid, 2)
publish_packet = mosq_test.gen_publish("qos2/test", 0, "message")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
publish_packet = mosq_test.gen_publish("qos2/test", 0, "message")
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, subscribe_packet, suback_packet, "subscribe")
conn.send(publish_packet)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-subscribe-helper-simple-qos2.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-helper-simple-qos2.test", [], do_test, None)
mosq_test.client_test("c/02-subscribe-helper-callback-qos2.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-helper-callback-qos2.test", [], do_test, None)

42
test/lib/02-subscribe-qos0.py
View File

@@ -14,44 +14,20 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("subscribe-qos0-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("subscribe-qos0-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos0/test", 0)
suback_packet = mosq_test.gen_suback(mid, 0)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos0/test", 0)
suback_packet = mosq_test.gen_suback(mid, 0)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, subscribe_packet, suback_packet, "subscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-subscribe-qos0.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-qos0.test", [], do_test, None)

47
test/lib/02-subscribe-qos1.py
View File

@@ -14,45 +14,24 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("subscribe-qos1-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("subscribe-qos1-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos1/test", 1)
suback_packet = mosq_test.gen_suback(mid, 1)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos1/test", 1)
suback_packet = mosq_test.gen_suback(mid, 1)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, subscribe_packet, suback_packet, "subscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
print(client.stdout.read().decode('utf-8'))
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-subscribe-qos1.test", [], do_test, None)
mosq_test.client_test("c/02-subscribe-qos1-async1.test", [], do_test, None)
mosq_test.client_test("c/02-subscribe-qos1-async2.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-qos1.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-qos1-async1.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-qos1-async2.test", [], do_test, None)

43
test/lib/02-subscribe-qos2.py
View File

@@ -14,44 +14,19 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("subscribe-qos2-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("subscribe-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos2/test", 2)
suback_packet = mosq_test.gen_suback(mid, 2)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "qos2/test", 2)
suback_packet = mosq_test.gen_suback(mid, 2)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, subscribe_packet, suback_packet, "subscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-subscribe-qos2.test", [], do_test, None)
mosq_test.client_test("cpp/02-subscribe-qos2.test", [], do_test, None)

48
test/lib/02-unsubscribe-multiple-v5.py
View File

@@ -5,49 +5,25 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("unsubscribe-test", proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
keepalive = 60
connect_packet = mosq_test.gen_connect("unsubscribe-test", keepalive=keepalive, proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "unsubscribe/test", 2, proto_ver=5)
suback_packet = mosq_test.gen_suback(mid, 2, proto_ver=5)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "unsubscribe/test", 2, proto_ver=5)
suback_packet = mosq_test.gen_suback(mid, 2, proto_ver=5)
mid = 2
unsubscribe_packet = mosq_test.gen_unsubscribe_multiple(mid, ["unsubscribe/test", "no-sub"], proto_ver=5)
unsuback_packet = mosq_test.gen_unsuback(mid, reason_code=[0, 17], proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
rc = 1
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 2
unsubscribe_packet = mosq_test.gen_unsubscribe_multiple(mid, ["unsubscribe/test", "no-sub"], proto_ver=5)
unsuback_packet = mosq_test.gen_unsuback(mid, reason_code=[0, 17], proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, subscribe_packet, suback_packet, "subscribe")
mosq_test.do_receive_send(conn, unsubscribe_packet, unsuback_packet, "unsubscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-unsubscribe-multiple-v5.test", [], do_test, None)
# FIXME - missing func in lib mosq_test.client_test("cpp/02-unsubscribe-multiple-v5.test", [], do_test, None)

45
test/lib/02-unsubscribe-v5.py
View File

@@ -4,44 +4,23 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("unsubscribe-test", proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
keepalive = 60
connect_packet = mosq_test.gen_connect("unsubscribe-test", keepalive=keepalive, proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "unsubscribe/test", proto_ver=5, properties=props)
unsuback_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 1
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "unsubscribe/test", proto_ver=5, properties=props)
unsuback_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, unsubscribe_packet, unsuback_packet, "unsubscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-unsubscribe-v5.test", [], do_test, None)
mosq_test.client_test("cpp/02-unsubscribe-v5.test", [], do_test, None)
# FIXME - missing lib function mosq_test.client_test("c/02-unsubscribe2-v5.test", [], do_test, None)
# FIXME - missing lib function mosq_test.client_test("cpp/02-unsubscribe2-v5.test", [], do_test, None)

42
test/lib/02-unsubscribe.py
View File

@@ -4,44 +4,20 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("unsubscribe-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("unsubscribe-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "unsubscribe/test")
unsuback_packet = mosq_test.gen_unsuback(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 1
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "unsubscribe/test")
unsuback_packet = mosq_test.gen_unsuback(mid)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, unsubscribe_packet, unsuback_packet, "unsubscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/02-unsubscribe.test", [], do_test, None)
mosq_test.client_test("cpp/02-unsubscribe.test", [], do_test, None)

47
test/lib/02-unsubscribe2-v5.py
View File

@@ -1,47 +0,0 @@
#!/usr/bin/env python3
# Test whether a v5 client sends a correct UNSUBSCRIBE packet, and handles the UNSUBACK.
from mosq_test_helper import *
port = mosq_test.get_lib_port()
keepalive = 60
connect_packet = mosq_test.gen_connect("unsubscribe-test", keepalive=keepalive, proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, "unsubscribe/test", proto_ver=5, properties=props)
unsuback_packet = mosq_test.gen_unsuback(mid, proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, unsubscribe_packet, unsuback_packet, "unsubscribe")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)

56
test/lib/03-publish-b2c-qos1-unexpected-puback.py
View File

@@ -2,51 +2,23 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
keepalive = 5
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 5
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 13423
puback_packet = mosq_test.gen_puback(mid)
pingreq_packet = mosq_test.gen_pingreq()
mid = 13423
puback_packet = mosq_test.gen_puback(mid)
pingreq_packet = mosq_test.gen_pingreq()
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
conn.send(puback_packet)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
mosq_test.expect_packet(conn, "pingreq", pingreq_packet)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
if mosq_test.expect_packet(conn, "connect", connect_packet):
conn.send(connack_packet)
conn.send(puback_packet)
if mosq_test.expect_packet(conn, "pingreq", pingreq_packet):
rc = 0
conn.close()
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if rc != 0 or client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-b2c-qos1-unexpected-puback.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-b2c-qos1-unexpected-puback.test", [], do_test, None)

49
test/lib/03-publish-b2c-qos1.py
View File

@@ -13,50 +13,19 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos1-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 123
publish_packet = mosq_test.gen_publish("pub/qos1/receive", qos=1, mid=mid, payload="message")
puback_packet = mosq_test.gen_puback(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 123
publish_packet = mosq_test.gen_publish("pub/qos1/receive", qos=1, mid=mid, payload="message")
puback_packet = mosq_test.gen_puback(mid)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_send_receive(conn, publish_packet, puback_packet, "puback")
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-b2c-qos1.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-b2c-qos1.test", [], do_test, None)

79
test/lib/03-publish-b2c-qos2-len.py
View File

@@ -5,70 +5,47 @@
from mosq_test_helper import *
mid = 56
def len_test(test, pubrel_packet):
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
publish_packet = mosq_test.gen_publish("len/qos2/test", qos=2, mid=mid, payload="message", proto_ver=5)
pubrec_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubcomp_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
publish_packet = mosq_test.gen_publish("len/qos2/test", qos=2, mid=data['mid'], payload="message", proto_ver=5)
pubrec_packet = mosq_test.gen_pubrec(data['mid'], proto_ver=5)
pubcomp_packet = mosq_test.gen_pubcomp(data['mid'], proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
mosq_test.do_send_receive(conn, publish_packet, pubrec_packet, "pubrec")
mosq_test.do_send_receive(conn, pubrel_packet, pubcomp_packet, "pubcomp")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if rc != 0:
print(test)
exit(rc)
mosq_test.do_send_receive(conn, publish_packet, pubrec_packet, "pubrec")
mosq_test.do_send_receive(conn, data['pubrel_packet'], pubcomp_packet, "pubcomp")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
data = {}
data['mid'] = 56
# No reason code, no properties
pubrel_packet = mosq_test.gen_pubrel(mid)
len_test("qos2 len 2", pubrel_packet)
data['pubrel_packet'] = mosq_test.gen_pubrel(data['mid'])
data['label'] = "qos2 len 2"
mosq_test.client_test("c/03-publish-b2c-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-b2c-qos2-len.test", [], do_test, data)
# Reason code, no properties
pubrel_packet = mosq_test.gen_pubrel(mid, proto_ver=5, reason_code=0x00)
len_test("qos2 len 3", pubrel_packet)
data['pubrel_packet'] = mosq_test.gen_pubrel(data['mid'], proto_ver=5, reason_code=0x00)
data['label'] = "qos2 len 3"
mosq_test.client_test("c/03-publish-b2c-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-b2c-qos2-len.test", [], do_test, data)
# Reason code, empty properties
pubrel_packet = mosq_test.gen_pubrel(mid, proto_ver=5, reason_code=0x00, properties="")
len_test("qos2 len 4", pubrel_packet)
data['pubrel_packet'] = mosq_test.gen_pubrel(data['mid'], proto_ver=5, reason_code=0x00, properties="")
data['label'] = "qos2 len 4"
mosq_test.client_test("c/03-publish-b2c-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-b2c-qos2-len.test", [], do_test, data)
# Reason code, one property
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
pubrel_packet = mosq_test.gen_pubrel(mid, proto_ver=5, reason_code=0x00, properties=props)
len_test("qos2 len >5", pubrel_packet)
data['pubrel_packet'] = mosq_test.gen_pubrel(data['mid'], proto_ver=5, reason_code=0x00, properties=props)
data['label'] = "qos2 len >5"
mosq_test.client_test("c/03-publish-b2c-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-b2c-qos2-len.test", [], do_test, data)

56
test/lib/03-publish-b2c-qos2-unexpected-pubcomp.py
View File

@@ -2,51 +2,23 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
keepalive = 5
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 5
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 13423
pubcomp_packet = mosq_test.gen_pubcomp(mid)
pingreq_packet = mosq_test.gen_pingreq()
mid = 13423
pubcomp_packet = mosq_test.gen_pubcomp(mid)
pingreq_packet = mosq_test.gen_pingreq()
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
conn.send(pubcomp_packet)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
mosq_test.expect_packet(conn, "pingreq", pingreq_packet)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
if mosq_test.expect_packet(conn, "connect", connect_packet):
conn.send(connack_packet)
conn.send(pubcomp_packet)
if mosq_test.expect_packet(conn, "pingreq", pingreq_packet):
rc = 0
conn.close()
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if rc != 0 or client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-b2c-qos2-unexpected-pubcomp.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-b2c-qos2-unexpected-pubcomp.test", [], do_test, None)

74
test/lib/03-publish-b2c-qos2-unexpected-pubrel.py
View File

@@ -2,66 +2,38 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
pubrel_unexpected = mosq_test.gen_pubrel(1000)
pubcomp_unexpected = mosq_test.gen_pubcomp(1000)
pubrel_unexpected = mosq_test.gen_pubrel(1000)
pubcomp_unexpected = mosq_test.gen_pubcomp(1000)
mid = 13423
publish_packet = mosq_test.gen_publish("pub/qos2/receive", qos=2, mid=mid, payload="message")
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
mid = 13423
publish_packet = mosq_test.gen_publish("pub/qos2/receive", qos=2, mid=mid, payload="message")
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
publish_quit_packet = mosq_test.gen_publish("quit", qos=0, payload="quit")
publish_quit_packet = mosq_test.gen_publish("quit", qos=0, payload="quit")
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
conn.send(pubrel_unexpected)
mosq_test.expect_packet(conn, "pubcomp", pubcomp_unexpected)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
conn.send(publish_packet)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
if mosq_test.expect_packet(conn, "connect", connect_packet):
conn.send(connack_packet)
conn.send(pubrel_unexpected)
if mosq_test.expect_packet(conn, "pubcomp", pubcomp_unexpected):
conn.send(publish_packet)
if mosq_test.expect_packet(conn, "pubrec", pubrec_packet):
conn.send(pubrel_packet)
if mosq_test.expect_packet(conn, "pubcomp", pubcomp_packet):
conn.send(publish_quit_packet)
rc = 0
mosq_test.expect_packet(conn, "pubrec", pubrec_packet)
conn.send(pubrel_packet)
mosq_test.expect_packet(conn, "pubcomp", pubcomp_packet)
conn.send(publish_quit_packet)
conn.close()
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-b2c-qos2-unexpected-pubrel.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-b2c-qos2-unexpected-pubrel.test", [], do_test, None)

53
test/lib/03-publish-b2c-qos2.py
View File

@@ -18,53 +18,22 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 13423
publish_packet = mosq_test.gen_publish("pub/qos2/receive", qos=2, mid=mid, payload="message")
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 13423
publish_packet = mosq_test.gen_publish("pub/qos2/receive", qos=2, mid=mid, payload="message")
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_send_receive(conn, publish_packet, pubrec_packet, "pubrec")
mosq_test.do_send_receive(conn, pubrel_packet, pubcomp_packet, "pubcomp")
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-b2c-qos2.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-b2c-qos2.test", [], do_test, None)

85
test/lib/03-publish-c2b-qos1-disconnect.py
View File

@@ -4,55 +4,60 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(client_cmd):
port = mosq_test.get_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
connect_packet = mosq_test.gen_connect("publish-qos1-test")
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message")
publish_packet_dup = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", dup=True)
puback_packet = mosq_test.gen_puback(mid)
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message")
publish_packet_dup = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", dup=True)
puback_packet = mosq_test.gen_puback(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
client_args = [client_cmd, str(port)]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = mosq_test.get_build_root() + '/lib:' + mosq_test.get_build_root() + '/lib/cpp'
try:
(conn, address) = sock.accept()
conn.settimeout(15)
client = mosq_test.start_client(filename=client_cmd.replace('/', '-'), cmd=client_args, env=env)
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
try:
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "publish", publish_packet)
# Disconnect client. It should reconnect.
conn.close()
mosq_test.expect_packet(conn, "connect", connect_packet)
conn.send(connack_packet)
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "publish", publish_packet)
# Disconnect client. It should reconnect.
conn.close()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet_dup, puback_packet, "retried publish")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
(conn, address) = sock.accept()
conn.settimeout(15)
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet_dup, puback_packet, "retried publish")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
exit(rc)
conn.close()
except mosq_test.TestError:
pass
finally:
sock.close()
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
exit(1)
do_test("c/03-publish-c2b-qos1-disconnect.test")
do_test("cpp/03-publish-c2b-qos1-disconnect.test")

70
test/lib/03-publish-c2b-qos1-len.py
View File

@@ -5,64 +5,42 @@
from mosq_test_helper import *
def len_test(test, puback_packet):
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive, proto_ver=5)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos1-test", proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", proto_ver=5)
publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=data['mid'], payload="message", proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet, puback_packet, "publish")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if rc != 0:
print(test)
exit(rc)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet, data['puback_packet'], "publish")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
data = {}
data['mid'] = 1
# No reason code, no properties
puback_packet = mosq_test.gen_puback(1)
len_test("qos1 len 2", puback_packet)
data['puback_packet'] = mosq_test.gen_puback(data['mid'])
data['label'] = "qos1 len 2"
mosq_test.client_test("c/03-publish-c2b-qos1-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos1-len.test", [], do_test, data)
# Reason code, no properties
puback_packet = mosq_test.gen_puback(1, proto_ver=5, reason_code=0x00)
len_test("qos1 len 3", puback_packet)
data['puback_packet'] = mosq_test.gen_puback(data['mid'], proto_ver=5, reason_code=0x00)
data['label'] = "qos1 len 3"
mosq_test.client_test("c/03-publish-c2b-qos1-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos1-len.test", [], do_test, data)
# Reason code, empty properties
puback_packet = mosq_test.gen_puback(1, proto_ver=5, reason_code=0x00, properties="")
len_test("qos1 len 4", puback_packet)
data['puback_packet'] = mosq_test.gen_puback(data['mid'], proto_ver=5, reason_code=0x00, properties="")
data['label'] = "qos1 len 4"
mosq_test.client_test("c/03-publish-c2b-qos1-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos1-len.test", [], do_test, data)
# Reason code, one property
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
puback_packet = mosq_test.gen_puback(1, proto_ver=5, reason_code=0x00, properties=props)
len_test("qos1 len >5", puback_packet)
data['puback_packet'] = mosq_test.gen_puback(data['mid'], proto_ver=5, reason_code=0x00, properties=props)
data['label'] = "qos1 len >5"
mosq_test.client_test("c/03-publish-c2b-qos1-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos1-len.test", [], do_test, data)

83
test/lib/03-publish-c2b-qos1-receive-maximum.py
View File

@@ -5,56 +5,38 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos1-test", proto_ver=5)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 3)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 3)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_1_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_1_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_2_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_2_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 3
publish_3_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_3_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 3
publish_3_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_3_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 4
publish_4_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_4_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 4
publish_4_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_4_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 5
publish_5_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_5_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 5
publish_5_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_5_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 6
publish_6_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_6_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 6
publish_6_packet = mosq_test.gen_publish("topic", qos=1, mid=mid, payload="12345", proto_ver=5)
puback_6_packet = mosq_test.gen_puback(mid, proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
@@ -72,22 +54,7 @@ try:
conn.send(puback_4_packet)
conn.send(puback_5_packet)
conn.send(puback_6_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos1-receive-maximum.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos1-receive-maximum.test", [], do_test, None)

67
test/lib/03-publish-c2b-qos1-timeout.py
View File

@@ -1,67 +0,0 @@
#!/usr/bin/env python3
# Test whether a client sends a correct PUBLISH to a topic with QoS 1 and responds to a delay.
# The client should connect to port 1888 with keepalive=60, clean session set,
# and client id publish-qos1-test
# The test will send a CONNACK message to the client with rc=0. Upon receiving
# the CONNACK the client should verify that rc==0. If not, it should exit with
# return code=1.
# On a successful CONNACK, the client should send a PUBLISH message with topic
# "pub/qos1/test", payload "message" and QoS=1.
# The test will not respond to the first PUBLISH message, so the client must
# resend the PUBLISH message with dup=1. Note that to keep test durations low, a
# message retry timeout of less than 10 seconds is required for this test.
# On receiving the second PUBLISH message, the test will send the correct
# PUBACK response. On receiving the correct PUBACK response, the client should
# send a DISCONNECT message.
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos1-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message")
publish_packet_dup = mosq_test.gen_publish("pub/qos1/test", qos=1, mid=mid, payload="message", dup=True)
puback_packet = mosq_test.gen_puback(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "publish", publish_packet)
# Delay for > 3 seconds (message retry time)
mosq_test.do_receive_send(conn, publish_packet_dup, puback_packet, "dup publish")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)

103
test/lib/03-publish-c2b-qos2-disconnect.py
View File

@@ -4,68 +4,73 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(client_cmd):
port = mosq_test.get_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
connect_packet = mosq_test.gen_connect("publish-qos2-test")
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message")
publish_dup_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message", dup=True)
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message")
publish_dup_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message", dup=True)
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
client_args = [client_cmd, str(port)]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = mosq_test.get_build_root() + '/lib:' + mosq_test.get_build_root() + '/lib/cpp'
try:
(conn, address) = sock.accept()
conn.settimeout(10)
client = mosq_test.start_client(filename=client_cmd.replace('/', '-'), cmd=client_args, env=env)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mosq_test.expect_packet(conn, "publish", publish_packet)
# Disconnect client. It should reconnect.
conn.close()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "publish", publish_packet)
# Disconnect client. It should reconnect.
conn.close()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_dup_packet, pubrec_packet, "retried publish")
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "pubrel", pubrel_packet)
# Disconnect client. It should reconnect.
conn.close()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_dup_packet, pubrec_packet, "retried publish")
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "pubrel", pubrel_packet)
# Disconnect client. It should reconnect.
conn.close()
# Complete connection and message flow.
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, pubrel_packet, pubcomp_packet, "retried pubrel")
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
# Complete connection and message flow.
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, pubrel_packet, pubcomp_packet, "retried pubrel")
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
exit(rc)
conn.close()
except mosq_test.TestError:
pass
finally:
sock.close()
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
exit(1)
do_test("c/03-publish-c2b-qos2-disconnect.test")
do_test("cpp/03-publish-c2b-qos2-disconnect.test")

78
test/lib/03-publish-c2b-qos2-len.py
View File

@@ -5,71 +5,45 @@
from mosq_test_helper import *
def len_test(test, pubrec_packet, pubcomp_packet):
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message", proto_ver=5)
pubrel_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=data['mid'], payload="message", proto_ver=5)
pubrel_packet = mosq_test.gen_pubrel(data['mid'], proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(15)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet, pubrec_packet, "publish")
mosq_test.do_receive_send(conn, pubrel_packet, pubcomp_packet, "pubrel")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if rc != 0:
print(test)
exit(rc)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet, data['pubrec_packet'], "publish")
mosq_test.do_receive_send(conn, pubrel_packet, data['pubcomp_packet'], "pubrel")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
data = {}
data['mid'] = 1
# No reason code, no properties
pubrec_packet = mosq_test.gen_pubrec(1)
pubcomp_packet = mosq_test.gen_pubcomp(1)
len_test("qos2 len 2", pubrec_packet, pubcomp_packet)
data['pubrec_packet'] = mosq_test.gen_pubrec(data['mid'])
data['pubcomp_packet'] = mosq_test.gen_pubcomp(data['mid'])
mosq_test.client_test("c/03-publish-c2b-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos2-len.test", [], do_test, data)
# Reason code, no properties
pubrec_packet = mosq_test.gen_pubrec(1, proto_ver=5, reason_code=0x00)
pubcomp_packet = mosq_test.gen_pubcomp(1, proto_ver=5, reason_code=0x00)
len_test("qos2 len 3", pubrec_packet, pubcomp_packet)
data['pubrec_packet'] = mosq_test.gen_pubrec(data['mid'], proto_ver=5, reason_code=0x00)
data['pubcomp_packet'] = mosq_test.gen_pubcomp(data['mid'], proto_ver=5, reason_code=0x00)
mosq_test.client_test("c/03-publish-c2b-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos2-len.test", [], do_test, data)
# Reason code, empty properties
pubrec_packet = mosq_test.gen_pubrec(1, proto_ver=5, reason_code=0x00, properties="")
pubcomp_packet = mosq_test.gen_pubcomp(1, proto_ver=5, reason_code=0x00, properties="")
len_test("qos2 len 4", pubrec_packet, pubcomp_packet)
data['pubrec_packet'] = mosq_test.gen_pubrec(data['mid'], proto_ver=5, reason_code=0x00, properties="")
data['pubcomp_packet'] = mosq_test.gen_pubcomp(data['mid'], proto_ver=5, reason_code=0x00, properties="")
mosq_test.client_test("c/03-publish-c2b-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos2-len.test", [], do_test, data)
# Reason code, one property
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
pubrec_packet = mosq_test.gen_pubrec(1, proto_ver=5, reason_code=0x00, properties=props)
data['pubrec_packet'] = mosq_test.gen_pubrec(data['mid'], proto_ver=5, reason_code=0x00, properties=props)
props = mqtt5_props.gen_string_pair_prop(mqtt5_props.PROP_USER_PROPERTY, "key", "value")
pubcomp_packet = mosq_test.gen_pubcomp(1, proto_ver=5, reason_code=0x00, properties=props)
len_test("qos2 len >5", pubrec_packet, pubcomp_packet)
data['pubcomp_packet'] = mosq_test.gen_pubcomp(data['mid'], proto_ver=5, reason_code=0x00, properties=props)
mosq_test.client_test("c/03-publish-c2b-qos2-len.test", [], do_test, data)
mosq_test.client_test("cpp/03-publish-c2b-qos2-len.test", [], do_test, data)

49
test/lib/03-publish-c2b-qos2-maximum-qos-0.py
View File

@@ -4,55 +4,24 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_MAXIMUM_QOS, 0)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_MAXIMUM_QOS, 0)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
publish_1_packet = mosq_test.gen_publish("maximum/qos/qos0", qos=0, payload="message", proto_ver=5)
publish_1_packet = mosq_test.gen_publish("maximum/qos/qos0", qos=0, payload="message", proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "publish 1", publish_1_packet)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos2-maximum-qos-0.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos2-maximum-qos-0.test", [], do_test, None)

55
test/lib/03-publish-c2b-qos2-maximum-qos-1.py
View File

@@ -4,61 +4,30 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_MAXIMUM_QOS, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
props = mqtt5_props.gen_byte_prop(mqtt5_props.PROP_MAXIMUM_QOS, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("maximum/qos/qos1", qos=1, mid=mid, payload="message", proto_ver=5)
puback_1_packet = mosq_test.gen_puback(mid, proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("maximum/qos/qos1", qos=1, mid=mid, payload="message", proto_ver=5)
puback_1_packet = mosq_test.gen_puback(mid, proto_ver=5)
publish_2_packet = mosq_test.gen_publish("maximum/qos/qos0", qos=0, payload="message", proto_ver=5)
publish_2_packet = mosq_test.gen_publish("maximum/qos/qos0", qos=0, payload="message", proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_1_packet, puback_1_packet, "publish 1")
mosq_test.expect_packet(conn, "publish 2", publish_2_packet)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos2-maximum-qos-1.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos2-maximum-qos-1.test", [], do_test, None)

65
test/lib/03-publish-c2b-qos2-pubrec-error.py
View File

@@ -6,66 +6,33 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="rejected", proto_ver=5)
pubrec_1_packet = mosq_test.gen_pubrec(mid, proto_ver=5, reason_code=0x80)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="rejected", proto_ver=5)
pubrec_1_packet = mosq_test.gen_pubrec(mid, proto_ver=5, reason_code=0x80)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="accepted", proto_ver=5)
pubrec_2_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_2_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_2_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="accepted", proto_ver=5)
pubrec_2_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_2_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_2_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_1_packet, pubrec_1_packet, "publish 1")
mosq_test.do_receive_send(conn, publish_2_packet, pubrec_2_packet, "publish 2")
mosq_test.do_receive_send(conn, pubrel_2_packet, pubcomp_2_packet, "pubrel 2")
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos2-pubrec-error.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos2-pubrec-error.test", [], do_test, None)

97
test/lib/03-publish-c2b-qos2-receive-maximum-1.py
View File

@@ -5,62 +5,43 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 1)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_1_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_1_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_1_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_1_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_1_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_1_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_2_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_2_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_2_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_2_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_2_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_2_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 3
publish_3_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_3_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_3_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_3_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 3
publish_3_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_3_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_3_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_3_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 4
publish_4_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_4_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_4_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_4_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 4
publish_4_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_4_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_4_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_4_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 5
publish_5_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_5_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_5_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_5_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 5
publish_5_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_5_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_5_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_5_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
@@ -79,22 +60,8 @@ try:
mosq_test.do_receive_send(conn, publish_5_packet, pubrec_5_packet, "publish 5")
mosq_test.do_receive_send(conn, pubrel_5_packet, pubcomp_5_packet, "pubrel 5")
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos2-receive-maximum.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos2-receive-maximum.test", [], do_test, None)

96
test/lib/03-publish-c2b-qos2-receive-maximum-2.py
View File

@@ -5,62 +5,43 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test", proto_ver=5)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive, proto_ver=5)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 2)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
props = mqtt5_props.gen_uint16_prop(mqtt5_props.PROP_RECEIVE_MAXIMUM, 2)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props, property_helper=False)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_1_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_1_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_1_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 1
publish_1_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_1_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_1_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_1_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_2_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_2_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_2_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 2
publish_2_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_2_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_2_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_2_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 3
publish_3_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_3_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_3_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_3_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 3
publish_3_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_3_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_3_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_3_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 4
publish_4_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_4_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_4_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_4_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 4
publish_4_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_4_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_4_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_4_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mid = 5
publish_5_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_5_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_5_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_5_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 5
publish_5_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="12345", proto_ver=5)
pubrec_5_packet = mosq_test.gen_pubrec(mid, proto_ver=5)
pubrel_5_packet = mosq_test.gen_pubrel(mid, proto_ver=5)
pubcomp_5_packet = mosq_test.gen_pubcomp(mid, proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
@@ -89,22 +70,9 @@ try:
mosq_test.expect_packet(conn, "pubrel 5", pubrel_5_packet)
conn.send(pubcomp_5_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
if client.returncode != 0:
exit(1)
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos2-receive-maximum.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos2-receive-maximum.test", [], do_test, None)

78
test/lib/03-publish-c2b-qos2-timeout.py
View File

@@ -1,78 +0,0 @@
#!/usr/bin/env python3
# Test whether a client sends a correct PUBLISH to a topic with QoS 1 and responds to a delay.
# The client should connect to port 1888 with keepalive=60, clean session set,
# and client id publish-qos2-test
# The test will send a CONNACK message to the client with rc=0. Upon receiving
# the CONNACK the client should verify that rc==0. If not, it should exit with
# return code=1.
# On a successful CONNACK, the client should send a PUBLISH message with topic
# "pub/qos2/test", payload "message" and QoS=2.
# The test will not respond to the first PUBLISH message, so the client must
# resend the PUBLISH message with dup=1. Note that to keep test durations low, a
# message retry timeout of less than 10 seconds is required for this test.
# On receiving the second PUBLISH message, the test will send the correct
# PUBREC response. On receiving the correct PUBREC response, the client should
# send a PUBREL message.
# The test will not respond to the first PUBREL message, so the client must
# resend the PUBREL message with dup=1. On receiving the second PUBREL message,
# the test will send the correct PUBCOMP response. On receiving the correct
# PUBCOMP response, the client should send a DISCONNECT message.
from mosq_test_helper import *
port = mosq_test.get_lib_port()
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message")
publish_dup_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message", dup=True)
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "publish", publish_packet)
# Delay for > 3 seconds (message retry time)
mosq_test.expect_packet(conn, "dup publish", publish_dup_packet)
conn.send(pubrec_packet)
mosq_test.expect_packet(conn, "pubrel", pubrel_packet)
mosq_test.expect_packet(conn, "dup pubrel", pubrel_packet)
conn.send(pubcomp_packet)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)

48
test/lib/03-publish-c2b-qos2.py
View File

@@ -22,48 +22,26 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos2-test")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos2-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
disconnect_packet = mosq_test.gen_disconnect()
disconnect_packet = mosq_test.gen_disconnect()
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message")
publish_dup_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message", dup=True)
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
mid = 1
publish_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message")
publish_dup_packet = mosq_test.gen_publish("pub/qos2/test", qos=2, mid=mid, payload="message", dup=True)
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, publish_packet, pubrec_packet, "publish")
mosq_test.do_receive_send(conn, pubrel_packet, pubcomp_packet, "pubrel")
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/03-publish-c2b-qos2.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-c2b-qos2.test", [], do_test, None)

31
test/lib/03-publish-loop.py Executable file
View File

@@ -0,0 +1,31 @@
#!/usr/bin/env python3
from mosq_test_helper import *
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("loop-test", proto_ver=5)
connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "loop/test", 0, proto_ver=5)
suback_packet = mosq_test.gen_suback(mid, 0, proto_ver=5)
disconnect_packet = mosq_test.gen_disconnect(proto_ver=5)
publish_packet = mosq_test.gen_publish("loop/test", qos=0, payload="message", proto_ver=5)
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.do_receive_send(conn, subscribe_packet, suback_packet, "subscribe")
conn.send(publish_packet)
mosq_test.expect_packet(conn, "publish", publish_packet)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
mosq_test.client_test("c/03-publish-loop.test", [], do_test, None)
mosq_test.client_test("c/03-publish-loop-forever.test", [], do_test, None)
mosq_test.client_test("c/03-publish-loop-manual.test", [], do_test, None)
mosq_test.client_test("c/03-publish-loop-start.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-loop.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-loop-forever.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-loop-manual.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-loop-start.test", [], do_test, None)

38
test/lib/03-publish-qos0-no-payload.py
View File

@@ -12,43 +12,21 @@
from mosq_test_helper import *
port = mosq_test.get_lib_port()
def do_test(conn, data):
connect_packet = mosq_test.gen_connect("publish-qos0-test-np")
connack_packet = mosq_test.gen_connack(rc=0)
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("publish-qos0-test-np", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
publish_packet = mosq_test.gen_publish("pub/qos0/no-payload/test", qos=0)
publish_packet = mosq_test.gen_publish("pub/qos0/no-payload/test", qos=0)
disconnect_packet = mosq_test.gen_disconnect()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.settimeout(10)
sock.bind(('', port))
sock.listen(5)
client_args = sys.argv[1:]
client = mosq_test.start_client(filename=sys.argv[1].replace('/', '-'), cmd=client_args, port=port)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
disconnect_packet = mosq_test.gen_disconnect()
mosq_test.do_receive_send(conn, connect_packet, connack_packet, "connect")
mosq_test.expect_packet(conn, "publish", publish_packet)
mosq_test.expect_packet(conn, "disconnect", disconnect_packet)
rc = 0
conn.close()
except mosq_test.TestError:
pass
finally:
if mosq_test.wait_for_subprocess(client):
print("test client not finished")
rc=1
sock.close()
exit(rc)
mosq_test.client_test("c/03-publish-qos0-no-payload.test", [], do_test, None)
mosq_test.client_test("cpp/03-publish-qos0-no-payload.test", [], do_test, None)

Some files were not shown because too many files have changed in this diff Show More