adminandClaude Opus 5
79843aa2ae
Erster Stand der Hintergrundprozesse, die auf der Synology unter /volume1/homes/wagner/SolarManager laufen: der Manager selbst, die Sammler je Geraet, die MQTT-Bruecke, der Wecker und - neu hinzugezogen - der AutoAction-Runner, der als Hintergrundprozess hierher gehoert und nicht ins Web-Verzeichnis. Zugangsdaten stehen nicht mehr im Quelltext, sondern in config.ini, die nicht mit eingecheckt wird. Vorlage ist config.ini.example, gelesen wird sie von konfig.py. Betroffen waren solarManager.py (Datenbank und Wattpilot), zeit.py, gatherWaterData.py, wecker.py und skoda_testdaten.py, das sich das Passwort bisher aus dem Quelltext eines anderen Moduls herausgesucht hat. Die Kia-Anbindung ist mit dem Fahrzeug entfallen: kiaTest.py, gatherCarData.py und hyundai_kia_connect_api sind nicht mehr dabei, ebenso gatherInverterData.py, auf das nur noch eine auskommentierte Zeile zeigte. Die mitgelieferten Bibliotheken bleiben im Repository - die NAS hat kein pip, sie muessen neben den Skripten liegen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1411 lines
56 KiB
Python
1411 lines
56 KiB
Python
"""
|
|
*******************************************************************
|
|
Copyright (c) 2013, 2019 IBM Corp.
|
|
|
|
All rights reserved. This program and the accompanying materials
|
|
are made available under the terms of the Eclipse Public License v2.0
|
|
and Eclipse Distribution License v1.0 which accompany this distribution.
|
|
|
|
The Eclipse Public License is available at
|
|
http://www.eclipse.org/legal/epl-v20.html
|
|
and the Eclipse Distribution License is available at
|
|
http://www.eclipse.org/org/documents/edl-v10.php.
|
|
|
|
Contributors:
|
|
Ian Craggs - initial implementation and/or documentation
|
|
*******************************************************************
|
|
"""
|
|
|
|
import logging
|
|
import queue
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
import paho.mqtt
|
|
import paho.mqtt.client
|
|
from paho.mqtt.enums import CallbackAPIVersion
|
|
from paho.mqtt.packettypes import PacketTypes
|
|
from paho.mqtt.properties import Properties
|
|
from paho.mqtt.subscribeoptions import SubscribeOptions
|
|
|
|
DEFAULT_TIMEOUT = 5
|
|
# timeout for something that should not happen but we wait to
|
|
# give it time to happen if it does due to a bug.
|
|
WAIT_NON_EVENT_TIMEOUT = 1
|
|
|
|
class Callbacks:
|
|
|
|
def __init__(self):
|
|
self.messages = queue.Queue()
|
|
self.publisheds = queue.Queue()
|
|
self.subscribeds = queue.Queue()
|
|
self.unsubscribeds = queue.Queue()
|
|
self.disconnecteds = queue.Queue()
|
|
self.connecteds = queue.Queue()
|
|
self.conn_failures = queue.Queue()
|
|
|
|
def __str__(self):
|
|
return str(self.messages.queue) + str(self.messagedicts.queue) + str(self.publisheds.queue) + \
|
|
str(self.subscribeds.queue) + \
|
|
str(self.unsubscribeds.queue) + str(self.disconnects.queue)
|
|
|
|
def clear(self):
|
|
self.__init__()
|
|
|
|
def on_connect(self, client, userdata, flags, reasonCode, properties):
|
|
self.connecteds.put({"userdata": userdata, "flags": flags,
|
|
"reasonCode": reasonCode, "properties": properties})
|
|
|
|
def on_connect_fail(self, client, userdata):
|
|
self.conn_failures.put({"userdata": userdata})
|
|
|
|
def wait_connect_fail(self):
|
|
return self.conn_failures.get(timeout=10)
|
|
|
|
def wait_connected(self):
|
|
return self.connecteds.get(timeout=2)
|
|
|
|
def on_disconnect(self, client, userdata, reasonCode, properties=None):
|
|
self.disconnecteds.put(
|
|
{"reasonCode": reasonCode, "properties": properties})
|
|
|
|
def wait_disconnected(self):
|
|
return self.disconnecteds.get(timeout=2)
|
|
|
|
def on_message(self, client, userdata, message):
|
|
self.messages.put({"userdata": userdata, "message": message})
|
|
|
|
def published(self, client, userdata, msgid):
|
|
self.publisheds.put(msgid)
|
|
|
|
def wait_published(self):
|
|
return self.publisheds.get(timeout=2)
|
|
|
|
def on_subscribe(self, client, userdata, mid, reasonCodes, properties):
|
|
self.subscribeds.put({"mid": mid, "userdata": userdata,
|
|
"properties": properties, "reasonCodes": reasonCodes})
|
|
|
|
def wait_subscribed(self):
|
|
return self.subscribeds.get(timeout=2)
|
|
|
|
def unsubscribed(self, client, userdata, mid, properties, reasonCodes):
|
|
self.unsubscribeds.put({"mid": mid, "userdata": userdata,
|
|
"properties": properties, "reasonCodes": reasonCodes})
|
|
|
|
def wait_unsubscribed(self):
|
|
return self.unsubscribeds.get(timeout=2)
|
|
|
|
def on_log(self, client, userdata, level, buf):
|
|
print(buf)
|
|
|
|
def register(self, client):
|
|
client.on_connect = self.on_connect
|
|
client.on_subscribe = self.on_subscribe
|
|
client.on_publish = self.published
|
|
client.on_unsubscribe = self.unsubscribed
|
|
client.on_message = self.on_message
|
|
client.on_disconnect = self.on_disconnect
|
|
client.on_connect_fail = self.on_connect_fail
|
|
client.on_log = self.on_log
|
|
|
|
def get_messages(self, count: int, timeout: float = DEFAULT_TIMEOUT):
|
|
result = []
|
|
deadline = time.time() + timeout
|
|
while len(result) < count:
|
|
get_timeout = deadline - time.time()
|
|
if get_timeout <= 0:
|
|
result.append(self.messages.get_nowait())
|
|
else:
|
|
result.append(self.messages.get(timeout=get_timeout))
|
|
|
|
return result
|
|
|
|
def get_at_most_messages(self, count: int, timeout: float = DEFAULT_TIMEOUT):
|
|
result = []
|
|
deadline = time.time() + timeout
|
|
try:
|
|
while len(result) < count:
|
|
get_timeout = deadline - time.time()
|
|
if get_timeout <= 0:
|
|
result.append(self.messages.get_nowait())
|
|
else:
|
|
result.append(self.messages.get(timeout=get_timeout))
|
|
except queue.Empty:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
def cleanRetained(port):
|
|
callback = Callbacks()
|
|
curclient = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1,
|
|
b"clean retained",
|
|
protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
callback.register(curclient)
|
|
curclient.connect(host="localhost", port=port)
|
|
curclient.loop_start()
|
|
callback.wait_connected()
|
|
curclient.subscribe("#", options=SubscribeOptions(qos=0))
|
|
callback.wait_subscribed() # wait for retained messages to arrive
|
|
try:
|
|
while True:
|
|
message = callback.messages.get(timeout=WAIT_NON_EVENT_TIMEOUT)
|
|
if message["message"].payload != b"":
|
|
logging.info("deleting retained message for topic", message["message"])
|
|
curclient.publish(message["message"].topic, b"", 0, retain=True)
|
|
except queue.Empty:
|
|
pass
|
|
curclient.disconnect()
|
|
curclient.loop_stop()
|
|
|
|
|
|
def cleanup(port):
|
|
# clean all client state
|
|
print("clean up starting")
|
|
clientids = ("aclient", "bclient")
|
|
|
|
def _on_connect(client, *args):
|
|
client.disconnect()
|
|
|
|
for clientid in clientids:
|
|
curclient = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1,
|
|
clientid.encode("utf-8"),
|
|
protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
curclient.on_connect = _on_connect
|
|
curclient.connect(host="localhost", port=port, clean_start=True)
|
|
curclient.loop_forever()
|
|
|
|
# clean retained messages
|
|
cleanRetained(port)
|
|
print("clean up finished")
|
|
|
|
|
|
class Test(unittest.TestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
global callback, callback2, aclient, bclient
|
|
|
|
sys.path.append("paho.mqtt.testing/interoperability/")
|
|
try:
|
|
import mqtt.brokers
|
|
except ImportError as ie:
|
|
raise unittest.SkipTest("paho.mqtt.testing not present.") from ie
|
|
|
|
# Hack: we need to patch `signal.signal()` because `mqtt.brokers.run()`
|
|
# calls it to set up a signal handler; however, that won't work
|
|
# from a thread...
|
|
with unittest.mock.patch("signal.signal", unittest.mock.MagicMock()):
|
|
cls._test_broker = threading.Thread(
|
|
target=mqtt.brokers.run,
|
|
kwargs={
|
|
"config": ["listener 0"],
|
|
},
|
|
)
|
|
cls._test_broker.daemon = True
|
|
cls._test_broker.start()
|
|
# Wait a bit for TCP server to bind to an address
|
|
for _ in range(20):
|
|
time.sleep(0.1)
|
|
if mqtt.brokers.listeners.TCPListeners.server is not None:
|
|
port = mqtt.brokers.listeners.TCPListeners.server.socket.getsockname()[1]
|
|
if port != 0:
|
|
cls._test_broker_port = port
|
|
break
|
|
else:
|
|
raise ValueError("can't find the test broker port")
|
|
setData()
|
|
cleanup(cls._test_broker_port)
|
|
|
|
callback = Callbacks()
|
|
callback2 = Callbacks()
|
|
|
|
#aclient = mqtt_client.Client(b"\xEF\xBB\xBF" + "myclientid".encode("utf-8"))
|
|
#aclient = mqtt_client.Client("myclientid".encode("utf-8"))
|
|
aclient = paho.mqtt.client.Client(CallbackAPIVersion.VERSION1, b"aclient", protocol=paho.mqtt.client.MQTTv5)
|
|
callback.register(aclient)
|
|
|
|
bclient = paho.mqtt.client.Client(CallbackAPIVersion.VERSION1, b"bclient", protocol=paho.mqtt.client.MQTTv5)
|
|
callback2.register(bclient)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
# Another hack to stop the test broker... we rely on fact that it use a sockserver.TCPServer
|
|
import mqtt.brokers
|
|
mqtt.brokers.listeners.TCPListeners.server.shutdown()
|
|
cls._test_broker.join(5)
|
|
|
|
def test_basic(self):
|
|
import datetime
|
|
print(datetime.datetime.now(), "start")
|
|
aclient.connect(host="localhost", port=self._test_broker_port)
|
|
aclient.loop_start()
|
|
print(datetime.datetime.now(), "loop_start")
|
|
response = callback.wait_connected()
|
|
print(datetime.datetime.now(), "connected")
|
|
self.assertEqual(response["reasonCode"].getName(), "Success")
|
|
|
|
aclient.subscribe(topics[0], options=SubscribeOptions(qos=2))
|
|
response = callback.wait_subscribed()
|
|
print(datetime.datetime.now(), "wait_subscribed")
|
|
self.assertEqual(response["reasonCodes"][0].getName(), "Granted QoS 2")
|
|
|
|
aclient.publish(topics[0], b"qos 0")
|
|
aclient.publish(topics[0], b"qos 1", 1)
|
|
aclient.publish(topics[0], b"qos 2", 2)
|
|
|
|
msgs = callback.get_messages(3)
|
|
print(datetime.datetime.now(), "publish get")
|
|
got_payload = {
|
|
x["message"].payload
|
|
for x in msgs
|
|
}
|
|
|
|
self.assertEqual(got_payload, {b"qos 0", b"qos 1", b"qos 2"})
|
|
aclient.disconnect()
|
|
|
|
callback.clear()
|
|
aclient.loop_stop()
|
|
|
|
def test_connect_fail(self):
|
|
clientid = "connection failure"
|
|
|
|
fclient, fcallback = self.new_client(clientid)
|
|
|
|
fclient.user_data_set(1)
|
|
fclient.connect_async("localhost", 1)
|
|
response = fcallback.wait_connect_fail()
|
|
self.assertEqual(response["userdata"], 1)
|
|
fclient.loop_stop()
|
|
|
|
def test_retained_message(self):
|
|
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.UserProperty = ("a", "2")
|
|
publish_properties.UserProperty = ("c", "3")
|
|
|
|
# retained messages
|
|
callback.clear()
|
|
aclient.connect(host="localhost", port=self._test_broker_port)
|
|
aclient.loop_start()
|
|
response = callback.wait_connected()
|
|
aclient.publish(topics[1], b"qos 0", 0,
|
|
retain=True, properties=publish_properties)
|
|
aclient.publish(topics[2], b"qos 1", 1,
|
|
retain=True, properties=publish_properties)
|
|
aclient.publish(topics[3], b"qos 2", 2,
|
|
retain=True, properties=publish_properties)
|
|
# wait until those messages are published
|
|
time.sleep(WAIT_NON_EVENT_TIMEOUT)
|
|
aclient.subscribe(wildtopics[5], options=SubscribeOptions(qos=2))
|
|
response = callback.wait_subscribed()
|
|
self.assertEqual(response["reasonCodes"][0].getName(), "Granted QoS 2")
|
|
msgs = callback.get_messages(3)
|
|
|
|
aclient.disconnect()
|
|
aclient.loop_stop()
|
|
|
|
self.assertTrue(callback.messages.empty())
|
|
|
|
userprops = msgs[0]["message"].properties.UserProperty
|
|
self.assertTrue(userprops in [[("a", "2"), ("c", "3")], [
|
|
("c", "3"), ("a", "2")]], userprops)
|
|
userprops = msgs[1]["message"].properties.UserProperty
|
|
self.assertTrue(userprops in [[("a", "2"), ("c", "3")], [
|
|
("c", "3"), ("a", "2")]], userprops)
|
|
userprops = msgs[2]["message"].properties.UserProperty
|
|
self.assertTrue(userprops in [[("a", "2"), ("c", "3")], [
|
|
("c", "3"), ("a", "2")]], userprops)
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
|
|
cleanRetained(self._test_broker_port)
|
|
|
|
def test_will_message(self):
|
|
# will messages and keep alive
|
|
callback.clear()
|
|
callback2.clear()
|
|
self.assertTrue(callback2.messages.empty(), callback2.messages.queue)
|
|
|
|
will_properties = Properties(PacketTypes.WILLMESSAGE)
|
|
will_properties.WillDelayInterval = 0 # this is the default anyway
|
|
will_properties.UserProperty = ("a", "2")
|
|
will_properties.UserProperty = ("c", "3")
|
|
|
|
aclient.will_set(topics[2], payload=b"will message",
|
|
properties=will_properties)
|
|
|
|
aclient.connect(host="localhost", port=self._test_broker_port, keepalive=2)
|
|
aclient.loop_start()
|
|
response = callback.wait_connected()
|
|
bclient.connect(host="localhost", port=self._test_broker_port)
|
|
bclient.loop_start()
|
|
response = callback2.wait_connected()
|
|
bclient.subscribe(topics[2], qos=2)
|
|
response = callback2.wait_subscribed()
|
|
self.assertEqual(response["reasonCodes"][0].getName(), "Granted QoS 2")
|
|
|
|
# keep alive timeout ought to be triggered so the will message is received
|
|
aclient.loop_stop() # so that pings aren't sent
|
|
msg = callback2.messages.get(timeout=10)
|
|
bclient.disconnect()
|
|
bclient.loop_stop()
|
|
|
|
props = msg["message"].properties
|
|
self.assertEqual(props.UserProperty, [("a", "2"), ("c", "3")])
|
|
|
|
def test_zero_length_clientid(self):
|
|
logging.info("Zero length clientid test starting")
|
|
|
|
callback0 = Callbacks()
|
|
|
|
client0 = paho.mqtt.client.Client(CallbackAPIVersion.VERSION1, protocol=paho.mqtt.client.MQTTv5)
|
|
callback0.register(client0)
|
|
client0.loop_start()
|
|
# should not be rejected
|
|
client0.connect(host="localhost", port=self._test_broker_port, clean_start=False)
|
|
response = callback0.wait_connected()
|
|
self.assertEqual(response["reasonCode"].getName(), "Success")
|
|
self.assertTrue(
|
|
len(response["properties"].AssignedClientIdentifier) > 0)
|
|
client0.disconnect()
|
|
client0.loop_stop()
|
|
|
|
client0 = paho.mqtt.client.Client(CallbackAPIVersion.VERSION1, protocol=paho.mqtt.client.MQTTv5)
|
|
callback0.register(client0)
|
|
client0.loop_start()
|
|
client0.connect(host="localhost", port=self._test_broker_port) # should work
|
|
response = callback0.wait_connected()
|
|
self.assertEqual(response["reasonCode"].getName(), "Success")
|
|
self.assertTrue(
|
|
len(response["properties"].AssignedClientIdentifier) > 0)
|
|
client0.disconnect()
|
|
client0.loop_stop()
|
|
|
|
# when we supply a client id, we should not get one assigned
|
|
client0 = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1, "client0", protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
callback0.register(client0)
|
|
client0.loop_start()
|
|
client0.connect(host="localhost", port=self._test_broker_port) # should work
|
|
response = callback0.wait_connected()
|
|
self.assertEqual(response["reasonCode"].getName(), "Success")
|
|
self.assertFalse(
|
|
hasattr(response["properties"], "AssignedClientIdentifier"))
|
|
client0.disconnect()
|
|
client0.loop_stop()
|
|
|
|
def test_offline_message_queueing(self):
|
|
# message queueing for offline clients
|
|
cleanRetained(self._test_broker_port)
|
|
ocallback = Callbacks()
|
|
clientid = b"offline message queueing"
|
|
|
|
oclient = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1, clientid, protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
ocallback.register(oclient)
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.SessionExpiryInterval = 99999
|
|
oclient.loop_start()
|
|
oclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
ocallback.wait_connected()
|
|
oclient.subscribe(wildtopics[5], qos=2)
|
|
ocallback.wait_subscribed()
|
|
oclient.disconnect()
|
|
oclient.loop_stop()
|
|
|
|
bclient.loop_start()
|
|
bclient.connect(host="localhost", port=self._test_broker_port)
|
|
callback2.wait_connected()
|
|
msg1 = bclient.publish(topics[1], b"qos 0", 0)
|
|
msg2 = bclient.publish(topics[2], b"qos 1", 1)
|
|
msg3 = bclient.publish(topics[3], b"qos 2", 2)
|
|
|
|
msg1.wait_for_publish()
|
|
msg2.wait_for_publish()
|
|
msg3.wait_for_publish()
|
|
|
|
bclient.disconnect()
|
|
bclient.loop_stop()
|
|
|
|
oclient = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1, clientid, protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
ocallback.register(oclient)
|
|
oclient.loop_start()
|
|
oclient.connect(host="localhost", port=self._test_broker_port, clean_start=False)
|
|
ocallback.wait_connected()
|
|
|
|
msgs = ocallback.get_at_most_messages(3)
|
|
|
|
oclient.disconnect()
|
|
oclient.loop_stop()
|
|
|
|
self.assertTrue(len(msgs) in [
|
|
2, 3], ocallback.messages.qsize())
|
|
logging.info("This server %s queueing QoS 0 messages for offline clients" %
|
|
("is" if len(msgs) == 3 else "is not"))
|
|
|
|
def test_overlapping_subscriptions(self):
|
|
# overlapping subscriptions. When there is more than one matching subscription for the same client for a topic,
|
|
# the server may send back one message with the highest QoS of any matching subscription, or one message for
|
|
# each subscription with a matching QoS.
|
|
ocallback = Callbacks()
|
|
clientid = b"overlapping subscriptions"
|
|
|
|
oclient = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1, clientid, protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
ocallback.register(oclient)
|
|
|
|
oclient.loop_start()
|
|
oclient.connect(host="localhost", port=self._test_broker_port)
|
|
ocallback.wait_connected()
|
|
oclient.subscribe([(wildtopics[6], SubscribeOptions(qos=2)),
|
|
(wildtopics[0], SubscribeOptions(qos=1))])
|
|
ocallback.wait_subscribed()
|
|
oclient.publish(topics[3], b"overlapping topic filters", 2)
|
|
ocallback.wait_published()
|
|
|
|
msgs = ocallback.get_at_most_messages(2)
|
|
if len(msgs) == 1:
|
|
logging.info(
|
|
"This server is publishing one message for all matching overlapping subscriptions, not one for each.")
|
|
self.assertEqual(
|
|
msgs[0]["message"].qos, 2, msgs[0]["message"].qos)
|
|
else:
|
|
logging.info(
|
|
"This server is publishing one message per each matching overlapping subscription.")
|
|
self.assertTrue((msgs[0]["message"].qos == 2 and msgs[1]["message"].qos == 1) or
|
|
(msgs[0]["message"].qos == 1 and msgs[1]["message"].qos == 2), msgs)
|
|
oclient.disconnect()
|
|
oclient.loop_stop()
|
|
ocallback.clear()
|
|
|
|
def test_subscribe_failure(self):
|
|
# Subscribe failure. A new feature of MQTT 3.1.1 is the ability to send back negative responses to subscribe
|
|
# requests. One way of doing this is to subscribe to a topic which is not allowed to be subscribed to.
|
|
logging.info("Subscribe failure test starting")
|
|
|
|
ocallback = Callbacks()
|
|
clientid = b"subscribe failure"
|
|
oclient = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1, clientid, protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
ocallback.register(oclient)
|
|
oclient.loop_start()
|
|
oclient.connect(host="localhost", port=self._test_broker_port)
|
|
ocallback.wait_connected()
|
|
oclient.subscribe(nosubscribe_topics[0], qos=2)
|
|
response = ocallback.wait_subscribed()
|
|
|
|
self.assertEqual(response["reasonCodes"][0].getName(), "Unspecified error",
|
|
f"return code should be 0x80 {response['reasonCodes'][0].getName()}")
|
|
oclient.disconnect()
|
|
oclient.loop_stop()
|
|
|
|
def test_unsubscribe(self):
|
|
callback2.clear()
|
|
bclient.connect(host="localhost", port=self._test_broker_port)
|
|
bclient.loop_start()
|
|
callback2.wait_connected()
|
|
bclient.subscribe(topics[0], qos=2)
|
|
callback2.wait_subscribed()
|
|
bclient.subscribe(topics[1], qos=2)
|
|
callback2.wait_subscribed()
|
|
bclient.subscribe(topics[2], qos=2)
|
|
callback2.wait_subscribed()
|
|
time.sleep(1) # wait for any retained messages, hopefully
|
|
# Unsubscribe from one topic
|
|
bclient.unsubscribe(topics[0])
|
|
callback2.wait_unsubscribed()
|
|
callback2.clear() # if there were any retained messages
|
|
|
|
aclient.connect(host="localhost", port=self._test_broker_port)
|
|
aclient.loop_start()
|
|
callback.wait_connected()
|
|
aclient.publish(topics[0], b"topic 0 - unsubscribed", 1, retain=False)
|
|
aclient.publish(topics[1], b"topic 1", 1, retain=False)
|
|
aclient.publish(topics[2], b"topic 2", 1, retain=False)
|
|
|
|
msgs = callback2.get_messages(2)
|
|
|
|
bclient.disconnect()
|
|
bclient.loop_stop()
|
|
aclient.disconnect()
|
|
aclient.loop_stop()
|
|
self.assertEqual(len(msgs), 2)
|
|
|
|
def new_client(self, clientid):
|
|
callback = Callbacks()
|
|
client = paho.mqtt.client.Client(
|
|
CallbackAPIVersion.VERSION1,
|
|
clientid.encode("utf-8"),
|
|
protocol=paho.mqtt.client.MQTTv5,
|
|
)
|
|
callback.register(client)
|
|
client.loop_start()
|
|
return client, callback
|
|
|
|
def test_session_expiry(self):
|
|
# no session expiry property == never expire
|
|
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.SessionExpiryInterval = 0 # expire immediately
|
|
|
|
clientid = "session expiry"
|
|
|
|
eclient, ecallback = self.new_client(clientid)
|
|
|
|
eclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = ecallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
eclient.subscribe(topics[0], qos=2)
|
|
ecallback.wait_subscribed()
|
|
eclient.disconnect()
|
|
ecallback.wait_disconnected()
|
|
eclient.loop_stop()
|
|
|
|
fclient, fcallback = self.new_client(clientid)
|
|
|
|
# session should immediately expire
|
|
fclient.connect_async(host="localhost", port=self._test_broker_port, clean_start=False,
|
|
properties=connect_properties)
|
|
connack = fcallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
fclient.disconnect()
|
|
fcallback.wait_disconnected()
|
|
|
|
connect_properties.SessionExpiryInterval = 5
|
|
|
|
eclient, ecallback = self.new_client(clientid)
|
|
|
|
eclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = ecallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
eclient.subscribe(topics[0], qos=2)
|
|
ecallback.wait_subscribed()
|
|
eclient.disconnect()
|
|
ecallback.wait_disconnected()
|
|
eclient.loop_stop()
|
|
|
|
time.sleep(2)
|
|
# session should still exist
|
|
fclient, fcallback = self.new_client(clientid)
|
|
fclient.connect(host="localhost", port=self._test_broker_port, clean_start=False,
|
|
properties=connect_properties)
|
|
connack = fcallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], True)
|
|
fclient.disconnect()
|
|
fcallback.wait_disconnected()
|
|
fclient.loop_stop()
|
|
|
|
time.sleep(6)
|
|
# session should not exist
|
|
fclient, fcallback = self.new_client(clientid)
|
|
fclient.connect(host="localhost", port=self._test_broker_port, clean_start=False,
|
|
properties=connect_properties)
|
|
connack = fcallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
fclient.disconnect()
|
|
fcallback.wait_disconnected()
|
|
fclient.loop_stop()
|
|
|
|
eclient, ecallback = self.new_client(clientid)
|
|
connect_properties.SessionExpiryInterval = 1
|
|
connack = eclient.connect(
|
|
host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = ecallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
eclient.subscribe(topics[0], qos=2)
|
|
ecallback.wait_subscribed()
|
|
disconnect_properties = Properties(PacketTypes.DISCONNECT)
|
|
disconnect_properties.SessionExpiryInterval = 5
|
|
eclient.disconnect(properties=disconnect_properties)
|
|
ecallback.wait_disconnected()
|
|
eclient.loop_stop()
|
|
|
|
time.sleep(3)
|
|
# session should still exist as we changed the expiry interval on disconnect
|
|
fclient, fcallback = self.new_client(clientid)
|
|
fclient.connect(host="localhost", port=self._test_broker_port, clean_start=False,
|
|
properties=connect_properties)
|
|
connack = fcallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], True)
|
|
disconnect_properties.SessionExpiryInterval = 0
|
|
fclient.disconnect(properties=disconnect_properties)
|
|
fcallback.wait_disconnected()
|
|
fclient.loop_stop()
|
|
|
|
# session should immediately expire
|
|
fclient, fcallback = self.new_client(clientid)
|
|
fclient.connect(host="localhost", port=self._test_broker_port, clean_start=False,
|
|
properties=connect_properties)
|
|
connack = fcallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
fclient.disconnect()
|
|
fcallback.wait_disconnected()
|
|
fclient.loop_stop()
|
|
|
|
fclient.loop_stop()
|
|
eclient.loop_stop()
|
|
|
|
def test_user_properties(self):
|
|
clientid = "user properties"
|
|
uclient, ucallback = self.new_client(clientid)
|
|
uclient.loop_start()
|
|
uclient.connect(host="localhost", port=self._test_broker_port)
|
|
ucallback.wait_connected()
|
|
|
|
uclient.subscribe(topics[0], qos=2)
|
|
ucallback.wait_subscribed()
|
|
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.UserProperty = ("a", "2")
|
|
publish_properties.UserProperty = ("c", "3")
|
|
uclient.publish(topics[0], b"", 0, retain=False,
|
|
properties=publish_properties)
|
|
uclient.publish(topics[0], b"", 1, retain=False,
|
|
properties=publish_properties)
|
|
uclient.publish(topics[0], b"", 2, retain=False,
|
|
properties=publish_properties)
|
|
|
|
msgs = ucallback.get_messages(3)
|
|
|
|
uclient.disconnect()
|
|
ucallback.wait_disconnected()
|
|
uclient.loop_stop()
|
|
self.assertTrue(ucallback.messages.empty(), ucallback.messages.queue)
|
|
userprops = msgs[0]["message"].properties.UserProperty
|
|
self.assertTrue(userprops in [[("a", "2"), ("c", "3")], [
|
|
("c", "3"), ("a", "2")]], userprops)
|
|
userprops = msgs[1]["message"].properties.UserProperty
|
|
self.assertTrue(userprops in [[("a", "2"), ("c", "3")], [
|
|
("c", "3"), ("a", "2")]], userprops)
|
|
userprops = msgs[2]["message"].properties.UserProperty
|
|
self.assertTrue(userprops in [[("a", "2"), ("c", "3")], [
|
|
("c", "3"), ("a", "2")]], userprops)
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
|
|
def test_payload_format(self):
|
|
clientid = "payload format"
|
|
pclient, pcallback = self.new_client(clientid)
|
|
pclient.loop_start()
|
|
pclient.connect_async(host="localhost", port=self._test_broker_port)
|
|
pcallback.wait_connected()
|
|
|
|
pclient.subscribe(topics[0], qos=2)
|
|
pcallback.wait_subscribed()
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.PayloadFormatIndicator = 1
|
|
publish_properties.ContentType = "My name"
|
|
info = pclient.publish(
|
|
topics[0], b"qos 0", 0, retain=False, properties=publish_properties)
|
|
info.wait_for_publish()
|
|
info = pclient.publish(
|
|
topics[0], b"qos 1", 1, retain=False, properties=publish_properties)
|
|
info.wait_for_publish()
|
|
info = pclient.publish(
|
|
topics[0], b"qos 2", 2, retain=False, properties=publish_properties)
|
|
info.wait_for_publish()
|
|
|
|
msgs = pcallback.get_messages(3)
|
|
|
|
pclient.disconnect()
|
|
pcallback.wait_disconnected()
|
|
pclient.loop_stop()
|
|
|
|
self.assertTrue(pcallback.messages.empty(), pcallback.messages.queue)
|
|
props = msgs[0]["message"].properties
|
|
self.assertEqual(props.ContentType, "My name", props.ContentType)
|
|
self.assertEqual(props.PayloadFormatIndicator,
|
|
1, props.PayloadFormatIndicator)
|
|
props = msgs[1]["message"].properties
|
|
self.assertEqual(props.ContentType, "My name", props.ContentType)
|
|
self.assertEqual(props.PayloadFormatIndicator,
|
|
1, props.PayloadFormatIndicator)
|
|
props = msgs[2]["message"].properties
|
|
self.assertEqual(props.ContentType, "My name", props.ContentType)
|
|
self.assertEqual(props.PayloadFormatIndicator,
|
|
1, props.PayloadFormatIndicator)
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
|
|
def test_message_expiry(self):
|
|
clientid = "message expiry"
|
|
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.SessionExpiryInterval = 99999
|
|
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.loop_start()
|
|
lbclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
lbcallback.wait_connected()
|
|
lbclient.subscribe(topics[0], qos=2)
|
|
lbcallback.wait_subscribed()
|
|
disconnect_properties = Properties(PacketTypes.DISCONNECT)
|
|
disconnect_properties.SessionExpiryInterval = 999999999
|
|
lbclient.disconnect(properties=disconnect_properties)
|
|
lbcallback.wait_disconnected()
|
|
lbclient.loop_stop()
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.loop_start()
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.MessageExpiryInterval = 1
|
|
laclient.publish(topics[0], b"qos 1 - expire", 1,
|
|
retain=False, properties=publish_properties)
|
|
laclient.publish(topics[0], b"qos 2 - expire", 2,
|
|
retain=False, properties=publish_properties)
|
|
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.MessageExpiryInterval = 6
|
|
laclient.publish(topics[0], b"qos 1 - don't expire",
|
|
1, retain=False, properties=publish_properties)
|
|
laclient.publish(topics[0], b"qos 2 - don't expire",
|
|
2, retain=False, properties=publish_properties)
|
|
|
|
time.sleep(3)
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.loop_start()
|
|
lbclient.connect(host="localhost", port=self._test_broker_port, clean_start=False)
|
|
lbcallback.wait_connected()
|
|
|
|
msgs = lbcallback.get_messages(2)
|
|
|
|
self.assertTrue(lbcallback.messages.empty(), lbcallback.messages.queue)
|
|
self.assertTrue(msgs[0]["message"].properties.MessageExpiryInterval < 6,
|
|
msgs[0]["message"].properties.MessageExpiryInterval)
|
|
self.assertTrue(msgs[1]["message"].properties.MessageExpiryInterval < 6,
|
|
msgs[1]["message"].properties.MessageExpiryInterval)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
lbclient.disconnect()
|
|
lbcallback.wait_disconnected()
|
|
lbclient.loop_stop()
|
|
|
|
def test_subscribe_options(self):
|
|
# noLocal
|
|
clientid = 'subscribe options - noLocal'
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
laclient.subscribe(
|
|
topics[0], options=SubscribeOptions(qos=2, noLocal=True))
|
|
lacallback.wait_subscribed()
|
|
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.connect(host="localhost", port=self._test_broker_port)
|
|
lbcallback.wait_connected()
|
|
lbclient.loop_start()
|
|
lbclient.subscribe(
|
|
topics[0], options=SubscribeOptions(qos=2, noLocal=True))
|
|
lbcallback.wait_subscribed()
|
|
|
|
laclient.publish(topics[0], b"noLocal test", 1, retain=False)
|
|
|
|
lbcallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
try:
|
|
lacallback.messages.get(timeout=WAIT_NON_EVENT_TIMEOUT)
|
|
raise ValueError("unexpected message received")
|
|
except queue.Empty:
|
|
pass
|
|
|
|
self.assertTrue(lacallback.messages.empty(), lacallback.messages.queue)
|
|
self.assertTrue(lbcallback.messages.empty(), lbcallback.messages.queue)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
lbclient.disconnect()
|
|
lbcallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
lbclient.loop_stop()
|
|
|
|
# retainAsPublished
|
|
clientid = 'subscribe options - retain as published'
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
lacallback.wait_connected()
|
|
laclient.subscribe(topics[0], options=SubscribeOptions(
|
|
qos=2, retainAsPublished=True))
|
|
lacallback.wait_subscribed()
|
|
laclient.publish(
|
|
topics[0], b"retain as published false", 1, retain=False)
|
|
laclient.publish(
|
|
topics[0], b"retain as published true", 1, retain=True)
|
|
|
|
msgs = lacallback.get_messages(2)
|
|
|
|
self.assertTrue(lacallback.messages.empty(), lacallback.messages.queue)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
self.assertEqual(msgs[0]["message"].retain, False)
|
|
self.assertEqual(msgs[1]["message"].retain, True)
|
|
|
|
# retainHandling
|
|
clientid = 'subscribe options - retain handling'
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
lacallback.wait_connected()
|
|
laclient.publish(topics[1], b"qos 0", 0, retain=True)
|
|
laclient.publish(topics[2], b"qos 1", 1, retain=True)
|
|
laclient.publish(topics[3], b"qos 2", 2, retain=True)
|
|
time.sleep(1)
|
|
|
|
# retain handling 1 only gives us retained messages on a new subscription
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=1))
|
|
lacallback.wait_subscribed()
|
|
|
|
msgs = lacallback.get_messages(3)
|
|
|
|
self.assertTrue(lacallback.messages.empty())
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
lacallback.clear()
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=1))
|
|
lacallback.wait_subscribed()
|
|
time.sleep(1)
|
|
self.assertTrue(lacallback.messages.empty())
|
|
|
|
# remove that subscription
|
|
properties = Properties(PacketTypes.UNSUBSCRIBE)
|
|
properties.UserProperty = ("a", "2")
|
|
properties.UserProperty = ("c", "3")
|
|
laclient.unsubscribe(wildtopics[5], properties)
|
|
lacallback.wait_unsubscribed()
|
|
|
|
# check that we really did remove that subscription
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=1))
|
|
lacallback.wait_subscribed()
|
|
msgs = lacallback.get_messages(3)
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
lacallback.clear()
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=1))
|
|
lacallback.wait_subscribed()
|
|
time.sleep(WAIT_NON_EVENT_TIMEOUT)
|
|
self.assertTrue(lacallback.messages.empty())
|
|
|
|
# remove that subscription
|
|
properties = Properties(PacketTypes.UNSUBSCRIBE)
|
|
properties.UserProperty = ("a", "2")
|
|
properties.UserProperty = ("c", "3")
|
|
laclient.unsubscribe(wildtopics[5], properties)
|
|
lacallback.wait_unsubscribed()
|
|
|
|
lacallback.clear()
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=2))
|
|
lacallback.wait_subscribed()
|
|
self.assertTrue(lacallback.messages.empty())
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=2))
|
|
lacallback.wait_subscribed()
|
|
self.assertTrue(lacallback.messages.empty())
|
|
|
|
# remove that subscription
|
|
laclient.unsubscribe(wildtopics[5])
|
|
lacallback.wait_unsubscribed()
|
|
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=0))
|
|
lacallback.wait_subscribed()
|
|
msgs = lacallback.get_messages(3)
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
lacallback.clear()
|
|
laclient.subscribe(
|
|
wildtopics[5], options=SubscribeOptions(2, retainHandling=0))
|
|
msgs = lacallback.get_messages(3)
|
|
qoss = [x["message"].qos for x in msgs]
|
|
self.assertTrue(1 in qoss and 2 in qoss and 0 in qoss, qoss)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
cleanRetained(self._test_broker_port)
|
|
|
|
def test_subscription_identifiers(self):
|
|
clientid = 'subscription identifiers'
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
sub_properties = Properties(PacketTypes.SUBSCRIBE)
|
|
sub_properties.SubscriptionIdentifier = 456789
|
|
laclient.subscribe(topics[0], qos=2, properties=sub_properties)
|
|
lacallback.wait_subscribed()
|
|
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.connect(host="localhost", port=self._test_broker_port)
|
|
lbcallback.wait_connected()
|
|
lbclient.loop_start()
|
|
sub_properties = Properties(PacketTypes.SUBSCRIBE)
|
|
sub_properties.SubscriptionIdentifier = 2
|
|
lbclient.subscribe(topics[0], qos=2, properties=sub_properties)
|
|
lbcallback.wait_subscribed()
|
|
|
|
sub_properties.clear()
|
|
sub_properties.SubscriptionIdentifier = 3
|
|
lbclient.subscribe(f"{topics[0]}/#", qos=2, properties=sub_properties)
|
|
|
|
lbclient.publish(topics[0], b"sub identifier test", 1, retain=False)
|
|
|
|
msg = lacallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
self.assertTrue(lacallback.messages.empty(), lacallback.messages.queue)
|
|
self.assertEqual(msg["message"].properties.SubscriptionIdentifier[0],
|
|
456789, msg["message"].properties.SubscriptionIdentifier)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
msg = lbcallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
self.assertTrue(lbcallback.messages.empty(), lbcallback.messages.queue)
|
|
expected_subsids = {2, 3}
|
|
received_subsids = set(
|
|
msg["message"].properties.SubscriptionIdentifier)
|
|
self.assertEqual(received_subsids, expected_subsids, received_subsids)
|
|
lbclient.disconnect()
|
|
lbcallback.wait_disconnected()
|
|
lbclient.loop_stop()
|
|
|
|
def test_request_response(self):
|
|
clientid = 'request response'
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.connect(host="localhost", port=self._test_broker_port)
|
|
lbcallback.wait_connected()
|
|
lbclient.loop_start()
|
|
|
|
laclient.subscribe(
|
|
topics[0], options=SubscribeOptions(2, noLocal=True))
|
|
lacallback.wait_subscribed()
|
|
|
|
lbclient.subscribe(
|
|
topics[0], options=SubscribeOptions(2, noLocal=True))
|
|
lbcallback.wait_subscribed()
|
|
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.ResponseTopic = topics[0]
|
|
publish_properties.CorrelationData = b"334"
|
|
# client a is the requester
|
|
laclient.publish(topics[0], b"request", 1,
|
|
properties=publish_properties)
|
|
|
|
# client b is the responder
|
|
msg = lbcallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
self.assertEqual(msg["message"].properties.ResponseTopic, topics[0],
|
|
msg["message"].properties)
|
|
self.assertEqual(msg["message"].properties.CorrelationData, b"334",
|
|
msg["message"].properties)
|
|
|
|
lbclient.publish(msg["message"].properties.ResponseTopic, b"response", 1,
|
|
properties=msg["message"].properties)
|
|
|
|
# client a gets the response
|
|
lacallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
lbclient.disconnect()
|
|
lbcallback.wait_disconnected()
|
|
lbclient.loop_stop()
|
|
|
|
def test_client_topic_alias(self):
|
|
clientid = 'client topic alias'
|
|
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.TopicAliasMaximum = 0 # server topic aliases not allowed
|
|
connect_properties.SessionExpiryInterval = 99999
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = lacallback.wait_connected()
|
|
clientTopicAliasMaximum = 0
|
|
if hasattr(connack["properties"], "TopicAliasMaximum"):
|
|
clientTopicAliasMaximum = connack["properties"].TopicAliasMaximum
|
|
|
|
if clientTopicAliasMaximum == 0:
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
return
|
|
|
|
laclient.subscribe(topics[0], qos=2)
|
|
lacallback.wait_subscribed()
|
|
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.TopicAlias = 1
|
|
laclient.publish(topics[0], b"topic alias 1",
|
|
1, properties=publish_properties)
|
|
lacallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
|
|
laclient.publish("", b"topic alias 2", 1,
|
|
properties=publish_properties)
|
|
lacallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
|
|
laclient.disconnect() # should get rid of the topic aliases but not subscriptions
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
# check aliases have been deleted
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port, clean_start=False,
|
|
properties=connect_properties)
|
|
|
|
laclient.publish(topics[0], b"topic alias 3", 1)
|
|
lacallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
|
|
publish_properties = Properties(PacketTypes.PUBLISH)
|
|
publish_properties.TopicAlias = 1
|
|
laclient.publish("", b"topic alias 4", 1,
|
|
properties=publish_properties)
|
|
|
|
# should get back a disconnect with Topic alias invalid
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
def test_server_topic_alias(self):
|
|
clientid = 'server topic alias'
|
|
|
|
serverTopicAliasMaximum = 1 # server topic alias allowed
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.TopicAliasMaximum = serverTopicAliasMaximum
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
laclient.subscribe(topics[0], qos=2)
|
|
lacallback.wait_subscribed()
|
|
|
|
for qos in range(3):
|
|
laclient.publish(topics[0], b"topic alias 1", qos)
|
|
msgs = lacallback.get_messages(3)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
# first message should set the topic alias
|
|
self.assertTrue(hasattr(
|
|
msgs[0]["message"].properties, "TopicAlias"), msgs[0]["message"].properties)
|
|
topicalias = msgs[0]["message"].properties.TopicAlias
|
|
|
|
self.assertTrue(topicalias > 0)
|
|
self.assertEqual(msgs[0]["message"].topic, topics[0])
|
|
|
|
self.assertEqual(
|
|
msgs[1]["message"].properties.TopicAlias, topicalias)
|
|
self.assertEqual(msgs[1]["message"].topic, "")
|
|
|
|
self.assertEqual(
|
|
msgs[2]["message"].properties.TopicAlias, topicalias)
|
|
self.assertEqual(msgs[2]["message"].topic, "")
|
|
|
|
serverTopicAliasMaximum = 0 # no server topic alias allowed
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
# connect_properties.TopicAliasMaximum = serverTopicAliasMaximum # default is 0
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
laclient.subscribe(topics[0], qos=2)
|
|
lacallback.wait_subscribed()
|
|
|
|
for qos in range(3):
|
|
laclient.publish(topics[0], b"topic alias 2", qos)
|
|
msgs = lacallback.get_messages(3)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
# No topic aliases
|
|
self.assertFalse(hasattr(
|
|
msgs[0]["message"].properties, "TopicAlias"), msgs[0]["message"].properties)
|
|
self.assertFalse(hasattr(
|
|
msgs[1]["message"].properties, "TopicAlias"), msgs[1]["message"].properties)
|
|
self.assertFalse(hasattr(
|
|
msgs[2]["message"].properties, "TopicAlias"), msgs[2]["message"].properties)
|
|
|
|
serverTopicAliasMaximum = 0 # no server topic alias allowed
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.TopicAliasMaximum = serverTopicAliasMaximum # default is 0
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
laclient.subscribe(topics[0], qos=2)
|
|
lacallback.wait_subscribed()
|
|
|
|
for qos in range(3):
|
|
laclient.publish(topics[0], b"topic alias 3", qos)
|
|
msgs = lacallback.get_messages(3)
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
# No topic aliases
|
|
self.assertFalse(hasattr(
|
|
msgs[0]["message"].properties, "TopicAlias"), msgs[0]["message"].properties)
|
|
self.assertFalse(hasattr(
|
|
msgs[1]["message"].properties, "TopicAlias"), msgs[1]["message"].properties)
|
|
self.assertFalse(hasattr(
|
|
msgs[2]["message"].properties, "TopicAlias"), msgs[2]["message"].properties)
|
|
|
|
def test_maximum_packet_size(self):
|
|
clientid = 'maximum packet size'
|
|
|
|
# 1. server max packet size
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
connack = lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
serverMaximumPacketSize = 2**28-1
|
|
if hasattr(connack["properties"], "MaximumPacketSize"):
|
|
serverMaximumPacketSize = connack["properties"].MaximumPacketSize
|
|
|
|
if serverMaximumPacketSize < 65535:
|
|
# publish bigger packet than server can accept
|
|
payload = b"."*serverMaximumPacketSize
|
|
laclient.publish(topics[0], payload, 0)
|
|
# should get back a disconnect with packet size too big
|
|
response = lacallback.wait_disconnected()
|
|
self.assertEqual(response["reasonCode"].getName(),
|
|
"Packet too large", response["reasonCode"].getName())
|
|
else:
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
# 1. client max packet size
|
|
maximumPacketSize = 64 # max packet size we want to receive
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
connect_properties.MaximumPacketSize = maximumPacketSize
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
serverMaximumPacketSize = 2**28-1
|
|
if hasattr(connack["properties"], "MaximumPacketSize"):
|
|
serverMaximumPacketSize = connack["properties"].MaximumPacketSize
|
|
|
|
laclient.subscribe(topics[0], qos=2)
|
|
response = lacallback.wait_subscribed()
|
|
|
|
# send a small enough packet, should get this one back
|
|
payload = b"."*(int(maximumPacketSize/2))
|
|
laclient.publish(topics[0], payload, 0)
|
|
lacallback.messages.get(timeout=DEFAULT_TIMEOUT)
|
|
|
|
# send a packet too big to receive
|
|
payload = b"."*maximumPacketSize
|
|
laclient.publish(topics[0], payload, 1)
|
|
try:
|
|
lacallback.messages.get(timeout=WAIT_NON_EVENT_TIMEOUT)
|
|
raise ValueError("unexpected message received")
|
|
except queue.Empty:
|
|
pass
|
|
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
"""
|
|
def test_server_keep_alive(self):
|
|
clientid = 'server keep alive'
|
|
|
|
laclient, lacallback = self.new_client(clientid+" a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
connack = lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
self.assertTrue(hasattr(connack["properties"], "ServerKeepAlive"))
|
|
self.assertEqual(connack["properties"].ServerKeepAlive, 60)
|
|
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
"""
|
|
|
|
def test_will_delay(self):
|
|
# the will message should be received earlier than the session expiry
|
|
|
|
clientid = 'will delay'
|
|
|
|
will_properties = Properties(PacketTypes.WILLMESSAGE)
|
|
connect_properties = Properties(PacketTypes.CONNECT)
|
|
|
|
# set the will delay and session expiry to the same value -
|
|
# then both should occur at the same time
|
|
will_properties.WillDelayInterval = 3 # in seconds
|
|
connect_properties.SessionExpiryInterval = 5
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.will_set(
|
|
topics[0], payload=b"test_will_delay will message", properties=will_properties)
|
|
laclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = lacallback.wait_connected()
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
laclient.loop_start()
|
|
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.connect(host="localhost", port=self._test_broker_port, properties=connect_properties)
|
|
connack = lbcallback.wait_connected()
|
|
lbclient.loop_start()
|
|
# subscribe to will message topic
|
|
lbclient.subscribe(topics[0], qos=2)
|
|
lbcallback.wait_subscribed()
|
|
|
|
# abort client a and wait for the will message
|
|
laclient.loop_stop()
|
|
laclient.socket().close()
|
|
start = time.time()
|
|
msg = lbcallback.messages.get(DEFAULT_TIMEOUT)
|
|
duration = time.time() - start
|
|
self.assertAlmostEqual(duration, 4, delta=1)
|
|
self.assertEqual(msg["message"].topic, topics[0])
|
|
self.assertEqual(
|
|
msg["message"].payload, b"test_will_delay will message")
|
|
|
|
lbclient.disconnect()
|
|
lbcallback.wait_disconnected()
|
|
lbclient.loop_stop()
|
|
|
|
def test_shared_subscriptions(self):
|
|
clientid = 'shared subscriptions'
|
|
|
|
shared_sub_topic = f"$share/sharename/{topic_prefix}x"
|
|
shared_pub_topic = f"{topic_prefix}x"
|
|
|
|
laclient, lacallback = self.new_client(f"{clientid} a")
|
|
laclient.connect(host="localhost", port=self._test_broker_port)
|
|
connack = lacallback.wait_connected()
|
|
laclient.loop_start()
|
|
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
|
|
laclient.subscribe(
|
|
[(shared_sub_topic, SubscribeOptions(2)), (topics[0], SubscribeOptions(2))])
|
|
lacallback.wait_subscribed()
|
|
|
|
lbclient, lbcallback = self.new_client(f"{clientid} b")
|
|
lbclient.connect(host="localhost", port=self._test_broker_port)
|
|
connack = lbcallback.wait_connected()
|
|
lbclient.loop_start()
|
|
|
|
self.assertEqual(connack["reasonCode"].getName(), "Success")
|
|
self.assertEqual(connack["flags"]["session present"], False)
|
|
|
|
lbclient.subscribe(
|
|
[(shared_sub_topic, SubscribeOptions(2)), (topics[0], 2)])
|
|
lbcallback.wait_subscribed()
|
|
|
|
lacallback.clear()
|
|
lbcallback.clear()
|
|
|
|
count = 1
|
|
for i in range(count):
|
|
lbclient.publish(topics[0], f"message {i}", 0)
|
|
|
|
lacallback.get_messages(count)
|
|
lbcallback.get_messages(count)
|
|
|
|
self.assertTrue(lacallback.messages.empty())
|
|
self.assertTrue(lbcallback.messages.empty())
|
|
|
|
lacallback.clear()
|
|
lbcallback.clear()
|
|
|
|
for i in range(count):
|
|
lbclient.publish(shared_pub_topic, f"message {i}", 0)
|
|
# Each message should only be received once
|
|
result = []
|
|
deadline = time.time() + DEFAULT_TIMEOUT
|
|
while len(result) < count and time.time() < deadline:
|
|
get_timeout = deadline - time.time()
|
|
try:
|
|
if get_timeout <= 0:
|
|
result.append(lacallback.messages.get_nowait())
|
|
else:
|
|
result.append(lacallback.messages.get(timeout=get_timeout))
|
|
except queue.Empty:
|
|
# The message could be sent to other client, so empty queue
|
|
# could be normal
|
|
pass
|
|
|
|
try:
|
|
get_timeout = deadline - time.time()
|
|
if get_timeout <= 0:
|
|
result.append(lbcallback.messages.get_nowait())
|
|
else:
|
|
result.append(lbcallback.messages.get(timeout=get_timeout))
|
|
except queue.Empty:
|
|
# The message could be sent to other client, so empty queue
|
|
# could be normal
|
|
pass
|
|
|
|
self.assertEqual(
|
|
{x["message"].payload for x in result},
|
|
{f"message {i}".encode() for i in range(count)}
|
|
)
|
|
|
|
laclient.disconnect()
|
|
lacallback.wait_disconnected()
|
|
laclient.loop_stop()
|
|
|
|
lbclient.disconnect()
|
|
lbcallback.wait_disconnected()
|
|
lbclient.loop_stop()
|
|
|
|
|
|
def setData():
|
|
global topics, wildtopics, nosubscribe_topics, topic_prefix
|
|
topics = ("TopicA", "TopicA/B", "Topic/C", "TopicA/C", "/TopicA")
|
|
wildtopics = ("TopicA/+", "+/C", "#", "/#", "/+", "+/+", "TopicA/#")
|
|
nosubscribe_topics = ("test/nosubscribe",)
|
|
topic_prefix = "paho.mqtt.client.mqttv5/"
|