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]) @pytest.mark.parametrize("power_setting", [0xC6, 0x34]) def test__run(caplog, mqtt_host, mqtt_port, power_setting): 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, power_setting=power_setting, ) 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 == power_setting 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, power_setting=0xC6, ) 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, power_setting=0xC6, ) @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, power_setting=0xC6, ) 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 )