python unittest for paho-mqtt not working - simple syntax issue

Question

I am trying to write a simple test using the unittest package for python which simply detects if there is a broker connection. It seems to fail despite making a successful broker connection and I am 90% sure it's an issue with the syntax - specifically the definition of the has_connected boolean variable.

import paho.mqtt.client as mqtt
import time     

class TestBrokerConnection(unittest.TestCase):  
    def setUp(self):
        self.client = mqtt.Client("Test Client")
        self.client.on_connect = self.on_connect
        self.broker = "10.0.2.4"
        self.port = 1883
        self.has_connected = False
   
    def on_connect(client, userdata, flags, rc): #connect function
        if rc==0:
            self.has_connected = True
   
    def test_connection(self): #test to check connection to broker
        self.client.connect(self.broker, self.port)
        self.client.loop_start()
        time.sleep(2)
        self.client.loop_stop()
        self.assertTrue(self.has_connected)

if __name__ == '__main__':
    unittest.main()

Any help would be greatly appreciated :)

Answer 1

I copied your code example and used the example that paho.mqtt client gives to connect to

client.connect("mqtt.eclipse.org", 1883, 60)

I think that your issue may be on your on_connect function, you are referencing self.has_connected but you don't have a reference to self passed into the function.

This works on my side, let me know if adding self into that on_connect fixes the issue you're seeing!

class TestBrokerConnection(unittest.TestCase):
def setUp(self):
    self.client = mqtt.Client("Test Client")
    self.client.on_connect = self.on_connect
    self.broker = "mqtt.eclipse.org"
    self.port = 1883
    self.has_connected = False

def on_connect(self, client, userdata, flags, rc):  # connect function
    if rc == 0:
        self.has_connected = True

def test_connection(self):  # test to check connection to broker
    self.client.connect(self.broker, self.port)
    self.client.loop_start()
    time.sleep(2)
    self.client.loop_stop()
    self.assertTrue(self.has_connected)