bmspy/tests/test_init.py

import argparse
import sys
from unittest.mock import patch, MagicMock

import pytest

from bmspy import parse_args, main
from bmspy.classes import UPS


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_ups_data():
    return {"testups": UPS.from_dict({
        "bms_voltage": {"help": "V", "raw_value": 52.0, "value": "52.00", "units": "V"},
        "bms_date": {"help": "Date", "info": "2023-01-15"},
    })}


class TestParseArgs:
    def _parse(self, args: list[str]):
        with patch("sys.argv", ["bmspy"] + args):
            return parse_args()

    def test_socket_default(self):
        assert self._parse([]).socket == "/run/bmspy/bms"

    def test_socket_long(self):
        assert self._parse(["--socket", "/tmp/test.sock"]).socket == "/tmp/test.sock"

    def test_socket_short(self):
        assert self._parse(["-s", "/tmp/test.sock"]).socket == "/tmp/test.sock"

    def test_ups_default_is_none(self):
        assert self._parse([]).ups is None

    def test_ups_filter(self):
        assert self._parse(["--ups", "myups"]).ups == "myups"

    def test_json_default_false(self):
        assert self._parse([]).report_json is False

    def test_json_long(self):
        assert self._parse(["--json"]).report_json is True

    def test_json_short(self):
        assert self._parse(["-j"]).report_json is True

    def test_print_default_true(self):
        assert self._parse([]).report_print is True

    def test_prometheus_default_false(self):
        assert self._parse([]).report_prometheus is False

    def test_prometheus_flag(self):
        assert self._parse(["--prometheus"]).report_prometheus is True

    def test_influxdb_default_false(self):
        assert self._parse([]).report_influxdb is False

    def test_influxdb_long(self):
        assert self._parse(["--influxdb"]).report_influxdb is True

    def test_influxdb_short(self):
        assert self._parse(["-i"]).report_influxdb is True

    def test_bucket_default(self):
        assert self._parse([]).influx_bucket == "ups"

    def test_bucket_long(self):
        assert self._parse(["--bucket", "mybucket"]).influx_bucket == "mybucket"

    def test_bucket_short(self):
        assert self._parse(["-b", "mybucket"]).influx_bucket == "mybucket"

    def test_url_default_false(self):
        assert self._parse([]).influx_url is False

    def test_url_long(self):
        assert self._parse(["--url", "http://influx.example.com"]).influx_url == "http://influx.example.com"

    def test_org_long(self):
        assert self._parse(["--org", "myorg"]).influx_org == "myorg"

    def test_token_long(self):
        assert self._parse(["--token", "mytoken"]).influx_token == "mytoken"

    def test_verbose_default_zero(self):
        assert self._parse([]).verbose == 0

    def test_verbose_once(self):
        assert self._parse(["-v"]).verbose == 1

    def test_verbose_multiple(self):
        assert self._parse(["-v", "-v", "-v"]).verbose == 3

    def test_verbose_long(self):
        assert self._parse(["--verbose"]).verbose == 1


# ---------------------------------------------------------------------------
# main()
# ---------------------------------------------------------------------------

class TestMain:
    def _run_main(self, args: list[str]):
        with patch("sys.argv", ["bmspy"] + args):
            main()

    def test_print_mode_default(self, capsys):
        ups_data = _make_ups_data()
        with patch("sys.argv", ["bmspy"]), \
             patch("bmspy.client.handle_registration"), \
             patch("bmspy.client.read_data", return_value=ups_data):
            main()
        captured = capsys.readouterr()
        assert "testups" in captured.out

    def test_json_mode(self, capsys):
        ups_data = _make_ups_data()
        with patch("sys.argv", ["bmspy", "--json"]), \
             patch("bmspy.client.handle_registration"), \
             patch("bmspy.client.read_data", return_value=ups_data):
            main()
        captured = capsys.readouterr()
        # JSON output should contain field names
        assert "bms_voltage" in captured.out

    def test_keyboard_interrupt_is_caught(self, capsys):
        with patch("sys.argv", ["bmspy"]), \
             patch("bmspy.client.handle_registration"), \
             patch("bmspy.client.read_data", side_effect=KeyboardInterrupt("stopped")):
            main()  # should not raise

    def test_prometheus_flag_calls_export(self):
        import bmspy.prometheus as prom_mod
        mock_export = MagicMock()
        with patch("sys.argv", ["bmspy", "--prometheus"]), \
             patch.object(prom_mod, "prometheus_export", mock_export):
            main()
        mock_export.assert_called_once()

    def test_textfile_flag_calls_prometheus_export(self, tmp_path):
        import bmspy.prometheus as prom_mod
        filename = str(tmp_path / "metrics.prom")
        mock_export = MagicMock()
        with patch("sys.argv", ["bmspy", "--file", filename]), \
             patch.object(prom_mod, "prometheus_export", mock_export):
            main()
        mock_export.assert_called_once()
        call_kwargs = mock_export.call_args[1]
        assert call_kwargs.get("daemonize") is False
        assert call_kwargs.get("filename") == filename

    def test_influxdb_flag_calls_export(self):
        pytest.importorskip("influxdb_client_3", reason="influxdb3-python not installed")
        import bmspy.influxdb as influx_mod
        mock_export = MagicMock()
        with patch("sys.argv", ["bmspy", "--influxdb", "--url", "http://influx", "--org", "org", "--token", "tok"]), \
             patch.object(influx_mod, "influxdb_export", mock_export):
            main()
        mock_export.assert_called_once()

    def test_influxdb_partial_args_exits(self, capsys):
        """Providing only one of --url/--org/--token raises ArgumentTypeError (caught by KeyboardInterrupt handler)."""
        with patch("sys.argv", ["bmspy", "--influxdb", "--url", "http://influx"]):
            with pytest.raises(argparse.ArgumentTypeError):
                main()

    def test_influxdb_partial_raises_argument_error(self):
        """Two of three influx args causes ArgumentTypeError."""
        import argparse as _ap
        with patch("sys.argv", ["bmspy", "--influxdb", "--url", "http://influx", "--org", "org"]):
            with pytest.raises(_ap.ArgumentTypeError):
                main()