intertechno-cc1101-mqtt/tests/test_mqtt.py

import json
import logging
import pathlib
import unittest.mock

import pytest

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,
            alias_file_path=None,
        )
    assert mqtt_client_mock.call_count == 1
    assert not mqtt_client_mock.call_args[0]
    assert set(mqtt_client_mock.call_args[1]) == {"userdata"}
    event_userdata = mqtt_client_mock.call_args[1]["userdata"]
    assert event_userdata.aliases == {}
    assert event_userdata.power_setting == 0xC6
    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(), event_userdata, {}, 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' (address & button index)",
        ),
    ]


@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,
            alias_file_path=None,
        )
    assert mqtt_client_mock.call_count == 1
    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,
                alias_file_path=None,
            )


@pytest.mark.parametrize("mqtt_host", ["mqtt-broker.local"])
@pytest.mark.parametrize("mqtt_port", [1833])
@pytest.mark.parametrize(
    "aliases", [{}, {"some-alias": {"address": 12345678, "button-index": 0}}]
)
def test__run_alias_file_path(caplog, tmp_path, mqtt_host, mqtt_port, aliases):
    alias_file_path = tmp_path.joinpath("aliases.json")
    alias_file_path.write_text(json.dumps(aliases))
    assert isinstance(alias_file_path, pathlib.Path)
    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=None,
            mqtt_password=None,
            alias_file_path=alias_file_path,
        )
    assert mqtt_client_mock.call_count == 1
    event_userdata = mqtt_client_mock.call_args[1]["userdata"]
    assert event_userdata.aliases == aliases
    mqtt_client_mock().socket().getpeername.return_value = (mqtt_host, mqtt_port)
    with unittest.mock.patch(
        "intertechno_cc1101_mqtt._publish_homeassistant_discovery_configs"
    ) as publish_discovery_config_mock, caplog.at_level(logging.INFO):
        mqtt_client_mock().on_connect(mqtt_client_mock(), event_userdata, {}, 0)
    assert mqtt_client_mock().subscribe.call_args_list[0] == unittest.mock.call(
        "intertechno-cc1101/+/+/set"
    )
    assert caplog.record_tuples[0] == (
        "intertechno_cc1101_mqtt",
        logging.INFO,
        "subscribing to MQTT topic 'intertechno-cc1101/+/+/set' (address & button index)",
    )
    if len(aliases) == 0:
        assert mqtt_client_mock().subscribe.call_count == 1
        assert len(caplog.records) == 1
        publish_discovery_config_mock.assert_not_called()
    else:
        assert mqtt_client_mock().subscribe.call_count == 2
        assert len(caplog.records) == 2
        assert mqtt_client_mock().subscribe.call_args_list[1] == unittest.mock.call(
            "intertechno-cc1101/+/set"
        )
        assert caplog.record_tuples[1] == (
            "intertechno_cc1101_mqtt",
            logging.INFO,
            "subscribing to MQTT topic 'intertechno-cc1101/+/set' (alias)",
        )
        publish_discovery_config_mock.assert_called_once_with(
            mqtt_client=mqtt_client_mock(), aliases=aliases
        )