123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300 |
- import datetime
- import typing
- import unittest.mock
- import pytest
- import systemctl_mqtt
- import systemctl_mqtt._homeassistant
- import systemctl_mqtt._utils
- @pytest.mark.parametrize(
- (
- "argv",
- "expected_mqtt_host",
- "expected_mqtt_port",
- "expected_mqtt_disable_tls",
- "expected_username",
- "expected_password",
- "expected_topic_prefix",
- ),
- [
- (
- ["", "--mqtt-host", "mqtt-broker.local"],
- "mqtt-broker.local",
- 8883,
- False,
- None,
- None,
- None,
- ),
- (
- ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-disable-tls"],
- "mqtt-broker.local",
- 1883,
- True,
- None,
- None,
- None,
- ),
- (
- ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8883"],
- "mqtt-broker.local",
- 8883,
- False,
- None,
- None,
- None,
- ),
- (
- ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8884"],
- "mqtt-broker.local",
- 8884,
- False,
- None,
- None,
- None,
- ),
- (
- [
- "",
- "--mqtt-host",
- "mqtt-broker.local",
- "--mqtt-port",
- "8884",
- "--mqtt-disable-tls",
- ],
- "mqtt-broker.local",
- 8884,
- True,
- None,
- None,
- None,
- ),
- (
- ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
- "mqtt-broker.local",
- 8883,
- False,
- "me",
- None,
- None,
- ),
- (
- [
- "",
- "--mqtt-host",
- "mqtt-broker.local",
- "--mqtt-username",
- "me",
- "--mqtt-password",
- "secret",
- ],
- "mqtt-broker.local",
- 8883,
- False,
- "me",
- "secret",
- None,
- ),
- (
- [
- "",
- "--mqtt-host",
- "mqtt-broker.local",
- "--mqtt-topic-prefix",
- "system/command",
- ],
- "mqtt-broker.local",
- 8883,
- False,
- None,
- None,
- "system/command",
- ),
- ],
- )
- def test__main(
- argv,
- expected_mqtt_host,
- expected_mqtt_port,
- expected_mqtt_disable_tls,
- expected_username,
- expected_password,
- expected_topic_prefix: typing.Optional[str],
- ):
-
- with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv", argv
- ), unittest.mock.patch(
- "systemctl_mqtt._utils.get_hostname", return_value="hostname"
- ):
-
- systemctl_mqtt._main()
- run_mock.assert_called_once_with(
- mqtt_host=expected_mqtt_host,
- mqtt_port=expected_mqtt_port,
- mqtt_disable_tls=expected_mqtt_disable_tls,
- mqtt_username=expected_username,
- mqtt_password=expected_password,
- mqtt_topic_prefix=expected_topic_prefix or "systemctl/hostname",
- homeassistant_discovery_prefix="homeassistant",
- homeassistant_node_id="hostname",
- poweroff_delay=datetime.timedelta(seconds=4),
- )
- @pytest.mark.parametrize(
- ("password_file_content", "expected_password"),
- [
- ("secret", "secret"),
- ("secret space", "secret space"),
- ("secret ", "secret "),
- (" secret ", " secret "),
- ("secret\n", "secret"),
- ("secret\n\n", "secret\n"),
- ("secret\r\n", "secret"),
- ("secret\n\r\n", "secret\n"),
- ("你好\n", "你好"),
- ],
- )
- def test__main_password_file(tmpdir, password_file_content, expected_password):
- mqtt_password_path = tmpdir.join("mqtt-password")
- with mqtt_password_path.open("w") as mqtt_password_file:
- mqtt_password_file.write(password_file_content)
- with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- [
- "",
- "--mqtt-host",
- "localhost",
- "--mqtt-username",
- "me",
- "--mqtt-password-file",
- str(mqtt_password_path),
- ],
- ), unittest.mock.patch(
- "systemctl_mqtt._utils.get_hostname", return_value="hostname"
- ):
-
- systemctl_mqtt._main()
- run_mock.assert_called_once_with(
- mqtt_host="localhost",
- mqtt_port=8883,
- mqtt_disable_tls=False,
- mqtt_username="me",
- mqtt_password=expected_password,
- mqtt_topic_prefix="systemctl/hostname",
- homeassistant_discovery_prefix="homeassistant",
- homeassistant_node_id="hostname",
- poweroff_delay=datetime.timedelta(seconds=4),
- )
- def test__main_password_file_collision(capsys):
- with unittest.mock.patch(
- "sys.argv",
- [
- "",
- "--mqtt-host",
- "localhost",
- "--mqtt-username",
- "me",
- "--mqtt-password",
- "secret",
- "--mqtt-password-file",
- "/var/lib/secrets/mqtt/password",
- ],
- ):
- with pytest.raises(SystemExit):
-
- systemctl_mqtt._main()
- out, err = capsys.readouterr()
- assert not out
- assert (
- "argument --mqtt-password-file: not allowed with argument --mqtt-password\n"
- in err
- )
- @pytest.mark.parametrize(
- ("args", "discovery_prefix"),
- [
- ([], "homeassistant"),
- (["--homeassistant-discovery-prefix", "home/assistant"], "home/assistant"),
- ],
- )
- def test__main_homeassistant_discovery_prefix(args, discovery_prefix):
- with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
- ):
- systemctl_mqtt._main()
- run_mock.assert_called_once()
- assert run_mock.call_args[1]["homeassistant_discovery_prefix"] == discovery_prefix
- @pytest.mark.parametrize(
- ("args", "node_id"),
- [
- ([], "fallback"),
- (["--homeassistant-node-id", "raspberrypi"], "raspberrypi"),
- ],
- )
- def test__main_homeassistant_node_id(args, node_id):
- with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
- ), unittest.mock.patch(
- "systemctl_mqtt._utils.get_hostname", return_value="fallback"
- ):
- systemctl_mqtt._main()
- run_mock.assert_called_once()
- assert run_mock.call_args[1]["homeassistant_node_id"] == node_id
- @pytest.mark.parametrize(
- "args",
- [["--homeassistant-node-id", "no pe"], ["--homeassistant-node-id", ""]],
- )
- def test__main_homeassistant_node_id_invalid(args):
- with unittest.mock.patch(
- "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
- ):
- with pytest.raises(ValueError):
- systemctl_mqtt._main()
- @pytest.mark.parametrize(
- ("args", "poweroff_delay"),
- [
- ([], datetime.timedelta(seconds=4)),
- (["--poweroff-delay-seconds", "42.21"], datetime.timedelta(seconds=42.21)),
- (["--poweroff-delay-seconds", "3600"], datetime.timedelta(hours=1)),
- ],
- )
- def test__main_poweroff_delay(args, poweroff_delay):
- with unittest.mock.patch("systemctl_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv", ["", "--mqtt-host", "mqtt-broker.local"] + args
- ):
- systemctl_mqtt._main()
- run_mock.assert_called_once()
- assert run_mock.call_args[1]["poweroff_delay"] == poweroff_delay
|