asserting events

[eugeneniemand]

Hi,

I do this in my app

def mqtt_message_recieved_event(self, event_name, data, kwargs):  
        print(data)      
        payload = json.loads(data["payload"])
        message = ""
        for msg in payload["Messages"]:
            message += self.message_builder(msg)

        if message != "":
            self.fire_event("MQTT_PUBLISH", topic = "appdaemon/notification/tts", message = json.dumps({"message": message})) 
        else:
            self.log("Nothing to publish")

def initialize(self):
        self.log("NotitfyApp Init")
        self.set_namespace("mqtt")
        self.listen_event(self.mqtt_message_recieved_event, "MQTT_MESSAGE", topic = "homeassistant/notification/tts")
        self.set_namespace("hass")

How do I raise an event to test my mqtt_message_recieved_event method and assert that an event was fired?

Thanks

[eugeneniemand]

I have now got this in my test

def test_callbacks_are_registered(given_that, notification_engine, assert_that):
    assert_that(notification_engine) \
        .listens_to.event('MQTT_MESSAGE', topic = "homeassistant/notification/tts") \
        .with_callback(notification_engine.mqtt_message_recieved_event)

def test_during_night_light_turn_on(given_that, notification_engine, assert_that):
    given_that.state_of('light').is_set_to(light_states)    
    given_that.state_of('timer').is_set_to(timer_states)
    notification_engine.mqtt_message_recieved_event(None, {'topic': 'homeassistant/notification/tts', 'payload': '{"Messages":["Windows","Doors","LightsCount","Tele"]}'},  None)

However I'm not sure how to assert the event got fired and I'm getting an error

=================================================================================================== FAILURES ====================================================================================================
______________________________________________________________________________ test_during_night_light_turn_on[NotificationEngine] ______________________________________________________________________________

given_that = <appdaemontestframework.given_that.GivenThatWrapper object at 0xb54cae30>, notification_engine = <apps.NotificationEngine.NotificationEngine object at 0xb53ce130>
assert_that = <appdaemontestframework.assert_that.AssertThatWrapper object at 0xb53ce110>

    def test_during_night_light_turn_on(given_that, notification_engine, assert_that):
        given_that.state_of('light').is_set_to(light_states)
        given_that.state_of('timer').is_set_to(timer_states)
>       notification_engine.mqtt_message_recieved_event(None, {'topic': 'homeassistant/notification/tts', 'payload': '{"Messages":["Windows","Doors","LightsCount","Tele"]}'},  None)

test/NotificationEngineTest.py:30: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
apps/NotificationEngine.py:19: in mqtt_message_recieved_event
    self.fire_event("MQTT_PUBLISH", topic = "appdaemon/notification/tts", message = json.dumps({"message": message}))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <apps.NotificationEngine.NotificationEngine object at 0xb53ce130>, args = ('MQTT_PUBLISH',), kwargs = {'message': '{"message": "1 light on"}', 'topic': 'appdaemon/notification/tts'}, is_async = True
f = <Task pending coro=<fire_event() running at /usr/local/lib/python3.6/site-packages/appdaemon/adapi.py:1666>>

    @wraps(coro)
    def inner_sync_wrapper(self, *args, **kwargs):
        is_async = None
        try:
            # do this first to get the exception
            # otherwise the coro could be started and never awaited
            asyncio.get_event_loop()
            is_async = True
        except RuntimeError:
            is_async = False

        if is_async is True:
            # don't use create_task. It's python3.7 only
            f = asyncio.ensure_future(coro(self, *args, **kwargs))
>           self.AD.futures.add_future(self.name, f)
E           AttributeError: 'NotificationEngine' object has no attribute 'AD'

/usr/local/lib/python3.6/site-packages/appdaemon/utils.py:189: AttributeError
--------------------------------------------------------------------------------------------- Captured stdout call ----------------------------------------------------------------------------------------------
{'topic': 'homeassistant/notification/tts', 'payload': '{"Messages":["Windows","Doors","LightsCount","Tele"]}'}
========================================================================================== 1 failed, 1 passed in 0.33s ==========================================================================================

[HelloThisIsFlo]

Unfortunately I personally never used events in appdaemon. So it's totally possible that it fails because the feature is not implemented.

That being said, it could be interesting to at least fix the bug and allow the code to run. Adding assertions on events would be a nice bonus for the future.

I'll probably look at that next week. Thank you for reporting 🙂

Summary

• Fix: Automations should be able to run without throwing errror even when using self.fire_event
• Feature: Add assertions on events.

[eugeneniemand]

Thanks

I think for now I may just split the method to abstract that away and it will allow me to test the rest of it.

No rush and thanks for your time, when I'm more familiar with python concepts I'd like to contribute.

[HelloThisIsFlo]

That's great to hear 😊

For sure, feel free to contribute whenever you feel like it. And even if you don't feel super confident with your python knowledge, once you submit a Pull-Request I (and probably @lifeisafractal) will give you some tips on how to improve (if there is something to improve). So really feel free to give it a shot and you'll see. Also if you have questions you can open a PR as a WIP so we can start a discussion 🙂 The only requirement is that PR must come with the associated tests that cover the new/changed behavior 🙂 But even with that we can help if you struggle a bit 🙂👍

I'll still look at the fix next week, no worries 🙂👍

[HelloThisIsFlo]

Hey @eugeneniemand,

Once merged #45 will allow to have automation that fire events without having the tests failing.

There is still no built in way to assert on events, it might be a good feature for the future BUT, if you want to assert on events with a bit of workaround, you can do so now 🙂. Look at: Appdaemon Test Framework - Direct call to mocked functions

[HelloThisIsFlo]

(reopening until PR is merged)

[HelloThisIsFlo]

Hey @eugeneniemand Just pinging to let you know version 3.0.5 is now released and your tests with automation sending events should not crash anymore.

[eugeneniemand]

@FlorianKempenich thanks I will look at this when I have some time. I have just moved home and not using HASS at the moment

[HelloThisIsFlo]

🙂👍 Also there is no rush, you raised the issue and that was helpful enough. Check it out whenever you want, and don’t hesitate to comment here or in a new issue if you find any other problem 😌