Pakobbix/nvcurve
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b69b7502b370204a64b31bfbd989d417a51f208c
nvcurve/tests/test_wireview.py
T
ARIA b69b7502b3 feat: wireview pin-imbalance warning, per-pin rating bars, pin_imbalance_a 
- Per-pin bar now scales against the 9.2 A per-pin rating (10 A full
  scale, red at/over rating) instead of the 55 A total connector OCP
- Debounced app-wide warning banner (visible on all tabs) when the
  current between two pins diverges by > 2 A for ~7 s - the signature
  of a bad pin contact that can overheat and melt the 12VHPWR
  connector; 5 s grace period after clearing
- New pin_imbalance_a sample field (max - min pin current) in both the
  serial and hwmon sample builders, plus frontend type
- Docs: README + WireView.md cover the warning and threshold; fresh
  screenshot with the new per-pin labels
- Tests: vitest setup + 9 unit tests for the debounce state machine;
  3 new backend checks for pin_imbalance_a
2026-10-03 22:13:13 +02:00

542 lines
19 KiB
Python
Raw Blame History

"""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(
"pin imbalance (uniform)",
s["pin_imbalance_a"] == 0.0,
)
# Imbalance is the spread between the highest- and lowest-loaded pin.
uneven = wv.parse_sensor_struct(
make_frame(pins=((11904, 6221, 74055),) * 5 + ((11900, 1100, 13090),))
)
check(
"pin imbalance (uneven)",
abs(uneven["pin_imbalance_a"] - 5.121) < 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,
)
check("hwmon pin imbalance (uniform)", s["pin_imbalance_a"] == 0.0)
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())
Reference in new issue View Git Blame Copy Permalink