Simple Mock Example

[fractal-joe]

Hello,

I'm trying to write a simple mock example using the getting started code in the readme.

from unittest.mock import create_autospec
import paho.mqtt.client as mqtt
from paho.mqtt import publish

def on_connect(client, userdata, flags, rc):
    client.subscribe("test")

def on_message():
    pass

mock_on_message = create_autospec(on_message)

def test_pubsub():
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = mock_on_message

    client.connect('localhost')
    client.loop_start()
    publish.single('test', 'foo')
    mock_on_message.assert_called_once()
    client.loop_stop()

I'm getting AssertionError: Expected 'mock' to have been called once. Called 0 times.

Thanks for taking a look :)

[fractal-joe]

I think it has something to do with the loop? I got it working with js.

const mqtt = require('mqtt')
const client = mqtt.connect('mqtt://localhost')

client.on('connect', function () {
  client.subscribe('presence', function (err) {
    if (!err) {
      client.publish('presence', 'Hello mqtt')
    }
  })
})

test('pub sub', done => {
  function onMessage(topic, message) {
    expect(message.toString()).toBe('Hello mqtt');
    done();
  }

  client.on('message', onMessage);
});

[fractal-joe]

I added sleeps. same result :(

def test_pubsub():
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = mock_on_message

    client.connect('localhost')
    client.loop_start()
    time.sleep(2)
    publish.single('test', 'foo')
    time.sleep(2)
    mock_on_message.assert_called_once()
    client.loop_stop()
    client.disconnect()

[MattBrittan]

The below test runs fine (tweaked slightly as I'm running it with a broker under Docker). Given the long delay in responding (sorry about that!) I'm going to go ahead and close this (please feel free to reopen if this does not help - but please provide more info, i.e. logs).

from unittest.mock import create_autospec
import paho.mqtt.client as mqtt
from paho.mqtt import publish
import time

def on_connect(client, userdata, flags, rc):
    client.subscribe("test")

def on_message(client, userdata, msg):
    pass

mock_on_message = create_autospec(on_message)

def test_pubsub():
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = mock_on_message

    client.connect('mosquitto')
    client.loop_start()
    publish.single('test', 'foo', hostname="mosquitto")
    time.sleep(2)
    mock_on_message.assert_called_once()
    client.loop_stop()

if __name__ == "__main__":
    test_pubsub()
    print("Everything passed")

Note: This is part of an exercise to clean up old issues so that the project can move forwards. Due to the number of issues being worked through mistakes will be made; please feel free to reopen this issue (or comment) if you believe it's been closed in error.