appdaemon-testing

Ergonomic and pythonic unit testing for AppDaemon apps. Utilities to allow you to test your AppDaemon home automation apps using all the pythonic testing patterns you are already familiar with.

Install

pip install appdaemon-testing

Full Documentation

An enhanced, source-linked version of the documentation below as well as complete API documentation is available here

Writing your first test

This demo assumes you will use pytest as your test runner. Install the appdaemon-testing and pytest packages:

pip install appdaemon-testing pytest

In your appdaemon configuration directory, introduce a new tests directory. This is where we are going to write the tests for your apps.

Additionally we also need to introduce an __init__.py file to tests and apps directories to make them an importable package. You should have a tree that looks something like this:

├── appdaemon.yaml
├── apps
│   ├── __init__.py
│   ├── apps.yaml
│   └── living_room_motion.py
├── dashboards
├── namespaces
└── tests
    ├── __init__.py
    └── test_living_room_motion.py

We have an automation, apps/living_room_motion.py that we wish to test. It looks like this:

import appdaemon.plugins.hass.hassapi as hass


class LivingRoomMotion(hass.Hass):
    def initialize(self):
        self.listen_state(self.on_motion_detected, self.args["motion_entity"])

    def on_motion_detected(self, entity, attribute, old, new, kwargs):
        if old == "off" and new == "on":
            for light in self.args["light_entities"]:
                self.turn_on(light)

Create a new file, tests/test_living_room_motion.py. This is where we will write the tests for our automation.

First we will declare an automation_fixture():

@automation_fixture(
    LivingRoomMotion,
    args={
        "motion_entity": "binary_sensor.motion_detected",
        "light_entities": ["light.1", "light.2", "light.3"],
    },
)
def living_room_motion() -> LivingRoomMotion:
    pass

With this fixture, it's now possible to write some tests. We will first write a test to check the state listener callbacks are registered:

def test_callbacks_are_registered(hass_driver, living_room_motion: LivingRoomMotion):
    listen_state = hass_driver.get_mock("listen_state")
    listen_state.assert_called_once_with(
        living_room_motion.on_motion_detected, "binary_sensor.motion_detected")

We use the hass_driver() fixture to obtain mock implementations of methods that exist on the AppDaemon Hass API. We can query these mocks and make assertions on their values. In this test we make an assertion that listen_state is called once with the specified parameters.

We will next write a test to make an assertion that the lights are turned on when motion is detected:

def test_lights_are_turned_on_when_motion_detected(
    hass_driver, living_room_motion: LivingRoomMotion
):
    with hass_driver.setup():
        hass_driver.set_state("binary_sensor.motion_detected", "off")

    hass_driver.set_state("binary_sensor.motion_detected", "on")

    turn_on = hass_driver.get_mock("turn_on")
    assert turn_on.call_count == 3
    turn_on.assert_has_calls(
        [mock.call("light.1"), mock.call("light.2"), mock.call("light.3")]
    )

This test uses the HassDriver.setup() context manager to set the initial state for testing. When execution is within HassDriver.setup() all state updates will not be triggered.

With the initial state configured, we can now proceed to triggering the state change (HassDriver.set_state()). After the state change has occured, we can then begin to make assertions about calls made to the underlying API. In this test we wish to make assertions that turn_on is called. We obtain the turn_on mock implementation and make assertions about its calls and call count.

You can see this full example and example directory structure within the example directory in this repo.

pytest plugin

The appdaemon_testing.pytest package provides a handy hass_driver() fixture to allow you easy access to the global HassDriver instance. This fixture takes care of ensuring AppDaemon base class methods are patched.

Additionally, it provides a decorator, automation_fixture() which can be used to declare automation fixtures. It can be used like so:

from appdaemon_testing.pytest import automation_fixture
from apps.living_room_motion import LivingRoomMotion


@automation_fixture(
    LivingRoomMotion,
    args={
        "motion_entity": "binary_sensor.motion_detected",
        "light_entities": ["light.1", "light.2", "light.3"],
    },
)
def living_room_motion() -> LivingRoomMotion:
    pass

"""
.. include:: ../README.md
"""
from .hass_driver import HassDriver

__all__ = ["HassDriver"]
__version__ = "0.0.0-dev"

Sub-modules

appdaemon_testing.pytest

Classes

class HassDriver

Expand source code

class HassDriver:
    def __init__(self):
        self._mocks = dict(
            log=mock.Mock(),
            error=mock.Mock(),
            call_service=mock.Mock(),
            cancel_timer=mock.Mock(),
            get_state=mock.Mock(side_effect=self._se_get_state),
            # TODO(NW): Implement side-effect for listen_event
            listen_event=mock.Mock(),
            fire_event=mock.Mock(),
            listen_state=mock.Mock(side_effect=self._se_listen_state),
            notify=mock.Mock(),
            run_at=mock.Mock(),
            run_at_sunrise=mock.Mock(),
            run_at_sunset=mock.Mock(),
            run_daily=mock.Mock(),
            run_every=mock.Mock(),
            run_hourly=mock.Mock(),
            run_in=mock.Mock(),
            run_minutely=mock.Mock(),
            set_state=mock.Mock(),
            time=mock.Mock(),
            turn_off=mock.Mock(),
            turn_on=mock.Mock(),
        )

        self._setup_active = False
        self._states: Dict[str, Dict[str, Any]] = defaultdict(lambda: {"state": None})
        self._state_spys: Dict[Union[str, None], List[StateSpy]] = defaultdict(
            lambda: []
        )

    def get_mock(self, meth: str) -> mock.Mock:
        """
        Returns the mock associated with the provided AppDaemon method

        Parameters:
            meth: The method to retreive the mock implementation for
        """
        return self._mocks[meth]

    def inject_mocks(self) -> None:
        """
        Monkey-patch the AppDaemon hassapi.Hass base-class methods with mock
        implementations.
        """
        for meth_name, impl in self._mocks.items():
            if getattr(hass.Hass, meth_name) is None:
                raise AssertionError("Attempt to mock non existing method: ", meth_name)
            _LOGGER.debug("Patching hass.Hass.%s", meth_name)
            setattr(hass.Hass, meth_name, impl)

    @contextlib.contextmanager
    def setup(self):
        """
        A context manager to indicate that execution is taking place during a
        "setup" phase.

        This context manager can be used to configure/set up any existing states
        that might be required to run the test. State changes during execution within
        this context manager will cause `listen_state` handlers to not be called.

        Example:

        ```py
        def test_my_app(hass_driver, my_app: MyApp):
            with hass_driver.setup():
                # Any registered listen_state handlers will not be called
                hass_driver.set_state("binary_sensor.motion_detected", "off")

            # Respective listen_state handlers will be called
            hass_driver.set_state("binary_sensor.motion_detected", "on")
            ...
        ```
        """
        self._setup_active = True
        yield None
        self._setup_active = False

    def set_state(
        self, entity, state, *, attribute_name="state", previous=None, trigger=None
    ) -> None:
        """
        Update/set state of an entity.

        State changes will cause listeners (via listen_state) to be called on
        their respective state changes.

        Parameters:
            entity: The entity to update
            state: The state value to set
            attribute_name: The attribute to set
            previous: Forced previous value
            trigger: Whether this change should trigger registered listeners
                     (via listen_state)
        """
        if trigger is None:
            # Avoid triggering state changes during state setup phase
            trigger = not self._setup_active

        domain, _ = entity.split(".")
        state_entry = self._states[entity]
        prev_state = copy(state_entry)
        old_value = previous or prev_state.get(attribute_name)
        new_value = state

        if old_value == new_value:
            return

        # Update the state entry
        state_entry[attribute_name] = new_value

        if not trigger:
            return

        # Notify subscribers of the change
        for spy in self._state_spys[domain] + self._state_spys[entity]:
            sat_attr = spy.attribute == attribute_name or spy.attribute == "all"
            sat_new = spy.new is None or spy.new == new_value
            sat_old = spy.old is None or spy.old == old_value

            param_old = prev_state if spy.attribute == "all" else old_value
            param_new = copy(state_entry) if spy.attribute == "all" else new_value
            param_attribute = None if spy.attribute == "all" else attribute_name

            if all([sat_old, sat_new, sat_attr]):
                spy.callback(entity, param_attribute, param_old, param_new, spy.kwargs)

    def _se_get_state(self, entity_id=None, attribute="state", default=None, **kwargs):
        _LOGGER.debug("Getting state for entity: %s", entity_id)

        fully_qualified = "." in entity_id
        matched_states = {}
        if fully_qualified:
            matched_states[entity_id] = self._states[entity_id]
        else:
            for s_eid, state in self._states.items():
                domain, entity = s_eid.split(".")
                if domain == entity_id:
                    matched_states[s_eid] = state

        # With matched states, map the provided attribute (if applicable)
        if attribute != "all":
            matched_states = {
                eid: state.get(attribute) for eid, state in matched_states.items()
            }

        if default is not None:
            matched_states = {
                eid: state or default for eid, state in matched_states.items()
            }

        if fully_qualified:
            return matched_states[entity_id]
        else:
            return matched_states

    def _se_listen_state(
        self, callback, entity=None, attribute=None, new=None, old=None, **kwargs
    ):
        spy = StateSpy(
            callback=callback,
            attribute=attribute or "state",
            new=new,
            old=old,
            kwargs=kwargs,
        )
        self._state_spys[entity].append(spy)

Methods

def get_mock(self, meth: str) ‑> unittest.mock.Mock

Returns the mock associated with the provided AppDaemon method

Parameters

meth: The method to retreive the mock implementation for

Expand source code

def get_mock(self, meth: str) -> mock.Mock:
    """
    Returns the mock associated with the provided AppDaemon method

    Parameters:
        meth: The method to retreive the mock implementation for
    """
    return self._mocks[meth]

def inject_mocks(self) ‑> None

Monkey-patch the AppDaemon hassapi.Hass base-class methods with mock implementations.

Expand source code

def inject_mocks(self) -> None:
    """
    Monkey-patch the AppDaemon hassapi.Hass base-class methods with mock
    implementations.
    """
    for meth_name, impl in self._mocks.items():
        if getattr(hass.Hass, meth_name) is None:
            raise AssertionError("Attempt to mock non existing method: ", meth_name)
        _LOGGER.debug("Patching hass.Hass.%s", meth_name)
        setattr(hass.Hass, meth_name, impl)

def set_state(self, entity, state, *, attribute_name='state', previous=None, trigger=None) ‑> None

Update/set state of an entity.

State changes will cause listeners (via listen_state) to be called on their respective state changes.

Parameters

entity: The entity to update state: The state value to set attribute_name: The attribute to set previous: Forced previous value trigger: Whether this change should trigger registered listeners (via listen_state)

Expand source code

def set_state(
    self, entity, state, *, attribute_name="state", previous=None, trigger=None
) -> None:
    """
    Update/set state of an entity.

    State changes will cause listeners (via listen_state) to be called on
    their respective state changes.

    Parameters:
        entity: The entity to update
        state: The state value to set
        attribute_name: The attribute to set
        previous: Forced previous value
        trigger: Whether this change should trigger registered listeners
                 (via listen_state)
    """
    if trigger is None:
        # Avoid triggering state changes during state setup phase
        trigger = not self._setup_active

    domain, _ = entity.split(".")
    state_entry = self._states[entity]
    prev_state = copy(state_entry)
    old_value = previous or prev_state.get(attribute_name)
    new_value = state

    if old_value == new_value:
        return

    # Update the state entry
    state_entry[attribute_name] = new_value

    if not trigger:
        return

    # Notify subscribers of the change
    for spy in self._state_spys[domain] + self._state_spys[entity]:
        sat_attr = spy.attribute == attribute_name or spy.attribute == "all"
        sat_new = spy.new is None or spy.new == new_value
        sat_old = spy.old is None or spy.old == old_value

        param_old = prev_state if spy.attribute == "all" else old_value
        param_new = copy(state_entry) if spy.attribute == "all" else new_value
        param_attribute = None if spy.attribute == "all" else attribute_name

        if all([sat_old, sat_new, sat_attr]):
            spy.callback(entity, param_attribute, param_old, param_new, spy.kwargs)

def setup(self)

A context manager to indicate that execution is taking place during a "setup" phase.

This context manager can be used to configure/set up any existing states that might be required to run the test. State changes during execution within this context manager will cause listen_state handlers to not be called.

Example:

def test_my_app(hass_driver, my_app: MyApp):
    with hass_driver.setup():
        # Any registered listen_state handlers will not be called
        hass_driver.set_state("binary_sensor.motion_detected", "off")

    # Respective listen_state handlers will be called
    hass_driver.set_state("binary_sensor.motion_detected", "on")
    ...

Expand source code

@contextlib.contextmanager
def setup(self):
    """
    A context manager to indicate that execution is taking place during a
    "setup" phase.

    This context manager can be used to configure/set up any existing states
    that might be required to run the test. State changes during execution within
    this context manager will cause `listen_state` handlers to not be called.

    Example:

    ```py
    def test_my_app(hass_driver, my_app: MyApp):
        with hass_driver.setup():
            # Any registered listen_state handlers will not be called
            hass_driver.set_state("binary_sensor.motion_detected", "off")

        # Respective listen_state handlers will be called
        hass_driver.set_state("binary_sensor.motion_detected", "on")
        ...
    ```
    """
    self._setup_active = True
    yield None
    self._setup_active = False