intertechno-cc1101-mqtt/tests/test_mqtt.py

import logging
import unittest.mock

import pytest
from paho.mqtt.client import MQTT_ERR_QUEUE_SIZE, MQTT_ERR_SUCCESS, MQTTMessage, Client

import intertechno_cc1101_mqtt

# pylint: disable=protected-access


@pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
@pytest.mark.parametrize("mqtt_port", [1833])
def test__run(caplog, mqtt_host, mqtt_port):
    with unittest.mock.patch(
        "paho.mqtt.client.Client"
    ) as mqtt_client_mock, caplog.at_level(logging.DEBUG):
        intertechno_cc1101_mqtt._run(
            mqtt_host=mqtt_host,
            mqtt_port=mqtt_port,
            mqtt_username=None,
            mqtt_password=None,
        )
    mqtt_client_mock.assert_called_once_with()
    assert not mqtt_client_mock().username_pw_set.called
    mqtt_client_mock().connect.assert_called_once_with(host=mqtt_host, port=mqtt_port)
    mqtt_client_mock().socket().getpeername.return_value = (mqtt_host, mqtt_port)
    with caplog.at_level(logging.DEBUG):
        mqtt_client_mock().on_connect(mqtt_client_mock(), None, {}, 0)
    # pylint: disable=comparison-with-callable
    assert mqtt_client_mock().on_message == intertechno_cc1101_mqtt._mqtt_on_message
    mqtt_client_mock().subscribe.assert_called_once_with("intertechno-cc1101/+/+/set")
    mqtt_client_mock().loop_forever.assert_called_once_with()
    assert caplog.record_tuples == [
        (
            "intertechno_cc1101_mqtt",
            logging.INFO,
            "connecting to MQTT broker {}:{}".format(mqtt_host, mqtt_port),
        ),
        (
            "intertechno_cc1101_mqtt",
            logging.DEBUG,
            "connected to MQTT broker {}:{}".format(mqtt_host, mqtt_port),
        ),
        (
            "intertechno_cc1101_mqtt",
            logging.INFO,
            "subscribing to MQTT topic 'intertechno-cc1101/+/+/set'",
        ),
    ]


@pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
@pytest.mark.parametrize("mqtt_port", [1833])
@pytest.mark.parametrize("mqtt_username", ["me"])
@pytest.mark.parametrize("mqtt_password", [None, "secret"])
def test__run_authentication(mqtt_host, mqtt_port, mqtt_username, mqtt_password):
    with unittest.mock.patch("paho.mqtt.client.Client") as mqtt_client_mock:
        intertechno_cc1101_mqtt._run(
            mqtt_host=mqtt_host,
            mqtt_port=mqtt_port,
            mqtt_username=mqtt_username,
            mqtt_password=mqtt_password,
        )
    mqtt_client_mock.assert_called_once_with()
    mqtt_client_mock().username_pw_set.assert_called_once_with(
        username=mqtt_username, password=mqtt_password
    )


@pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
@pytest.mark.parametrize("mqtt_port", [1833])
@pytest.mark.parametrize("mqtt_password", ["secret"])
def test__run_authentication_missing_username(mqtt_host, mqtt_port, mqtt_password):
    with unittest.mock.patch("paho.mqtt.client.Client"):
        with pytest.raises(ValueError):
            intertechno_cc1101_mqtt._run(
                mqtt_host=mqtt_host,
                mqtt_port=mqtt_port,
                mqtt_username=None,
                mqtt_password=mqtt_password,
            )


@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", None, 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", None, 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)),
)
def test__mqtt_on_message_button_index_action(topic, button_index, payload, turn_on):
    message = MQTTMessage(topic=topic)
    message.payload = payload
    with unittest.mock.patch("intertechno_cc1101.RemoteControl") as remote_control_mock:
        intertechno_cc1101_mqtt._mqtt_on_message("dummy", None, message)
    if turn_on:
        remote_control_mock().turn_on.assert_called_once_with(button_index=button_index)
        remote_control_mock().turn_off.assert_not_called()
    else:
        remote_control_mock().turn_off.assert_called_once_with(
            button_index=button_index
        )
        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", None, 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", (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", None, 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], AssertionError)


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", None, 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)