Add test for MQTTv5 connection to MQTTv3 broker

Author: PierreF

tests/test_client.py

@@ -3,6 +3,7 @@
 
 import paho.mqtt.client as client
 import pytest
+from paho.mqtt.reasoncodes import ReasonCodes
 
 import tests.paho_test as paho_test
 
@@ -92,6 +93,50 @@ def on_connect(mqttc, obj, flags, rc):
             mqttc.loop_stop()
 
 
+class Test_connect_v5:
+    """
+    Tests on connect/disconnect behaviour of the client with MQTTv5
+    """
+
+    def test_01_broker_no_support(self, fake_broker):
+        mqttc = client.Client(
+            "01-broker-no-support", protocol=client.MQTTv5)
+
+        def on_connect(mqttc, obj, flags, reason, properties):
+            assert reason == 132
+            assert reason == ReasonCodes(client.CONNACK >> 4, aName="Unsupported protocol version")
+            mqttc.disconnect()
+
+        mqttc.on_connect = on_connect
+
+        mqttc.connect_async("localhost", fake_broker.port)
+        mqttc.loop_start()
+
+        try:
+            fake_broker.start()
+
+            # Can't test the connect_packet, we can't yet generate MQTTv5 packet.
+            # connect_packet = paho_test.gen_connect(
+            #     "01-con-discon-success", keepalive=60,
+            #     proto_ver=client.MQTTv311)
+            packet_in = fake_broker.receive_packet(1000)
+            assert packet_in  # Check connection was not closed
+            # assert packet_in == connect_packet
+
+            # The reply packet is a MQTTv3 connack. But that the propose of this test,
+            # ensure client convert it to a reason code 132 "Unsupported protocol version"
+            connack_packet = paho_test.gen_connack(rc=1)
+            count = fake_broker.send_packet(connack_packet)
+            assert count  # Check connection was not closed
+            assert count == len(connack_packet)
+
+            packet_in = fake_broker.receive_packet(1)
+            assert not packet_in  # Check connection is closed
+
+        finally:
+            mqttc.loop_stop()
+
+
 class TestPublishBroker2Client:
 
     def test_invalid_utf8_topic(self, fake_broker):