sigenergy-prometheus-exporter/test_sigenergy_prometheus_exporter.py at main · edricklabs/sigenergy-prometheus-exporter · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
import importlib
import sys
import time
from pathlib import Path
from unittest.mock import Mock

import pytest


sys.path.insert(0, str(Path(__file__).resolve().parent))
exporter = importlib.import_module("sigenergy_prometheus_exporter")


def test_load_config_accepts_sample_mode(monkeypatch):
    monkeypatch.delenv("SIGENERGY_HOST", raising=False)
    monkeypatch.setenv("SIGENERGY_MODE", "sample")
    cfg = exporter.load_config()
    assert cfg.mode == "sample"


def test_load_config_rejects_invalid_mode(monkeypatch):
    monkeypatch.setenv("SIGENERGY_MODE", "bad")
    with pytest.raises(exporter.ConfigError, match="SIGENERGY_MODE"):
        exporter.load_config()


def test_load_config_requires_host_in_modbus(monkeypatch):
    monkeypatch.setenv("SIGENERGY_MODE", "modbus")
    monkeypatch.delenv("SIGENERGY_HOST", raising=False)
    with pytest.raises(exporter.ConfigError, match="SIGENERGY_HOST"):
        exporter.load_config()


def test_sample_collection_sets_observability_metrics(monkeypatch):
    seq = iter([0, 0, 0, 0])
    monkeypatch.setattr(exporter.random, "uniform", lambda a, b: next(seq))
    monkeypatch.setattr(exporter.time, "time", lambda: 123.0)
    exporter.collect_sample_metrics()
    assert exporter.sigenergy_device_up._value.get() == 1
    assert exporter.sigenergy_scrape_success._value.get() == 1
    assert exporter.sigenergy_last_success_timestamp_seconds._value.get() == 123.0
    assert exporter.pv_power_watts._value.get() == 4200.0
    assert exporter.house_load_power_watts._value.get() == 2600.0
    assert exporter.battery_power_watts._value.get() == -900.0
    assert exporter.load_supplied_by_solar_watts._value.get() == 2600.0
    assert exporter.load_supplied_by_grid_watts._value.get() == 0.0
    assert exporter.pv_to_battery_watts._value.get() == 900.0
    assert exporter.battery_charge_from_solar_watts._value.get() == 900.0


def test_derive_power_flows_splits_load_and_pv():
    derived = exporter._derive_power_flows(
        pv_power=5000,
        house_load_power=3000,
        battery_charge_power=800,
        battery_discharge_power=400,
        grid_import=200,
        grid_export=1200,
    )

    assert derived["load_from_solar"] == 3000
    assert derived["load_from_battery"] == 0
    assert derived["load_from_grid"] == 0
    assert derived["pv_to_battery"] == 800
    assert derived["pv_to_grid"] == 1200
    assert derived["battery_charge_from_solar"] == 800
    assert derived["battery_charge_from_grid"] == 0
    assert derived["pv_to_load_pct"] == 60.0


def test_import_tariff_schedule(monkeypatch):
    monkeypatch.setenv(
        "SIGENERGY_IMPORT_TARIFF_SCHEDULE",
        "0-2:16.93,2-4:9.94,4-8:16.93,8-23:34.34,23-24:16.93",
    )

    def localtime_at(hour, minute=0):
        return time.struct_time((2026, 1, 1, hour, minute, 0, 0, 1, -1))

    monkeypatch.setattr(exporter.time, "localtime", lambda timestamp: localtime_at(8))
    assert exporter._current_import_tariff_cents_per_kwh(0) == 34.34
    monkeypatch.setattr(exporter.time, "localtime", lambda timestamp: localtime_at(23))
    assert exporter._current_import_tariff_cents_per_kwh(0) == 16.93
    monkeypatch.setattr(exporter.time, "localtime", lambda timestamp: localtime_at(2))
    assert exporter._current_import_tariff_cents_per_kwh(0) == 9.94
    monkeypatch.setattr(exporter.time, "localtime", lambda timestamp: localtime_at(4))
    assert exporter._current_import_tariff_cents_per_kwh(0) == 16.93


def test_calculate_cost_rates():
    rates = exporter._calculate_cost_rates(
        grid_import_watts=1000,
        grid_export_watts=2000,
        avoided_import_watts=500,
        import_tariff_cents_per_kwh=34.34,
        export_tariff_cents_per_kwh=19.5,
    )

    assert rates["import_cost_cents_per_hour"] == 34.34
    assert rates["export_credit_cents_per_hour"] == 39.0
    assert rates["avoided_savings_cents_per_hour"] == 17.17
    assert rates["net_grid_cost_cents_per_hour"] == pytest.approx(-4.66)


def test_modbus_collection_reads_registers_and_scales(monkeypatch):
    client = Mock()
    client.connect.return_value = True
    client.close.return_value = None
    client.read_holding_registers.side_effect = [
        Mock(isError=lambda: False, registers=[0xFFFF, 0xFC00]),  # grid active -1024 W => export
        Mock(isError=lambda: False, registers=[1000]),  # SoC 100.0%
        Mock(isError=lambda: False, registers=[0, 1300]),  # plant active
        Mock(isError=lambda: False, registers=[0, 1800]),  # PV
        Mock(isError=lambda: False, registers=[0, 500]),  # ESS charging 500 W
        Mock(isError=lambda: False, registers=[0, 300]),  # house load
    ]
    monkeypatch.setattr(exporter.time, "time", lambda: 1000.0)
    monkeypatch.setattr(exporter.time, "monotonic", lambda: 10.0)
    cfg = exporter.AppConfig("modbus", 8000, "1.2.3.4", 502, 247, 1, 5.0, 15.0)

    exporter.collect_modbus_metrics(cfg, client_factory=lambda **kwargs: client)

    assert client.read_holding_registers.call_args_list[0].kwargs["address"] == 30005
    assert client.read_holding_registers.call_args_list[0].kwargs["slave"] == 247
    assert client.read_holding_registers.call_args_list[1].kwargs["address"] == 30014
    assert exporter.battery_state_of_charge_percent._value.get() == 100.0
    assert exporter.grid_active_power_watts._value.get() == -1024.0
    assert exporter.grid_export_power_watts._value.get() == 1024.0
    assert exporter.pv_power_watts._value.get() == 1800.0
    assert exporter.house_load_power_watts._value.get() == 300.0
    assert exporter.battery_charge_power_watts._value.get() == 500.0
    assert exporter.pv_to_load_watts._value.get() == 300.0
    assert exporter.pv_to_battery_watts._value.get() == 500.0
    assert exporter.sigenergy_device_up._value.get() == 1
    assert exporter.sigenergy_scrape_success._value.get() == 1
    assert exporter.sigenergy_last_success_timestamp_seconds._value.get() == 1000.0
    assert exporter.sigenergy_collection_duration_seconds._value.get() == 0.0
    client.close.assert_called_once()


def test_modbus_collection_failure_updates_error_metrics(monkeypatch):
    client = Mock()
    client.connect.return_value = True
    client.close.return_value = None
    client.read_holding_registers.return_value = Mock(isError=lambda: True, registers=[])
    monkeypatch.setattr(exporter.time, "monotonic", lambda: 1.0)
    cfg = exporter.AppConfig("modbus", 8000, "1.2.3.4", 502, 247, 1, 5.0, 15.0)

    with pytest.raises(RuntimeError, match="failed to read holding register"):
        exporter.collect_modbus_metrics(cfg, client_factory=lambda **kwargs: client)

    assert exporter.sigenergy_device_up._value.get() == 0
    assert exporter.sigenergy_scrape_success._value.get() == 0
    assert exporter.sigenergy_collection_errors_total._value.get() >= 1


def test_holding_register_logical_address_conversion():
    assert exporter._holding_register_pdu_address(40001) == 0
    assert exporter._holding_register_pdu_address(40002) == 1
    assert exporter._holding_register_pdu_address(12) == 12