tim
f36bb2510c
Correct the spelling in the JBD and EcoFlow snapshots, the diagnostic tool, and the tests. UPS.from_dict maps the old name to the new name, so a client still reads data from an older server during an upgrade. InfluxDB and Prometheus now receive this measurement under the new name.
369 lines
14 KiB
Python
369 lines
14 KiB
Python
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
from bmspy.classes import BMSInfoField, BMSMultiField, BMSScalarField, UPS
|
|
from bmspy.client import read_data
|
|
|
|
|
|
MOCK_RESPONSE = {
|
|
"myups": {
|
|
"bms_voltage_total_volts": {
|
|
"help": "Total Voltage",
|
|
"raw_value": 52.0,
|
|
"value": "52.00",
|
|
"units": "V",
|
|
},
|
|
"bms_manufacture_date": {
|
|
"help": "Date of Manufacture",
|
|
"info": "2023-01-15",
|
|
},
|
|
"bms_temperature_celsius": {
|
|
"help": "Temperature",
|
|
"label": "sensor",
|
|
"raw_values": {1: 25.0},
|
|
"values": {1: "25.00"},
|
|
"units": "°C",
|
|
},
|
|
},
|
|
"officeups": {
|
|
"bms_voltage_total_volts": {
|
|
"help": "Total Voltage",
|
|
"raw_value": 48.5,
|
|
"value": "48.50",
|
|
"units": "V",
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class TestReadData:
|
|
def _call(self, response=None):
|
|
with patch("bmspy.client.socket_comms", return_value=response or MOCK_RESPONSE):
|
|
return read_data("/fake/socket", "test")
|
|
|
|
def test_returns_dict_of_ups(self):
|
|
result = self._call()
|
|
assert isinstance(result, dict)
|
|
for v in result.values():
|
|
assert isinstance(v, UPS)
|
|
|
|
def test_all_devices_present(self):
|
|
result = self._call()
|
|
assert set(result.keys()) == {"myups", "officeups"}
|
|
|
|
def test_scalar_field_deserialized(self):
|
|
result = self._call()
|
|
items = dict(result["myups"].items())
|
|
f = items["bms_voltage_total_volts"]
|
|
assert isinstance(f, BMSScalarField)
|
|
assert f.raw_value == 52.0
|
|
assert f.units == "V"
|
|
|
|
def test_info_field_deserialized(self):
|
|
result = self._call()
|
|
items = dict(result["myups"].items())
|
|
f = items["bms_manufacture_date"]
|
|
assert isinstance(f, BMSInfoField)
|
|
assert f.info == "2023-01-15"
|
|
|
|
def test_multi_field_deserialized(self):
|
|
result = self._call()
|
|
items = dict(result["myups"].items())
|
|
f = items["bms_temperature_celsius"]
|
|
assert isinstance(f, BMSMultiField)
|
|
assert f.raw_values == {1: 25.0}
|
|
assert f.label == "sensor"
|
|
|
|
def test_ups_is_truthy_when_populated(self):
|
|
result = self._call()
|
|
assert bool(result["myups"]) is True
|
|
|
|
def test_empty_device_response(self):
|
|
result = self._call({"emptyups": {}})
|
|
assert isinstance(result["emptyups"], UPS)
|
|
assert bool(result["emptyups"]) is False
|
|
|
|
def test_ups_filter_forwarded(self):
|
|
with patch("bmspy.client.socket_comms") as mock_comms:
|
|
mock_comms.return_value = {"myups": MOCK_RESPONSE["myups"]}
|
|
read_data("/fake/socket", "test", ups="myups")
|
|
call_args = mock_comms.call_args[0][1]
|
|
assert call_args.get("ups") == "myups"
|
|
|
|
def test_no_ups_filter_not_in_request(self):
|
|
with patch("bmspy.client.socket_comms") as mock_comms:
|
|
mock_comms.return_value = {}
|
|
read_data("/fake/socket", "test")
|
|
call_args = mock_comms.call_args[0][1]
|
|
assert "ups" not in call_args
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# handle_registration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import bmspy.client as _client_mod
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_is_registered():
|
|
_client_mod.is_registered = False
|
|
yield
|
|
_client_mod.is_registered = False
|
|
|
|
|
|
class TestHandleRegistration:
|
|
def test_register_sets_flag(self):
|
|
with patch("bmspy.client.socket_comms", return_value={"status": "REGISTERED", "client": "test"}):
|
|
from bmspy.client import handle_registration
|
|
handle_registration("/fake/socket", "test")
|
|
assert _client_mod.is_registered is True
|
|
|
|
def test_register_returns_response(self):
|
|
response = {"status": "REGISTERED", "client": "test"}
|
|
with patch("bmspy.client.socket_comms", return_value=response):
|
|
from bmspy.client import handle_registration
|
|
result = handle_registration("/fake/socket", "test")
|
|
assert result == response
|
|
|
|
def test_register_sends_register_command(self):
|
|
with patch("bmspy.client.socket_comms") as mock_comms:
|
|
mock_comms.return_value = {"status": "REGISTERED", "client": "test"}
|
|
from bmspy.client import handle_registration
|
|
handle_registration("/fake/socket", "test")
|
|
sent = mock_comms.call_args[0][1]
|
|
assert sent["command"] == "REGISTER"
|
|
assert sent["client"] == "test"
|
|
|
|
def test_deregister_clears_flag(self):
|
|
_client_mod.is_registered = True
|
|
with patch("bmspy.client.socket_comms", return_value={"status": "DEREGISTERED", "client": "test"}):
|
|
from bmspy.client import handle_registration
|
|
handle_registration("/fake/socket", "test")
|
|
assert _client_mod.is_registered is False
|
|
|
|
def test_deregister_sends_deregister_command(self):
|
|
_client_mod.is_registered = True
|
|
with patch("bmspy.client.socket_comms") as mock_comms:
|
|
mock_comms.return_value = {"status": "DEREGISTERED", "client": "test"}
|
|
from bmspy.client import handle_registration
|
|
handle_registration("/fake/socket", "test")
|
|
sent = mock_comms.call_args[0][1]
|
|
assert sent["command"] == "DEREGISTER"
|
|
|
|
def test_socket_error_does_not_raise(self):
|
|
with patch("bmspy.client.socket_comms", side_effect=Exception("connection refused")):
|
|
from bmspy.client import handle_registration
|
|
result = handle_registration("/fake/socket", "test")
|
|
assert result == {}
|
|
|
|
def test_invalid_status_does_not_raise(self):
|
|
with patch("bmspy.client.socket_comms", return_value={"status": "UNKNOWN"}):
|
|
from bmspy.client import handle_registration
|
|
handle_registration("/fake/socket", "test")
|
|
|
|
def test_debug_3_on_deregister_failure(self, capsys):
|
|
"""debug=3 path in exception handler when is_registered=True."""
|
|
_client_mod.is_registered = True
|
|
with patch("bmspy.client.socket_comms", side_effect=Exception("fail")):
|
|
from bmspy.client import handle_registration
|
|
handle_registration("/fake/socket", "test", debug=3)
|
|
captured = capsys.readouterr()
|
|
assert "deregister" in captured.out.lower() or "fail" in captured.out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# socket_comms — real Unix socket
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import json
|
|
import socket
|
|
import struct
|
|
import threading
|
|
|
|
|
|
class TestSocketComms:
|
|
"""Test socket_comms with a real Unix socket server running in a thread."""
|
|
|
|
def _run_server(self, sock_path: str, response: dict, ready_event: threading.Event):
|
|
"""Minimal server: accept one connection, read framed request, send framed response."""
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(sock_path)
|
|
srv.listen(1)
|
|
ready_event.set()
|
|
conn, _ = srv.accept()
|
|
try:
|
|
# read length
|
|
raw_len = conn.recv(4)
|
|
if len(raw_len) < 4:
|
|
return # Client disconnected early
|
|
length = struct.unpack("!I", raw_len)[0]
|
|
data = conn.recv(length)
|
|
# send response
|
|
payload = json.dumps(response).encode()
|
|
conn.sendall(struct.pack("!I", len(payload)) + payload)
|
|
finally:
|
|
conn.close()
|
|
srv.close()
|
|
import os
|
|
try:
|
|
os.unlink(sock_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
def test_returns_response_dict(self, tmp_path):
|
|
from bmspy.client import socket_comms
|
|
sock_path = str(tmp_path / "test.sock")
|
|
response = {"status": "REGISTERED", "client": "test"}
|
|
ready = threading.Event()
|
|
t = threading.Thread(
|
|
target=self._run_server, args=(sock_path, response, ready), daemon=True
|
|
)
|
|
t.start()
|
|
ready.wait(timeout=2)
|
|
result = socket_comms(sock_path, {"command": "REGISTER", "client": "test"})
|
|
t.join(timeout=2)
|
|
assert result == response
|
|
|
|
def test_debug_3_does_not_raise(self, tmp_path, capsys):
|
|
from bmspy.client import socket_comms
|
|
sock_path = str(tmp_path / "test_dbg.sock")
|
|
response = {"status": "OK"}
|
|
ready = threading.Event()
|
|
t = threading.Thread(
|
|
target=self._run_server, args=(sock_path, response, ready), daemon=True
|
|
)
|
|
t.start()
|
|
ready.wait(timeout=2)
|
|
result = socket_comms(sock_path, {"command": "GET", "client": "test"}, debug=3)
|
|
t.join(timeout=2)
|
|
assert result == response
|
|
|
|
def test_debug_4_does_not_raise(self, tmp_path, capsys):
|
|
from bmspy.client import socket_comms
|
|
sock_path = str(tmp_path / "test_dbg4.sock")
|
|
response = {"status": "OK"}
|
|
ready = threading.Event()
|
|
t = threading.Thread(
|
|
target=self._run_server, args=(sock_path, response, ready), daemon=True
|
|
)
|
|
t.start()
|
|
ready.wait(timeout=2)
|
|
result = socket_comms(sock_path, {"command": "GET", "client": "test"}, debug=4)
|
|
t.join(timeout=2)
|
|
assert result == response
|
|
|
|
def test_debug_5_does_not_raise(self, tmp_path, capsys):
|
|
from bmspy.client import socket_comms
|
|
sock_path = str(tmp_path / "test_dbg5.sock")
|
|
response = {"status": "OK"}
|
|
ready = threading.Event()
|
|
t = threading.Thread(
|
|
target=self._run_server, args=(sock_path, response, ready), daemon=True
|
|
)
|
|
t.start()
|
|
ready.wait(timeout=2)
|
|
result = socket_comms(sock_path, {"command": "GET", "client": "test"}, debug=5)
|
|
t.join(timeout=2)
|
|
assert result == response
|
|
|
|
def test_connection_refused_does_not_raise(self, tmp_path, capsys):
|
|
"""socket_comms gracefully handles ENOENT (no server)."""
|
|
from bmspy.client import socket_comms
|
|
import sys
|
|
sock_path = str(tmp_path / "nonexistent.sock")
|
|
# socket_comms calls sys.exit(1) on encode errors, but connection failures
|
|
# just print and continue (then fail on recv). We expect SystemExit.
|
|
with pytest.raises((SystemExit, OSError, Exception)):
|
|
socket_comms(sock_path, {"command": "GET", "client": "test"})
|
|
|
|
def _run_bad_response_server(self, sock_path: str, bad_payload: bytes, ready_event: threading.Event):
|
|
"""Server that sends a bad (non-JSON) response."""
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(sock_path)
|
|
srv.listen(1)
|
|
ready_event.set()
|
|
conn, _ = srv.accept()
|
|
try:
|
|
# read request
|
|
raw_len = conn.recv(4)
|
|
length = struct.unpack("!I", raw_len)[0]
|
|
conn.recv(length)
|
|
# send bad response
|
|
conn.sendall(struct.pack("!I", len(bad_payload)) + bad_payload)
|
|
finally:
|
|
conn.close()
|
|
srv.close()
|
|
import os
|
|
try:
|
|
os.unlink(sock_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
def test_invalid_json_response_exits(self, tmp_path, capsys):
|
|
"""socket_comms calls sys.exit(1) on invalid JSON response."""
|
|
from bmspy.client import socket_comms
|
|
sock_path = str(tmp_path / "bad_json.sock")
|
|
bad_payload = b"not json!!!"
|
|
ready = threading.Event()
|
|
t = threading.Thread(
|
|
target=self._run_bad_response_server,
|
|
args=(sock_path, bad_payload, ready),
|
|
daemon=True,
|
|
)
|
|
t.start()
|
|
ready.wait(timeout=2)
|
|
with pytest.raises(SystemExit):
|
|
socket_comms(sock_path, {"command": "GET", "client": "test"})
|
|
t.join(timeout=2)
|
|
|
|
def test_json_encode_failure_exits(self, tmp_path, capsys):
|
|
"""socket_comms calls sys.exit(1) when json.dumps raises."""
|
|
from bmspy.client import socket_comms
|
|
sock_path = str(tmp_path / "encode_err.sock")
|
|
# Create a minimal server so the socket connect succeeds
|
|
ready = threading.Event()
|
|
response = {"status": "OK"}
|
|
t = threading.Thread(
|
|
target=self._run_server, args=(sock_path, response, ready), daemon=True
|
|
)
|
|
t.start()
|
|
ready.wait(timeout=2)
|
|
with patch("bmspy.client.json.dumps", side_effect=TypeError("not serializable")):
|
|
with pytest.raises(SystemExit):
|
|
socket_comms(sock_path, {"command": "GET", "client": "test"})
|
|
t.join(timeout=2)
|
|
|
|
def test_non_enoent_socket_error_logs_message(self, tmp_path, capsys):
|
|
"""socket_comms logs a different message for non-ENOENT socket errors."""
|
|
from bmspy.client import socket_comms
|
|
import socket as _socket
|
|
import errno
|
|
|
|
sock_path = str(tmp_path / "test_err.sock")
|
|
# Make connect raise a socket.error with errno != 2
|
|
err = _socket.error("connection refused")
|
|
err.errno = errno.ECONNREFUSED # not 2 (ENOENT)
|
|
|
|
with patch("bmspy.client.socket.socket") as mock_sock_cls:
|
|
mock_sock = MagicMock()
|
|
mock_sock_cls.return_value = mock_sock
|
|
mock_sock.connect.side_effect = err
|
|
# after connect fails, sendall will also fail, triggering sys.exit
|
|
with pytest.raises((SystemExit, Exception)):
|
|
socket_comms(sock_path, {"command": "GET", "client": "test"})
|
|
captured = capsys.readouterr()
|
|
assert "socket client" in captured.out or "connection refused" in captured.out.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# read_data — socket_comms returns None path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReadDataNone:
|
|
def test_raises_runtime_error_when_none(self):
|
|
with patch("bmspy.client.socket_comms", return_value=None):
|
|
with pytest.raises(RuntimeError, match="No data received"):
|
|
read_data("/fake/socket", "test")
|