ARIA
b8e91e3c91
A stale venv after a code update (editable install + git pull) left pyserial out of the environment, and the hard module-level import in wireview.py took the entire web server down — even on machines without a WireView device. Guard the import: without pyserial the serial transport is disabled (one-time warning, connect fails, reads return None) while the rest of the server keeps running. The hwmon transport is unaffected. Also simplify the except clauses to OSError (SerialException is an OSError subclass) so they no longer reference the possibly-None module.
529 lines
18 KiB
Python
529 lines
18 KiB
Python
"""Tests for the native WireView Pro II reader (nvcurve/wireview.py).
|
|
|
|
Standalone (no pytest required):
|
|
|
|
python tests/test_wireview.py
|
|
|
|
Also works under pytest. Covers the sensor-frame parser, corruption
|
|
detection, fault decoding, USB port discovery, and the hwmon (sysfs)
|
|
transport.
|
|
"""
|
|
|
|
import builtins
|
|
import io
|
|
import os
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from nvcurve import wireview as wv # noqa: E402
|
|
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
|
|
def check(name: str, cond: bool) -> None:
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
FAIL += 1
|
|
print(f" FAIL {name}")
|
|
|
|
|
|
def make_frame(
|
|
ts=(392, 352, 273, 395),
|
|
vdd=11900,
|
|
fan=0,
|
|
pins=((11904, 6221, 74055),) * 6,
|
|
total_power=450568,
|
|
total_current=37863,
|
|
avg_voltage=11900,
|
|
psu_cap=0,
|
|
fault_status=0,
|
|
fault_log=0,
|
|
pad1=0,
|
|
pad2=0,
|
|
) -> bytes:
|
|
"""Build a 100-byte sensor frame with controllable fields."""
|
|
frame = struct.pack("<4hHB", *ts, vdd, fan)
|
|
frame += bytes([pad1])
|
|
for voltage, current, power in pins:
|
|
frame += struct.pack("<hxxII", voltage, current, power)
|
|
frame += struct.pack("<IIHB", total_power, total_current, avg_voltage, psu_cap)
|
|
frame += bytes([pad2])
|
|
frame += struct.pack("<HH", fault_status, fault_log)
|
|
assert len(frame) == wv.SENSOR_STRUCT_SIZE
|
|
return frame
|
|
|
|
|
|
# ── Sensor frame parser ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_parse_sensor_struct():
|
|
s = wv.parse_sensor_struct(make_frame())
|
|
check("temp in", abs(s["temp_in_c"] - 39.2) < 1e-9)
|
|
check("temp out", abs(s["temp_out_c"] - 35.2) < 1e-9)
|
|
check("temp ext1", abs(s["temp_ext1_c"] - 27.3) < 1e-9)
|
|
check("temp ext2", abs(s["temp_ext2_c"] - 39.5) < 1e-9)
|
|
check("fan duty", s["fan_duty_pct"] == 0)
|
|
check("psu capability", s["psu_capability_w"] == 600)
|
|
check("fault status", s["fault_status"] == 0)
|
|
check("fault log", s["fault_log"] == 0)
|
|
check("6 pins", len(s["pins"]) == 6)
|
|
pin = s["pins"][0]
|
|
check("pin voltage", abs(pin["voltage_v"] - 11.904) < 1e-9)
|
|
check("pin current", abs(pin["current_a"] - 6.221) < 1e-9)
|
|
check("pin power", abs(pin["power_w"] - 74.055) < 1e-9)
|
|
# Totals are computed from the pins (exporter behavior), not the
|
|
# device's total fields.
|
|
check(
|
|
"total current",
|
|
abs(s["current_total_a"] - round(6 * 6.221, 3)) < 1e-9,
|
|
)
|
|
check(
|
|
"total power",
|
|
abs(s["power_total_w"] - round(6 * 11.904 * 6.221, 3)) < 1e-9,
|
|
)
|
|
check(
|
|
"avg voltage",
|
|
abs(s["voltage_avg_v"] - 11.904) < 1e-3,
|
|
)
|
|
|
|
|
|
def test_parse_psu_capabilities():
|
|
for cap, watts in ((0, 600), (1, 450), (2, 300), (3, 150)):
|
|
s = wv.parse_sensor_struct(make_frame(psu_cap=cap))
|
|
check(f"psu cap {cap} -> {watts}W", s["psu_capability_w"] == watts)
|
|
|
|
|
|
def test_parse_negative_temps():
|
|
s = wv.parse_sensor_struct(make_frame(ts=(-10, 0, 555, 999)))
|
|
check("negative temp", abs(s["temp_in_c"] - (-1.0)) < 1e-9)
|
|
check("zero temp", s["temp_out_c"] == 0.0)
|
|
check("high temp", abs(s["temp_ext2_c"] - 99.9) < 1e-9)
|
|
|
|
|
|
# ── Corruption detection ──────────────────────────────────────────────────────
|
|
|
|
|
|
def test_corruption_check():
|
|
check("clean frame accepted", not wv.sensor_frame_is_corrupt(make_frame()))
|
|
check(
|
|
"fan duty > 100 rejected",
|
|
wv.sensor_frame_is_corrupt(make_frame(fan=101)),
|
|
)
|
|
check(
|
|
"fan duty 100 accepted",
|
|
not wv.sensor_frame_is_corrupt(make_frame(fan=100)),
|
|
)
|
|
check("pad1 dirty rejected", wv.sensor_frame_is_corrupt(make_frame(pad1=1)))
|
|
check("pad2 dirty rejected", wv.sensor_frame_is_corrupt(make_frame(pad2=1)))
|
|
check(
|
|
"short frame rejected",
|
|
wv.sensor_frame_is_corrupt(make_frame()[:50]),
|
|
)
|
|
|
|
|
|
# ── Fault decoding ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_decode_faults():
|
|
check("no faults", wv.decode_faults(0) == [])
|
|
check(
|
|
"chip over-temp",
|
|
wv.decode_faults(1) == ["Chip over-temperature"],
|
|
)
|
|
check(
|
|
"over-power",
|
|
wv.decode_faults(16) == ["Over-power (OPP)"],
|
|
)
|
|
check(
|
|
"multiple faults",
|
|
wv.decode_faults(1 | 32)
|
|
== ["Chip over-temperature", "Current imbalance"],
|
|
)
|
|
check(
|
|
"unknown bits ignored",
|
|
wv.decode_faults(1 | 0x80) == ["Chip over-temperature"],
|
|
)
|
|
|
|
|
|
def test_product_support():
|
|
check("Pro II supported", wv.is_supported_product(0xEF, 0x05))
|
|
check("Noctua Edition supported", wv.is_supported_product(0xEF, 0x06))
|
|
check("WireView II unsupported", not wv.is_supported_product(0xEF, 0x07))
|
|
check("other vendor unsupported", not wv.is_supported_product(0x12, 0x05))
|
|
|
|
|
|
# ── USB port discovery ────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_find_wireview_ports():
|
|
"""Fake a sysfs tree with one WireView (ttyACM0) and one other CDC
|
|
device (ttyACM1), plus the udev symlink."""
|
|
contents = {
|
|
"/sys/devices/fake/usb0/idVendor": "0483\n",
|
|
"/sys/devices/fake/usb0/idProduct": "5740\n",
|
|
"/sys/devices/fake/usb1/idVendor": "1234\n",
|
|
"/sys/devices/fake/usb1/idProduct": "5678\n",
|
|
}
|
|
|
|
def fake_islink(p):
|
|
return p == "/dev/wireview-pro2"
|
|
|
|
def fake_realpath(p):
|
|
return {
|
|
"/dev/wireview-pro2": "/dev/ttyACM0",
|
|
"/sys/class/tty/ttyACM0": "/sys/devices/fake/usb0",
|
|
"/sys/class/tty/ttyACM1": "/sys/devices/fake/usb1",
|
|
}.get(p, p)
|
|
|
|
def fake_isdir(p):
|
|
return p == "/sys/class/tty"
|
|
|
|
def fake_listdir(p):
|
|
return ["ttyACM0", "ttyACM1"] if p == "/sys/class/tty" else []
|
|
|
|
def fake_isfile(p):
|
|
return p in contents
|
|
|
|
def fake_exists(p):
|
|
return p == "/dev/ttyACM0"
|
|
|
|
def fake_open(p, *a, **k):
|
|
if p in contents:
|
|
return io.StringIO(contents[p])
|
|
return real_open(p, *a, **k)
|
|
|
|
real_open = open
|
|
with mock.patch.object(wv.os.path, "islink", fake_islink), mock.patch.object(
|
|
wv.os.path, "realpath", fake_realpath
|
|
), mock.patch.object(wv.os.path, "isdir", fake_isdir), mock.patch.object(
|
|
wv.os, "listdir", fake_listdir
|
|
), mock.patch.object(wv.os.path, "isfile", fake_isfile), mock.patch.object(
|
|
wv.os.path, "exists", fake_exists
|
|
), mock.patch.object(
|
|
wv.os.path, "dirname", os.path.dirname
|
|
), mock.patch.object(
|
|
builtins, "open", fake_open
|
|
):
|
|
ports = wv.find_wireview_ports()
|
|
check("exactly one port found", ports == ["/dev/ttyACM0"])
|
|
|
|
|
|
def test_find_wireview_ports_none():
|
|
with mock.patch.object(wv.os.path, "islink", lambda p: False), mock.patch.object(
|
|
wv.os.path, "isdir", lambda p: False
|
|
), mock.patch.object(wv.os.path, "exists", lambda p: False):
|
|
check("no ports", wv.find_wireview_ports() == [])
|
|
|
|
|
|
def test_find_hwmon_path():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
# A non-wireview hwmon and a wireview one.
|
|
for name, dev in (("hwmon0", "coretemp"), ("hwmon1", "wireview")):
|
|
d = os.path.join(tmp, name)
|
|
os.makedirs(d)
|
|
with open(os.path.join(d, "name"), "w") as f:
|
|
f.write(dev + "\n")
|
|
|
|
real_join = os.path.join
|
|
real_listdir = os.listdir
|
|
|
|
def fake_join(*parts):
|
|
if parts and parts[0] == "/sys/class/hwmon":
|
|
parts = (tmp,) + parts[1:]
|
|
return real_join(*parts)
|
|
|
|
def fake_listdir(p):
|
|
if p == "/sys/class/hwmon":
|
|
return real_listdir(tmp)
|
|
return real_listdir(p)
|
|
|
|
with mock.patch.object(wv.os.path, "join", fake_join), mock.patch.object(
|
|
wv.os, "listdir", fake_listdir
|
|
):
|
|
found = wv.find_hwmon_path()
|
|
check("hwmon path found", found == os.path.join(tmp, "hwmon1"))
|
|
|
|
with mock.patch.object(wv.os.path, "isdir", lambda p: False):
|
|
check("hwmon absent", wv.find_hwmon_path() is None)
|
|
|
|
|
|
# ── hwmon transport ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_hwmon_device():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
p = os.path.join(tmp, "hwmon2")
|
|
os.makedirs(p)
|
|
with open(os.path.join(p, "name"), "w") as f:
|
|
f.write("wireview\n")
|
|
for i in range(6):
|
|
with open(os.path.join(p, f"in{i}_input"), "w") as f:
|
|
f.write("11904\n")
|
|
with open(os.path.join(p, f"curr{i + 1}_input"), "w") as f:
|
|
f.write("6221\n")
|
|
for name, val in (
|
|
("temp1_input", "39200"),
|
|
("temp2_input", "35200"),
|
|
("temp3_input", "27300"),
|
|
("temp4_input", "39500"),
|
|
("fault_status_raw", "0"),
|
|
("fault_log_raw", "0"),
|
|
("power1_cap", "600000000"),
|
|
("pwm1", "128"),
|
|
):
|
|
with open(os.path.join(p, name), "w") as f:
|
|
f.write(val + "\n")
|
|
|
|
dev = wv.WireViewHwmonDevice(p)
|
|
check("hwmon connect", dev.connect())
|
|
check("hwmon node exists", dev.node_exists())
|
|
s = dev.read_sample()
|
|
check("hwmon sample", s is not None)
|
|
if s:
|
|
check("hwmon temp in", abs(s["temp_in_c"] - 39.2) < 1e-9)
|
|
check("hwmon fan duty ~50%", 49 <= s["fan_duty_pct"] <= 51)
|
|
check("hwmon psu cap", s["psu_capability_w"] == 600)
|
|
check("hwmon 6 pins", len(s["pins"]) == 6)
|
|
check(
|
|
"hwmon total current",
|
|
abs(s["current_total_a"] - round(6 * 6.221, 3)) < 1e-9,
|
|
)
|
|
dev.close()
|
|
check("hwmon closed", not dev.connected)
|
|
|
|
# A missing temp channel must yield null (not NaN): NaN would break
|
|
# the WebSocket JSON (bare NaN token) and 500 the REST endpoint.
|
|
os.remove(os.path.join(p, "temp3_input"))
|
|
dev3 = wv.WireViewHwmonDevice(p)
|
|
dev3.connect()
|
|
s3 = dev3.read_sample()
|
|
check("hwmon missing temp sample", s3 is not None)
|
|
if s3:
|
|
check("hwmon missing temp is None", s3["temp_ext1_c"] is None)
|
|
check("hwmon missing temp present", s3["temp_in_c"] == 39.2)
|
|
import json
|
|
|
|
check(
|
|
"hwmon sample JSON-safe",
|
|
json.dumps(s3, allow_nan=False) is not None,
|
|
)
|
|
dev3.close()
|
|
|
|
# Wrong device name is rejected.
|
|
with open(os.path.join(p, "name"), "w") as f:
|
|
f.write("coretemp\n")
|
|
dev2 = wv.WireViewHwmonDevice(p)
|
|
check("wrong name rejected", not dev2.connect())
|
|
|
|
|
|
def test_hwmon_legacy_attrs():
|
|
"""Older modules expose psu_cap/fan1_input instead of power1_cap/pwm1."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
p = os.path.join(tmp, "hwmon3")
|
|
os.makedirs(p)
|
|
with open(os.path.join(p, "name"), "w") as f:
|
|
f.write("wireview\n")
|
|
for i in range(6):
|
|
with open(os.path.join(p, f"in{i}_input"), "w") as f:
|
|
f.write("12000\n")
|
|
with open(os.path.join(p, f"curr{i + 1}_input"), "w") as f:
|
|
f.write("1000\n")
|
|
for name, val in (
|
|
("temp1_input", "30000"),
|
|
("temp2_input", "30000"),
|
|
("temp3_input", "30000"),
|
|
("temp4_input", "30000"),
|
|
("psu_cap", "2"),
|
|
("fan1_input", "42"),
|
|
("intrusion0_alarm", "0"),
|
|
("intrusion1_alarm", "0"),
|
|
):
|
|
with open(os.path.join(p, name), "w") as f:
|
|
f.write(val + "\n")
|
|
|
|
dev = wv.WireViewHwmonDevice(p)
|
|
check("legacy connect", dev.connect())
|
|
s = dev.read_sample()
|
|
check("legacy sample", s is not None)
|
|
if s:
|
|
check("legacy psu cap 300W", s["psu_capability_w"] == 300)
|
|
check("legacy fan pct", s["fan_duty_pct"] == 42)
|
|
|
|
|
|
# ── Serial transport (pty-based fake device) ──────────────────────────────────
|
|
|
|
|
|
def _build_info_struct(product_name: str, build_info: str) -> bytes:
|
|
"""BuildStruct: VendorData(3) + ProductName(32) + BuildInfo(32) + NameLength(1)."""
|
|
return (
|
|
bytes([0xEF, 0x05, 1])
|
|
+ product_name.encode().ljust(32, b"\x00")
|
|
+ build_info.encode().ljust(32, b"\x00")
|
|
+ bytes([len(product_name)])
|
|
)
|
|
|
|
|
|
def _pty_fake_device(responses: dict):
|
|
"""Create a pty pair whose master side answers protocol commands in a
|
|
background thread. Returns (slave_path, master_fd, thread).
|
|
|
|
The fake never sends the welcome string, so the device exercises its
|
|
documented fallback: identification via the vendor-data reply.
|
|
"""
|
|
import pty
|
|
import threading
|
|
|
|
master, slave = pty.openpty()
|
|
slave_path = os.ttyname(slave)
|
|
|
|
def run():
|
|
while True:
|
|
try:
|
|
cmd = os.read(master, 1)
|
|
except OSError:
|
|
return
|
|
if not cmd:
|
|
return
|
|
resp = responses.get(cmd[0])
|
|
if resp:
|
|
try:
|
|
os.write(master, resp)
|
|
except OSError:
|
|
return
|
|
|
|
t = threading.Thread(target=run, daemon=True)
|
|
t.start()
|
|
return slave_path, master, t
|
|
|
|
|
|
def test_serial_device_protocol():
|
|
"""Full connect handshake + sensor read against a fake device."""
|
|
uid = bytes.fromhex("A7003100015045324B383120")
|
|
responses = {
|
|
wv.CMD_READ_VENDOR_DATA: bytes([0xEF, 0x05, 1]),
|
|
wv.CMD_READ_CONFIG: bytes([0, 0, 1, 0]), # version at offset 2
|
|
wv.CMD_READ_UID: uid,
|
|
wv.CMD_READ_BUILD_INFO: _build_info_struct(
|
|
"WireView Pro II", "TG-WV-PRO2-FW_20251211_1547"
|
|
),
|
|
wv.CMD_READ_SENSOR_VALUES: make_frame(),
|
|
# CMD_SCREEN_CHANGE expects no response.
|
|
}
|
|
slave_path, master, t = _pty_fake_device(responses)
|
|
try:
|
|
dev = wv.WireViewSerialDevice(slave_path)
|
|
check("serial connect", dev.connect())
|
|
check("serial not rejected", not dev.rejected)
|
|
info = dev.info()
|
|
check("serial device name", info["device_name"] == "WireView Pro II")
|
|
check("serial hw_rev", info["hw_rev"] == "EF05")
|
|
check("serial firmware", info["firmware_version"] == "1")
|
|
check("serial uid", info["uid"] == "A7003100015045324B383120")
|
|
check("serial build", info["build"] == "TG-WV-PRO2-FW_20251211_1547")
|
|
check("serial transport", info["transport"] == "serial")
|
|
|
|
s = dev.read_sample()
|
|
check("serial sample", s is not None)
|
|
if s:
|
|
check("serial sample temp", abs(s["temp_in_c"] - 39.2) < 1e-9)
|
|
check("serial sample pins", len(s["pins"]) == 6)
|
|
|
|
dev.close()
|
|
check("serial closed", not dev.connected)
|
|
finally:
|
|
os.close(master)
|
|
t.join(timeout=2)
|
|
|
|
|
|
def test_serial_device_rejected_product():
|
|
"""An unsupported product id is rejected and memoized via .rejected."""
|
|
responses = {wv.CMD_READ_VENDOR_DATA: bytes([0xEF, 0x07, 1])}
|
|
slave_path, master, t = _pty_fake_device(responses)
|
|
try:
|
|
dev = wv.WireViewSerialDevice(slave_path)
|
|
check("serial reject connect", not dev.connect())
|
|
check("serial rejected flag", dev.rejected)
|
|
finally:
|
|
os.close(master)
|
|
t.join(timeout=2)
|
|
|
|
|
|
def test_serial_device_no_response():
|
|
"""A silent port fails the connect (vendor-data read times out)."""
|
|
slave_path, master, t = _pty_fake_device({})
|
|
try:
|
|
dev = wv.WireViewSerialDevice(slave_path)
|
|
check("serial no-response connect", not dev.connect())
|
|
check("serial no-response not rejected", not dev.rejected)
|
|
check("serial no-response sample", dev.read_sample() is None)
|
|
finally:
|
|
os.close(master)
|
|
t.join(timeout=2)
|
|
|
|
|
|
def test_serial_transport_missing_pyserial():
|
|
"""A missing pyserial degrades gracefully: connect fails, reads are
|
|
None, and the warning is logged once — not on every attempt."""
|
|
import logging
|
|
|
|
records: list[str] = []
|
|
|
|
class Capture(logging.Handler):
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
records.append(record.getMessage())
|
|
|
|
logger = logging.getLogger("nvcurve.wireview")
|
|
handler = Capture()
|
|
old_level = logger.level
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.WARNING)
|
|
try:
|
|
with mock.patch.object(wv, "serial", None):
|
|
wv._serial_missing_warned = False
|
|
dev = wv.WireViewSerialDevice("/dev/ttyACM99")
|
|
check("no-pyserial connect", not dev.connect())
|
|
check("no-pyserial not rejected", not dev.rejected)
|
|
check("no-pyserial sample", dev.read_sample() is None)
|
|
# A second attempt must not re-warn.
|
|
dev2 = wv.WireViewSerialDevice("/dev/ttyACM99")
|
|
check("no-pyserial second attempt", not dev2.connect())
|
|
warnings = [r for r in records if "pyserial" in r]
|
|
check("no-pyserial warns once", len(warnings) == 1)
|
|
finally:
|
|
logger.removeHandler(handler)
|
|
logger.setLevel(old_level)
|
|
wv._serial_missing_warned = False
|
|
|
|
|
|
def main() -> int:
|
|
print("wireview tests:")
|
|
test_parse_sensor_struct()
|
|
test_parse_psu_capabilities()
|
|
test_parse_negative_temps()
|
|
test_corruption_check()
|
|
test_decode_faults()
|
|
test_product_support()
|
|
test_find_wireview_ports()
|
|
test_find_wireview_ports_none()
|
|
test_find_hwmon_path()
|
|
test_hwmon_device()
|
|
test_hwmon_legacy_attrs()
|
|
test_serial_device_protocol()
|
|
test_serial_device_rejected_product()
|
|
test_serial_device_no_response()
|
|
test_serial_transport_missing_pyserial()
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
return 1 if FAIL else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|