intertechno-cc1101-mqtt/tests/test_mqtt_message.py

import logging
import unittest.mock

import pytest
from paho.mqtt.client import MQTTMessage

import intertechno_cc1101_mqtt

# pylint: disable=protected-access


@pytest.mark.parametrize(
    ("topic", "address"),
    (
        (b"intertechno-cc1101/12345678/0/set", 12345678),
        (b"intertechno-cc1101/1234/0/set", 1234),
    ),
)
def test__mqtt_on_message_address(topic, address):
    message = MQTTMessage(topic=topic)
    message.payload = b"ON"
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        intertechno_cc1101_mqtt._mqtt_on_message(
            "dummy",
            intertechno_cc1101_mqtt._MQTTEventUserData(aliases={}, power_setting=0xC6),
            message,
        )
    remote_control_mock.assert_called_once_with(address=address)


@pytest.mark.parametrize(
    ("topic", "address_str"),
    (
        (b"intertechno-cc1101/abcdef/0/set", "abcdef"),
        (b"intertechno-cc1101//0/set", ""),
    ),
)
def test__mqtt_on_message_invalid_address(caplog, topic, address_str):
    message = MQTTMessage(topic=topic)
    message.payload = b"ON"
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        with caplog.at_level(logging.WARNING):
            intertechno_cc1101_mqtt._mqtt_on_message(
                "dummy",
                intertechno_cc1101_mqtt._MQTTEventUserData(
                    aliases={}, power_setting=0xC6
                ),
                message,
            )
    remote_control_mock.assert_not_called()
    assert caplog.record_tuples == [
        (
            "intertechno_cc1101_mqtt",
            logging.WARNING,
            "failed to parse address {!r}, expected integer; ignoring message".format(
                address_str
            ),
        )
    ]


@pytest.mark.parametrize(
    ("topic", "button_index"),
    ((b"intertechno-cc1101/12345678/0/set", 0), (b"intertechno-cc1101/1234/7/set", 7)),
)
@pytest.mark.parametrize(
    ("payload", "turn_on"),
    ((b"ON", True), (b"On", True), (b"on", True), (b"OFF", False), (b"off", False)),
)
@pytest.mark.parametrize("retain", [True, False])
@pytest.mark.parametrize("power_setting", [0xC6, 0xC0])
def test__mqtt_on_message_button_index_action(
    topic, button_index, payload, turn_on, retain, power_setting
):
    # pylint: disable=too-many-arguments
    message = MQTTMessage(topic=topic)
    message.payload = payload
    message.retain = retain
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        intertechno_cc1101_mqtt._mqtt_on_message(
            "dummy",
            intertechno_cc1101_mqtt._MQTTEventUserData(
                aliases={}, power_setting=power_setting
            ),
            message,
        )
    if turn_on:
        remote_control_mock().turn_on.assert_called_once_with(
            button_index=button_index, power_setting=power_setting
        )
        remote_control_mock().turn_off.assert_not_called()
    else:
        remote_control_mock().turn_off.assert_called_once_with(
            button_index=button_index, power_setting=power_setting
        )
        remote_control_mock().turn_on.assert_not_called()


@pytest.mark.parametrize(
    ("topic", "button_index_str"),
    (
        (b"intertechno-cc1101/12345678/abc/set", "abc"),
        (b"intertechno-cc1101/12345678//set", ""),
    ),
)
def test__mqtt_on_message_invalid_button_index(caplog, topic, button_index_str):
    message = MQTTMessage(topic=topic)
    message.payload = b"ON"
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        with caplog.at_level(logging.WARNING):
            intertechno_cc1101_mqtt._mqtt_on_message(
                "dummy",
                intertechno_cc1101_mqtt._MQTTEventUserData(
                    aliases={}, power_setting=0xC6
                ),
                message,
            )
    remote_control_mock().turn_on.assert_not_called()
    remote_control_mock().turn_off.assert_not_called()
    assert caplog.record_tuples == [
        (
            "intertechno_cc1101_mqtt",
            logging.WARNING,
            "failed to parse button index {!r}, expected integer; ignoring message".format(
                button_index_str
            ),
        )
    ]


@pytest.mark.parametrize(
    ("topic", "address", "button_index"),
    (
        (b"intertechno-cc1101/some-name/set", 21, 0),
        (b"intertechno-cc1101/another-name/set", 42, 7),
    ),
)
def test__mqtt_on_message_alias(topic, address, button_index):
    message = MQTTMessage(topic=topic)
    message.payload = b"ON"
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        intertechno_cc1101_mqtt._mqtt_on_message(
            "dummy",
            intertechno_cc1101_mqtt._MQTTEventUserData(
                aliases={
                    "some-name": {"address": 21, "button-index": 0},
                    "another-name": {"address": "42", "button-index": "7"},  # :O
                },
                power_setting=0xC6,
            ),
            message,
        )
    remote_control_mock.assert_called_once_with(address=address)
    remote_control_mock().turn_on.assert_called_once_with(
        button_index=button_index, power_setting=0xC6
    )


@pytest.mark.parametrize(
    ("topic", "alias"),
    (
        (b"intertechno-cc1101//set", ""),
        (b"intertechno-cc1101/unknown-name/set", "unknown-name"),
    ),
)
def test__mqtt_on_message_undefined_alias(caplog, topic, alias):
    message = MQTTMessage(topic=topic)
    message.payload = b"ON"
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        with caplog.at_level(logging.WARNING):
            intertechno_cc1101_mqtt._mqtt_on_message(
                "dummy",
                intertechno_cc1101_mqtt._MQTTEventUserData(
                    aliases={"some-name": {"address": 21, "button-index": 0}},
                    power_setting=0xC6,
                ),
                message,
            )
    remote_control_mock.assert_not_called()
    assert caplog.record_tuples == [
        (
            "intertechno_cc1101_mqtt",
            logging.WARNING,
            "unknown alias {!r}; ignoring message".format(alias),
        )
    ]


@pytest.mark.parametrize(
    "aliases",
    (
        {"some-name": {"address": 21}},
        {"some-name": {"button-index": 0}},
        {"some-name": {"adresse": 21, "button-index": 0}},
    ),
)
def test__mqtt_on_message_alias_missing_attrs(caplog, aliases):
    message = MQTTMessage(topic=b"intertechno-cc1101/some-name/set")
    message.payload = b"ON"
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        with caplog.at_level(logging.WARNING):
            intertechno_cc1101_mqtt._mqtt_on_message(
                "dummy",
                intertechno_cc1101_mqtt._MQTTEventUserData(
                    aliases=aliases, power_setting=0xC6
                ),
                message,
            )
    remote_control_mock.assert_not_called()
    assert caplog.record_tuples == [
        (
            "intertechno_cc1101_mqtt",
            logging.ERROR,
            "alias file must provide fields 'address' and 'button-index' for each alias",
        )
    ]


@pytest.mark.parametrize(
    "topic", (b"intertechno-cc1101/123456789/0/set", b"intertechno-cc1101/-21/0/set")
)
def test__mqtt_on_message_remote_init_failed(caplog, topic):
    message = MQTTMessage(topic=topic)
    message.payload = b"ON"
    with caplog.at_level(logging.WARNING):
        intertechno_cc1101_mqtt._mqtt_on_message(
            "dummy",
            intertechno_cc1101_mqtt._MQTTEventUserData(aliases={}, power_setting=0xC6),
            message,
        )
    assert len(caplog.records) == 1
    assert caplog.records[0].levelno == logging.WARNING
    assert (
        caplog.records[0].message
        == "failed to initialize remote control, invalid address? ignoring message"
    )
    assert isinstance(caplog.records[0].exc_info[1], ValueError)


def test__mqtt_on_message_transmission_failed(caplog):
    message = MQTTMessage(topic=b"intertechno-cc1101/12345678/3/set")
    message.payload = b"ON"
    with unittest.mock.patch(
        "cc1101.CC1101.__enter__",
        side_effect=FileNotFoundError("[Errno 2] No such file or directory"),
    ), caplog.at_level(logging.ERROR):
        intertechno_cc1101_mqtt._mqtt_on_message(
            "dummy",
            intertechno_cc1101_mqtt._MQTTEventUserData(aliases={}, power_setting=0xC6),
            message,
        )
    assert len(caplog.records) == 1
    assert caplog.records[0].levelno == logging.ERROR
    assert caplog.records[0].message == "failed to send signal"
    assert isinstance(caplog.records[0].exc_info[1], FileNotFoundError)


@pytest.mark.parametrize("payload", (b"EIN", b"aus", b""))
def test__mqtt_on_message_invalid_payload(caplog, payload):
    message = MQTTMessage(topic=b"intertechno-cc1101/1234/7/set")
    message.payload = payload
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        intertechno_cc1101_mqtt._mqtt_on_message(
            "dummy",
            intertechno_cc1101_mqtt._MQTTEventUserData(aliases={}, power_setting=0xC6),
            message,
        )
    remote_control_mock().turn_off.assert_not_called()
    remote_control_mock().turn_on.assert_not_called()
    assert caplog.record_tuples == [
        (
            "intertechno_cc1101_mqtt",
            30,
            "unexpected payload {!r}; expected 'ON' or 'OFF'".format(payload),
        )
    ]