import pytest pytest.importorskip("prometheus_client", reason="prometheus-client not installed") import os import tempfile from unittest.mock import patch, MagicMock from prometheus_client import CollectorRegistry from bmspy.classes import UPS, BMSScalarField, BMSMultiField, BMSInfoField from bmspy.prometheus import ( prometheus_create_metric, prometheus_populate_metric, prometheus_export, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _scalar_ups(**fields): """Build a UPS with one scalar field per kwarg (name -> raw_value).""" field_dicts = { name: {"help": name, "raw_value": val, "value": str(val), "units": "V"} for name, val in fields.items() } return UPS.from_dict(field_dicts) def _make_ups_data(): """Build a dict[str, UPS] with scalar, multi, and info fields.""" return { "testups": UPS.from_dict({ "bms_voltage": { "help": "Total Voltage", "raw_value": 52.0, "value": "52.00", "units": "V", }, "bms_cells": { "help": "Cell Voltages", "label": "cell", "raw_values": {1: 3.6, 2: 3.61}, "values": {1: "3.600", 2: "3.610"}, "units": "V", }, "bms_date": { "help": "Manufacture Date", "info": "2023-01-15", }, }) } # --------------------------------------------------------------------------- # prometheus_create_metric # --------------------------------------------------------------------------- class TestPrometheusCreateMetric: def test_scalar_creates_gauge(self): registry = CollectorRegistry(auto_describe=True) ups_data = _scalar_ups(bms_voltage=52.0) metric = prometheus_create_metric(registry, {"myups": ups_data}) from prometheus_client import Gauge assert isinstance(metric["bms_voltage"], Gauge) def test_multi_creates_gauge_with_label(self): registry = CollectorRegistry(auto_describe=True) ups_data = {"myups": UPS.from_dict({ "bms_cells": { "help": "Cells", "label": "cell", "raw_values": {1: 3.6}, "values": {1: "3.600"}, "units": "V", }, })} metric = prometheus_create_metric(registry, ups_data) from prometheus_client import Gauge assert isinstance(metric["bms_cells"], Gauge) def test_info_creates_info_metric(self): registry = CollectorRegistry(auto_describe=True) ups_data = {"myups": UPS.from_dict({ "bms_date": {"help": "Date", "info": "2023-01-15"}, })} metric = prometheus_create_metric(registry, ups_data) from prometheus_client import Info assert isinstance(metric["bms_date"], Info) def test_skips_duplicate_across_ups_devices(self): registry = CollectorRegistry(auto_describe=True) field = {"help": "V", "raw_value": 52.0, "value": "52.00", "units": "V"} ups_data = { "ups1": UPS.from_dict({"bms_voltage": dict(field)}), "ups2": UPS.from_dict({"bms_voltage": dict(field)}), } metric = prometheus_create_metric(registry, ups_data) # Should only have one entry, not raise on duplicate registration assert "bms_voltage" in metric def test_all_field_types_in_one_call(self): registry = CollectorRegistry(auto_describe=True) metric = prometheus_create_metric(registry, _make_ups_data()) assert "bms_voltage" in metric assert "bms_cells" in metric assert "bms_date" in metric def test_empty_ups_data_returns_empty_dict(self): registry = CollectorRegistry(auto_describe=True) metric = prometheus_create_metric(registry, {"myups": UPS.from_dict({})}) assert metric == {} # --------------------------------------------------------------------------- # prometheus_populate_metric # --------------------------------------------------------------------------- class TestPrometheusPopulateMetric: def test_scalar_value_set(self): registry = CollectorRegistry(auto_describe=True) ups_data = {"testups": _scalar_ups(bms_voltage=52.0)} metric = prometheus_create_metric(registry, ups_data) prometheus_populate_metric(metric, ups_data) # Verify via registry output from prometheus_client import generate_latest output = generate_latest(registry).decode() assert "52.0" in output def test_multi_values_set(self): registry = CollectorRegistry(auto_describe=True) ups_data = {"testups": UPS.from_dict({ "bms_cells": { "help": "Cells", "label": "cell", "raw_values": {1: 3.6, 2: 3.61}, "values": {1: "3.600", 2: "3.610"}, "units": "V", }, })} metric = prometheus_create_metric(registry, ups_data) prometheus_populate_metric(metric, ups_data) from prometheus_client import generate_latest output = generate_latest(registry).decode() assert "3.6" in output def test_info_value_set(self): registry = CollectorRegistry(auto_describe=True) ups_data = {"testups": UPS.from_dict({ "bms_date": {"help": "Date", "info": "2023-01-15"}, })} metric = prometheus_create_metric(registry, ups_data) prometheus_populate_metric(metric, ups_data) from prometheus_client import generate_latest output = generate_latest(registry).decode() assert "2023-01-15" in output def test_missing_metric_key_is_skipped(self): """populate should not crash if a field name is not in metric dict.""" registry = CollectorRegistry(auto_describe=True) ups_data = {"testups": _scalar_ups(bms_voltage=52.0)} # Create metrics for a different field name other_metric = {} prometheus_populate_metric(other_metric, ups_data) # should not raise # --------------------------------------------------------------------------- # prometheus_export daemonize=True (start_http_server path) # --------------------------------------------------------------------------- class TestPrometheusExportDaemonize: def test_daemonize_calls_start_http_server(self): """When daemonize=True, start_http_server is called and loop runs once then exits.""" ups_data = {"testups": _scalar_ups(bms_voltage=52.0)} call_count = 0 def _read_data(*args, **kwargs): nonlocal call_count call_count += 1 return ups_data # Make time.sleep raise StopIteration after first call to exit the daemon loop with patch("bmspy.prometheus.client.read_data", side_effect=_read_data), \ patch("bmspy.prometheus.prometheus_client.start_http_server") as mock_start, \ patch("bmspy.prometheus.time.sleep", side_effect=StopIteration("stop")): with pytest.raises(StopIteration): prometheus_export(daemonize=True) mock_start.assert_called_once() # --------------------------------------------------------------------------- # prometheus_export # --------------------------------------------------------------------------- class TestPrometheusExport: def test_export_to_textfile(self, tmp_path): filename = str(tmp_path / "metrics.prom") ups_data = {"testups": _scalar_ups(bms_voltage=52.0)} with patch("bmspy.prometheus.client.read_data", return_value=ups_data): result = prometheus_export(daemonize=False, filename=filename) assert result is True assert os.path.exists(filename) def test_export_no_filename_returns_false(self): ups_data = {"testups": _scalar_ups(bms_voltage=52.0)} with patch("bmspy.prometheus.client.read_data", return_value=ups_data): result = prometheus_export(daemonize=False, filename=None) assert result is False def test_export_waits_for_data(self, tmp_path): """When read_data returns empty first, then real data, export should work.""" filename = str(tmp_path / "metrics.prom") ups_data = {"testups": _scalar_ups(bms_voltage=52.0)} call_count = 0 def _read_data(*args, **kwargs): nonlocal call_count call_count += 1 if call_count < 2: return {} return ups_data with patch("bmspy.prometheus.client.read_data", side_effect=_read_data), \ patch("bmspy.prometheus.time.sleep"): result = prometheus_export(daemonize=False, filename=filename) assert result is True assert call_count == 2 def test_export_with_all_field_types(self, tmp_path): filename = str(tmp_path / "metrics.prom") ups_data = _make_ups_data() with patch("bmspy.prometheus.client.read_data", return_value=ups_data): result = prometheus_export(daemonize=False, filename=filename) assert result is True content = open(filename).read() assert "bms_voltage" in content assert "bms_cells" in content assert "bms_date" in content # --------------------------------------------------------------------------- # prometheus main() # --------------------------------------------------------------------------- class TestPrometheusMain: def _ups_dict(self): return {"testups": _scalar_ups(bms_voltage=52.0)} def test_main_with_file_arg(self, tmp_path): from bmspy.prometheus import main filename = str(tmp_path / "metrics.prom") with patch("sys.argv", ["bmspy-prometheus", "--file", filename]), \ patch("bmspy.prometheus.client.read_data", return_value=self._ups_dict()): main() assert os.path.exists(filename) def test_main_with_socket_arg(self, tmp_path): from bmspy.prometheus import main filename = str(tmp_path / "metrics.prom") with patch("sys.argv", ["bmspy-prometheus", "--file", filename, "--socket", "/tmp/test.sock"]), \ patch("bmspy.prometheus.client.read_data", return_value=self._ups_dict()): main() def test_main_with_ups_arg(self, tmp_path): from bmspy.prometheus import main filename = str(tmp_path / "metrics.prom") with patch("sys.argv", ["bmspy-prometheus", "--file", filename, "--ups", "myups"]), \ patch("bmspy.prometheus.client.read_data", return_value=self._ups_dict()): main() def test_main_with_verbose(self, tmp_path): from bmspy.prometheus import main filename = str(tmp_path / "metrics.prom") with patch("sys.argv", ["bmspy-prometheus", "--file", filename, "-v"]), \ patch("bmspy.prometheus.client.read_data", return_value=self._ups_dict()): main()