"""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())