123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287 |
- # switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
- # compatible with home-assistant.io's MQTT Switch & Cover platform
- #
- # Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
- #
- # This program is free software: you can redistribute it and/or modify
- # it under the terms of the GNU General Public License as published by
- # the Free Software Foundation, either version 3 of the License, or
- # any later version.
- #
- # This program is distributed in the hope that it will be useful,
- # but WITHOUT ANY WARRANTY; without even the implied warranty of
- # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- # GNU General Public License for more details.
- #
- # You should have received a copy of the GNU General Public License
- # along with this program. If not, see <https://www.gnu.org/licenses/>.
- import json
- import logging
- import subprocess
- import typing
- import unittest.mock
- import pytest
- import switchbot_mqtt
- import switchbot_mqtt._cli
- # pylint: disable=protected-access; tests
- # pylint: disable=too-many-arguments; these are tests, no API
- def test_console_entry_point():
- assert subprocess.run(
- ["switchbot-mqtt", "--help"], stdout=subprocess.PIPE, check=True
- ).stdout.startswith(b"usage: ")
- @pytest.mark.parametrize(
- (
- "argv",
- "expected_mqtt_host",
- "expected_mqtt_port",
- "expected_username",
- "expected_password",
- "expected_retry_count",
- ),
- [
- (
- ["", "--mqtt-host", "mqtt-broker.local"],
- "mqtt-broker.local",
- 1883,
- None,
- None,
- 3,
- ),
- (
- ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-port", "8883"],
- "mqtt-broker.local",
- 8883,
- None,
- None,
- 3,
- ),
- (
- ["", "--mqtt-host", "mqtt-broker.local", "--mqtt-username", "me"],
- "mqtt-broker.local",
- 1883,
- "me",
- None,
- 3,
- ),
- (
- [
- "",
- "--mqtt-host",
- "mqtt-broker.local",
- "--mqtt-username",
- "me",
- "--mqtt-password",
- "secret",
- "--retries",
- "21",
- ],
- "mqtt-broker.local",
- 1883,
- "me",
- "secret",
- 21,
- ),
- ],
- )
- def test__main(
- argv,
- expected_mqtt_host,
- expected_mqtt_port,
- expected_username,
- expected_password,
- expected_retry_count,
- ):
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv", argv
- ):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(
- mqtt_host=expected_mqtt_host,
- mqtt_port=expected_mqtt_port,
- mqtt_username=expected_username,
- mqtt_password=expected_password,
- retry_count=expected_retry_count,
- device_passwords={},
- fetch_device_info=False,
- )
- @pytest.mark.parametrize(
- ("mqtt_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_mqtt_password_file(
- tmpdir, mqtt_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(mqtt_password_file_content)
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- [
- "",
- "--mqtt-host",
- "localhost",
- "--mqtt-username",
- "me",
- "--mqtt-password-file",
- str(mqtt_password_path),
- ],
- ):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(
- mqtt_host="localhost",
- mqtt_port=1883,
- mqtt_username="me",
- mqtt_password=expected_password,
- retry_count=3,
- device_passwords={},
- fetch_device_info=False,
- )
- def test__main_mqtt_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):
- switchbot_mqtt._cli._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(
- "device_passwords",
- [
- {},
- {"11:22:33:44:55:66": "password", "aa:bb:cc:dd:ee:ff": "secret"},
- ],
- )
- def test__main_device_password_file(tmpdir, device_passwords):
- device_passwords_path = tmpdir.join("passwords.json")
- device_passwords_path.write_text(json.dumps(device_passwords), encoding="utf8")
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- [
- "",
- "--mqtt-host",
- "localhost",
- "--device-password-file",
- str(device_passwords_path),
- ],
- ):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(
- mqtt_host="localhost",
- mqtt_port=1883,
- mqtt_username=None,
- mqtt_password=None,
- retry_count=3,
- device_passwords=device_passwords,
- fetch_device_info=False,
- )
- def test__main_fetch_device_info():
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- [
- "",
- "--mqtt-host",
- "localhost",
- ],
- ):
- switchbot_mqtt._cli._main()
- default_kwargs = dict(
- mqtt_host="localhost",
- mqtt_port=1883,
- mqtt_username=None,
- mqtt_password=None,
- retry_count=3,
- device_passwords={},
- )
- run_mock.assert_called_once_with(fetch_device_info=False, **default_kwargs)
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- ["", "--mqtt-host", "localhost", "--fetch-device-info"],
- ):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- ["", "--mqtt-host", "localhost"],
- ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": "21"}):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- ["", "--mqtt-host", "localhost"],
- ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": ""}):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(fetch_device_info=False, **default_kwargs)
- with unittest.mock.patch("switchbot_mqtt._run") as run_mock, unittest.mock.patch(
- "sys.argv",
- ["", "--mqtt-host", "localhost"],
- ), unittest.mock.patch.dict("os.environ", {"FETCH_DEVICE_INFO": " "}):
- switchbot_mqtt._cli._main()
- run_mock.assert_called_once_with(fetch_device_info=True, **default_kwargs)
- @pytest.mark.parametrize(
- ("additional_argv", "root_log_level", "log_format"),
- [
- ([], logging.INFO, "%(message)s"),
- (
- ["--debug"],
- logging.DEBUG,
- "%(asctime)s:%(levelname)s:%(name)s:%(funcName)s:%(message)s",
- ),
- ],
- )
- def test__main_log_config(
- additional_argv: typing.List[str], root_log_level: int, log_format: str
- ):
- with unittest.mock.patch(
- "sys.argv", ["", "--mqtt-host", "localhost"] + additional_argv
- ), unittest.mock.patch(
- "logging.basicConfig"
- ) as logging_basic_config_mock, unittest.mock.patch(
- "switchbot_mqtt._run"
- ):
- switchbot_mqtt._cli._main()
- logging_basic_config_mock.assert_called_once_with(
- level=root_log_level, format=log_format, datefmt="%Y-%m-%dT%H:%M:%S%z"
- )
|