tim
99bb321e3b
The server read each Bluetooth device every 60 seconds, and each read is one scan and one connection. An adapter that fails after many connections had no way to get fewer of them. --bluetooth-interval sets the wait after a good Bluetooth read, and the default stays at 60 seconds. Clients receive the last read during the wait. A failed read still tries again after 60 seconds, or after the interval for a shorter one.
278 lines
10 KiB
Python
278 lines
10 KiB
Python
"""Server routing and caching tests with synthetic device identifiers."""
|
|
|
|
import threading
|
|
import time
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from bmspy import server
|
|
from bmspy.classes import BMSScalarField
|
|
from bmspy.jbd_bms import JBDBMS
|
|
from bmspy.ecoflow_bt import initialise_device
|
|
|
|
ADDRESS = "02:00:00:00:00:01"
|
|
SERIAL = "R601TEST00000001"
|
|
SPEC = f"{ADDRESS}/{SERIAL}/test-user"
|
|
|
|
|
|
@pytest.mark.parametrize("spec,name,path", [
|
|
(SPEC, SERIAL, SPEC),
|
|
(f"battery@{SPEC}", "battery", SPEC),
|
|
("/tmp/serial-device", "serial-device", "/tmp/serial-device"),
|
|
("ports/ttyUSB0", "ttyUSB0", "ports/ttyUSB0"),
|
|
("office:ports/ttyUSB0", "office", "ports/ttyUSB0"),
|
|
("office:/tmp/serial-device", "office", "/tmp/serial-device"),
|
|
("/tmp/serial@device", "serial@device", "/tmp/serial@device"),
|
|
])
|
|
def test_parse_preserves_serial_paths_and_bluetooth_names(spec, name, path):
|
|
assert server.parse_device(spec) == (name, path)
|
|
assert server.is_ecoflow_device(path) is (path == SPEC)
|
|
|
|
|
|
def test_parse_bluetooth_spec():
|
|
assert server.parse_ecoflow_spec(SPEC) == (ADDRESS, SERIAL, "test-user")
|
|
assert server.parse_ecoflow_spec("02:00:00:aa:bb:01 / SERIAL / user") == (
|
|
"02:00:00:AA:BB:01", "SERIAL", "user",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("spec", [
|
|
ADDRESS, f"{ADDRESS}/{SERIAL}", f"{ADDRESS}//test-user", f"{ADDRESS}/{SERIAL}/",
|
|
f"{SPEC}/extra", "GG:00:00:00:00:01/SERIAL/test-user",
|
|
])
|
|
def test_invalid_spec_does_not_echo_account_details(spec):
|
|
assert server.is_ecoflow_device(spec)
|
|
with pytest.raises(ValueError) as error:
|
|
server.parse_ecoflow_spec(spec)
|
|
assert str(error.value) == "EcoFlow BT device spec must be ADDRESS/SERIAL/USERID"
|
|
|
|
|
|
def test_existing_positional_device_state_arguments_remain_valid():
|
|
port, data = Mock(), JBDBMS()
|
|
state = server.DeviceState(port, data, 123.0)
|
|
assert state.ser is port
|
|
assert state.data is data
|
|
assert state.timestamp == 123.0
|
|
assert state.ecoflow is None
|
|
|
|
|
|
def test_collection_dispatches_to_the_selected_backend(monkeypatch):
|
|
serial_data, bluetooth_data = JBDBMS(), JBDBMS()
|
|
for data in (serial_data, bluetooth_data):
|
|
data.bms_capacity_charge_ratio = BMSScalarField(
|
|
help="Battery Charge", raw_value=0.5, value="0.5",
|
|
)
|
|
serial_read = Mock(return_value=serial_data)
|
|
bluetooth_read = Mock(return_value=bluetooth_data)
|
|
monkeypatch.setattr(server, "collect_data", serial_read)
|
|
monkeypatch.setattr(server, "ecoflow_collect_data", bluetooth_read)
|
|
port = Mock()
|
|
config = initialise_device(ADDRESS, SERIAL, "test-user")
|
|
assert server.collect_device_data(server.DeviceState(ser=port), 2) is serial_data
|
|
serial_read.assert_called_once_with(port, 2)
|
|
assert server.collect_device_data(server.DeviceState(ecoflow=config), 3) is bluetooth_data
|
|
bluetooth_read.assert_called_once_with(config, 3, None)
|
|
assert server.collect_device_data(server.DeviceState()) is None
|
|
|
|
|
|
class StopAfter:
|
|
"""Stand-in for the stop event: record each wait and stop after *loops* polls."""
|
|
|
|
def __init__(self, loops, observe=lambda: None):
|
|
self.loops, self.observe = loops, observe
|
|
self.waits, self.seen = [], []
|
|
|
|
def is_set(self):
|
|
return len(self.waits) >= self.loops
|
|
|
|
def wait(self, delay):
|
|
self.waits.append(delay)
|
|
self.seen.append(self.observe())
|
|
|
|
|
|
@pytest.mark.parametrize("bluetooth,interval", [(False, 5), (True, 60)])
|
|
def test_poll_interval_depends_on_backend(monkeypatch, bluetooth, interval):
|
|
fresh = JBDBMS()
|
|
config = initialise_device(ADDRESS, SERIAL, "test-user") if bluetooth else None
|
|
state = server.DeviceState(ser=None if bluetooth else Mock(), ecoflow=config)
|
|
read = Mock(return_value=fresh)
|
|
monkeypatch.setattr(server, "collect_device_data", read)
|
|
monkeypatch.setattr(server.time, "time", lambda: 100)
|
|
stop = StopAfter(1)
|
|
server.poll_device(state, stop, 2)
|
|
assert state.data is fresh
|
|
assert state.timestamp == 100
|
|
assert state.polled.is_set()
|
|
assert stop.waits == [interval]
|
|
read.assert_called_once_with(state, 2, stop)
|
|
|
|
|
|
def test_offline_bluetooth_does_not_retry_at_once_or_serve_stale_data(monkeypatch):
|
|
state = server.DeviceState(
|
|
data=JBDBMS(), timestamp=1, ecoflow=initialise_device(ADDRESS, SERIAL, "test-user"),
|
|
)
|
|
monkeypatch.setattr(server, "collect_device_data", Mock(return_value=None))
|
|
monkeypatch.setattr(server.time, "time", lambda: 100)
|
|
stop = StopAfter(1)
|
|
server.poll_device(state, stop)
|
|
assert state.data is None
|
|
assert state.timestamp == 100
|
|
# A client must not wait for a device that is known to be offline.
|
|
assert state.polled.is_set()
|
|
assert stop.waits == [60]
|
|
|
|
|
|
@pytest.mark.parametrize("interval,read,wait", [
|
|
(300, JBDBMS(), 300),
|
|
# A failed read retries after 60 seconds, or sooner with a shorter interval.
|
|
(300, None, 60),
|
|
(30, None, 30),
|
|
])
|
|
def test_bluetooth_interval_sets_the_wait_after_a_good_read(monkeypatch, interval, read, wait):
|
|
state = server.DeviceState(
|
|
ecoflow=initialise_device(ADDRESS, SERIAL, "test-user"), poll_seconds=interval,
|
|
)
|
|
monkeypatch.setattr(server, "collect_device_data", Mock(return_value=read))
|
|
stop = StopAfter(1)
|
|
server.poll_device(state, stop)
|
|
assert state.data is read
|
|
assert stop.waits == [wait]
|
|
|
|
|
|
def _main_devices(monkeypatch, tmp_path, *arguments):
|
|
"""Run main() until it starts to poll, and return the devices it registered."""
|
|
start = Mock(side_effect=KeyboardInterrupt)
|
|
monkeypatch.setattr(server, "start_polling", start)
|
|
monkeypatch.setattr(server.signal, "signal", Mock())
|
|
monkeypatch.setattr(server.atexit, "register", Mock())
|
|
monkeypatch.setattr(server, "initialise_serial", Mock())
|
|
monkeypatch.setattr(server.os.path, "isdir", lambda path: True)
|
|
monkeypatch.setattr("sys.argv", [
|
|
"bmspy-server", "--socket", str(tmp_path / "interval.sock"),
|
|
"--device", f"river@{SPEC}", "--device", "pack:/dev/ttyUSB0", *arguments,
|
|
])
|
|
with pytest.raises((SystemExit, KeyboardInterrupt)):
|
|
server.main()
|
|
return start.call_args.args[0]
|
|
|
|
|
|
@pytest.mark.parametrize("arguments,interval", [((), 60), (("--bluetooth-interval", "300"), 300)])
|
|
def test_bluetooth_interval_option_applies_to_bluetooth_devices(
|
|
monkeypatch, tmp_path, arguments, interval,
|
|
):
|
|
devices = _main_devices(monkeypatch, tmp_path, *arguments)
|
|
assert devices["river"].poll_seconds == interval
|
|
assert devices["river"].ecoflow is not None
|
|
assert devices["pack"].ecoflow is None
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["0", "-5", "soon"])
|
|
def test_bluetooth_interval_option_rejects_an_unusable_value(monkeypatch, capsys, value):
|
|
monkeypatch.setattr(server.signal, "signal", Mock())
|
|
monkeypatch.setattr("sys.argv", ["bmspy-server", "--bluetooth-interval", value])
|
|
with pytest.raises(SystemExit) as exit_info:
|
|
server.main()
|
|
assert exit_info.value.code == 2
|
|
assert "--bluetooth-interval" in capsys.readouterr().err
|
|
|
|
|
|
def test_failed_serial_read_retries_soon_and_hides_a_short_failure(monkeypatch):
|
|
data = JBDBMS()
|
|
state = server.DeviceState(ser=Mock())
|
|
monkeypatch.setattr(server, "collect_device_data", Mock(side_effect=[data, None, None]))
|
|
clock = iter([100, 110, 131])
|
|
monkeypatch.setattr(server.time, "time", lambda: next(clock))
|
|
stop = StopAfter(3, observe=lambda: state.data)
|
|
server.poll_device(state, stop)
|
|
# The last snapshot stays for 30 seconds, then the device reads as absent.
|
|
assert stop.seen == [data, data, None]
|
|
assert stop.waits == [5, 1, 1]
|
|
|
|
|
|
def test_collection_error_does_not_end_polling(monkeypatch, capsys):
|
|
data = JBDBMS()
|
|
state = server.DeviceState(ser=Mock())
|
|
monkeypatch.setattr(server, "collect_device_data", Mock(side_effect=[RuntimeError("boom"), data]))
|
|
stop = StopAfter(2)
|
|
server.poll_device(state, stop)
|
|
assert state.data is data
|
|
assert stop.waits == [1, 5]
|
|
assert "collection error: boom" in capsys.readouterr().out
|
|
|
|
|
|
def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch):
|
|
data = JBDBMS()
|
|
offline_started, release = threading.Event(), threading.Event()
|
|
working, offline = server.DeviceState(ser=Mock()), server.DeviceState(ser=Mock())
|
|
|
|
def read(device, debug=0, stop=None):
|
|
if device is offline:
|
|
offline_started.set()
|
|
release.wait(5)
|
|
return None
|
|
return data
|
|
|
|
monkeypatch.setattr(server, "collect_device_data", read)
|
|
stop = threading.Event()
|
|
threads = server.start_polling({"offline": offline, "working": working}, stop)
|
|
try:
|
|
assert offline_started.wait(5)
|
|
assert working.polled.wait(5)
|
|
assert working.data is data
|
|
assert not offline.polled.is_set()
|
|
finally:
|
|
stop.set()
|
|
release.set()
|
|
for thread in threads:
|
|
thread.join(5)
|
|
assert not any(thread.is_alive() for thread in threads)
|
|
|
|
|
|
def test_stop_polling_waits_for_the_bluetooth_disconnect(monkeypatch):
|
|
events = []
|
|
|
|
def bluetooth_read(config, debug=0, stop=None):
|
|
if stop.is_set():
|
|
return False
|
|
events.append("read started")
|
|
assert stop.wait(5)
|
|
events.append("disconnected")
|
|
return False
|
|
|
|
monkeypatch.setattr(server, "ecoflow_collect_data", bluetooth_read)
|
|
# The second device waits for the first one. It must not start a read.
|
|
devices = {
|
|
name: server.DeviceState(ecoflow=initialise_device(ADDRESS, SERIAL, "test-user"))
|
|
for name in ("first", "second")
|
|
}
|
|
stop = threading.Event()
|
|
threads = server.start_polling(devices, stop)
|
|
while not events:
|
|
time.sleep(0.01)
|
|
server.stop_polling(threads, stop)
|
|
assert events == ["read started", "disconnected"]
|
|
assert not any(thread.is_alive() for thread in threads)
|
|
|
|
|
|
def test_stop_polling_reports_a_read_that_does_not_end(monkeypatch, capsys):
|
|
started, release = threading.Event(), threading.Event()
|
|
|
|
def stuck_read(device, debug=0, stop=None):
|
|
started.set()
|
|
release.wait(5)
|
|
|
|
monkeypatch.setattr(server, "collect_device_data", stuck_read)
|
|
monkeypatch.setattr(server, "POLL_STOP_TIMEOUT", 0.05)
|
|
stop = threading.Event()
|
|
threads = server.start_polling({"stuck": server.DeviceState(ser=Mock())}, stop)
|
|
try:
|
|
assert started.wait(5)
|
|
server.stop_polling(threads, stop)
|
|
assert "server: no clean stop in 0.05 seconds for poll-stuck" in capsys.readouterr().out
|
|
finally:
|
|
release.set()
|
|
for thread in threads:
|
|
thread.join(5)
|
|
assert not any(thread.is_alive() for thread in threads)
|