#!/usr/bin/env python3
"""
test_v0.2_all.py  --  SDG2042X v0.2 regression test suite
Tests: Issue #1, #2, #6, #12  (milestone: v0.2.0 - first bugfix round finished)

Usage:
    python3 test_v0.2_all.py              # run all tests
    python3 test_v0.2_all.py --issue 12   # run single issue

Place this file next to SDG2042X_V0.2.py and run from that directory.
No instrument required. All tests are offline / mocked.

Author: ToGo-Lab test suite — generated 2026-04-18
"""

import ast
import importlib.util
import os
import re
import socket
import sys
import tempfile
import threading
import time
import types

SCRIPT_NAME = "SDG2042X_V0.2.py"

# ─────────────────────────────────────────────────────────────────────────────
# ANSI colours (disabled on Windows)
# ─────────────────────────────────────────────────────────────────────────────
USE_COLOUR = sys.platform != "win32"
def _c(code, s): return f"\033[{code}m{s}\033[0m" if USE_COLOUR else s
OK   = lambda s: _c("32;1", s)
FAIL = lambda s: _c("31;1", s)
HEAD = lambda s: _c("36;1", s)
DIM  = lambda s: _c("2",    s)

# ─────────────────────────────────────────────────────────────────────────────
# Result collector
# ─────────────────────────────────────────────────────────────────────────────
results: dict[str, list[tuple[bool, str]]] = {}

def record(issue: str, ok: bool, label: str, detail: str = ""):
    results.setdefault(issue, []).append((ok, label))
    mark = OK("  OK  ") if ok else FAIL(" FAIL ")
    print(f"  [{mark}] {label}")
    if detail:
        for line in detail.strip().splitlines():
            print(DIM(f"         {line}"))

def section(title: str):
    print()
    print(HEAD(f"── {title} ──"))

# ─────────────────────────────────────────────────────────────────────────────
# Locate the script under test
# ─────────────────────────────────────────────────────────────────────────────
def find_script() -> str:
    candidate = os.path.join(os.path.dirname(__file__), SCRIPT_NAME)
    if not os.path.exists(candidate):
        sys.exit(FAIL(f"ERROR: {SCRIPT_NAME} not found next to this test file.\n"
                      f"Expected: {candidate}"))
    return candidate

SCRIPT_PATH = find_script()
SRC = open(SCRIPT_PATH, encoding="utf-8").read()

# ─────────────────────────────────────────────────────────────────────────────
# Minimal PyQt5 mock so the module can be imported without a display
# ─────────────────────────────────────────────────────────────────────────────
def _install_pyqt5_mock():
    def _noop(*a, **k): return None
    def _cls(name): return type(name, (), {
        "__init__": _noop, "__call__": _noop,
        "setEnabled": _noop, "addItems": _noop, "setText": _noop,
        "text": lambda s: "", "currentText": lambda s: "",
        "appendPlainText": _noop, "setReadOnly": _noop,
        "setMinimumHeight": _noop, "setPlaceholderText": _noop,
        "setWindowTitle": _noop, "resize": _noop, "show": _noop,
        "setRange": _noop, "setValue": _noop, "value": lambda s: 1,
        "addWidget": _noop, "setLayout": _noop, "addStretch": _noop,
        "setColumnStretch": _noop, "setHorizontalHeaderLabels": _noop,
        "horizontalHeader": lambda s: type("h",(),{"setStretchLastSection":_noop})(),
        "verticalHeader": lambda s: type("v",(),{"setVisible":_noop})(),
        "setItem": _noop, "item": lambda s,r,c: type("i",(),{"setFlags":_noop,"setText":_noop,"text":lambda ss:""})(),
        "setFlags": _noop, "clicked": type("sig",(),{"connect":_noop})(),
        "currentTextChanged": type("sig",(),{"connect":_noop})(),
        "returnPressed": type("sig",(),{"connect":_noop})(),
        "isRunning": lambda s: False, "start": _noop,
        "done": type("sig",(),{"connect":_noop, "emit":_noop})(),
        "isNull": lambda s: True,
    })

    # Build mock modules
    for mod in ["PyQt5","PyQt5.QtWidgets","PyQt5.QtCore","PyQt5.QtGui"]:
        if mod not in sys.modules:
            sys.modules[mod] = types.ModuleType(mod)

    qw = sys.modules["PyQt5.QtWidgets"]
    qc = sys.modules["PyQt5.QtCore"]
    qg = sys.modules["PyQt5.QtGui"]

    for name in ["QWidget","QLabel","QLineEdit","QComboBox","QPushButton",
                 "QPlainTextEdit","QTabWidget","QGridLayout","QHBoxLayout",
                 "QVBoxLayout","QApplication","QSpinBox","QCheckBox",
                 "QListWidget","QTableWidget","QTableWidgetItem",
                 "QFileDialog","QSizePolicy"]:
        setattr(qw, name, _cls(name))

    # QThread: base for workers
    setattr(qc, "QThread", _cls("QThread"))
    setattr(qc, "Qt", type("Qt", (), {"WaitCursor": 0, "ItemIsEnabled": 0}))

    class _Signal:
        def __init__(self, *types): pass
        def connect(self, fn): pass
        def emit(self, *a): pass
    setattr(qc, "pyqtSignal", _Signal)

    setattr(qg, "QImage",  _cls("QImage"))
    setattr(qg, "QPixmap", _cls("QPixmap"))

_install_pyqt5_mock()

# ─────────────────────────────────────────────────────────────────────────────
# Load module under test (SDG2042X_V0.2.py)
# ─────────────────────────────────────────────────────────────────────────────
spec = importlib.util.spec_from_file_location("sdg", SCRIPT_PATH)
sdg  = importlib.util.module_from_spec(spec)
try:
    spec.loader.exec_module(sdg)
except Exception as e:
    sys.exit(FAIL(f"ERROR: failed to import {SCRIPT_NAME}: {e}"))


# ═════════════════════════════════════════════════════════════════════════════
# ISSUE #12 — human_to_eng() MHz regression
# ═════════════════════════════════════════════════════════════════════════════
def test_issue_12():
    section("Issue #12 — human_to_eng() MHz regression")

    fn = sdg.human_to_eng

    cases = [
        # (input, expected, description)
        # ── THE BUG: these were silently wrong before v0.2 ──
        ("1.5 MHz",   "1500000",      "BUG-REG: 1.5 MHz → 1500000 (was 0.0000015)"),
        ("30 MHz",    "30000000",     "BUG-REG: 30 MHz → 30000000"),
        ("100 MHz",   "100000000",    "BUG-REG: 100 MHz → 100000000"),
        # ── Frequency ──
        ("1 kHz",     "1000",         "1 kHz"),
        ("10 kHz",    "10000",        "10 kHz"),
        ("100 Hz",    "100",          "100 Hz"),
        ("1 GHz",     "1000000000",   "1 GHz"),
        ("1 hz",      "1",            "lowercase hz"),
        ("1 KHZ",     "1000",         "uppercase KHZ"),
        # ── Time ──
        ("500 ms",    "0.5",          "500 ms"),
        ("1 us",      "1e-06",        "1 us"),
        ("250 ns",    repr(250e-9),   "250 ns"),
        # ── Voltage ──
        ("2.5 V",     "2.5",          "2.5 V"),
        ("500 mV",    "0.5",          "500 mV"),
        ("1.0 Vpp",   "1",            "1.0 Vpp"),
        # ── Bare numbers (duty, phase, offset) ──
        ("50",        "50",           "bare integer"),
        ("0.001",     "0.001",        "bare float"),
        ("1e-6",      "1e-06",        "scientific notation"),
        # ── Fallback ──
        ("SINE",      "SINE",         "non-numeric fallback"),
        ("",          "",             "empty string"),
    ]

    def _eq(a, b):
        """String equality first; fall back to float comparison for IEEE 754 noise."""
        if a == b:
            return True
        try:
            return abs(float(a) - float(b)) <= 1e-12 * max(abs(float(b)), 1e-30)
        except (ValueError, TypeError):
            return False

    for inp, expected, desc in cases:
        result = fn(inp)
        ok = _eq(result, expected)
        record("12", ok, desc,
               f"input={inp!r}  expected≈{expected!r}  got={result!r}" if not ok else "")


# ═════════════════════════════════════════════════════════════════════════════
# ISSUE #1 — _recv_line() partial TCP receive
# ═════════════════════════════════════════════════════════════════════════════
def _fragmented_server(host, port, response: bytes, chunk_sizes: list, ready: threading.Event):
    """Mini TCP server: sends response in deliberate fragments."""
    srv = socket.create_server((host, port))
    srv.settimeout(5)
    ready.set()
    try:
        conn, _ = srv.accept()
        offset = 0
        for size in chunk_sizes:
            conn.sendall(response[offset:offset + size])
            offset += size
            time.sleep(0.012)
        if offset < len(response):
            conn.sendall(response[offset:])
        conn.close()
    finally:
        srv.close()

def _next_port(base=15025):
    """Find a free port starting from base."""
    for p in range(base, base + 100):
        try:
            s = socket.create_server(("127.0.0.1", p)); s.close(); return p
        except OSError:
            continue
    raise RuntimeError("No free port found")

def test_issue_1():
    section("Issue #1 — _recv_line() handles partial TCP receive")

    def run_tc(label, response: bytes, chunk_sizes: list):
        port  = _next_port()
        ready = threading.Event()
        t = threading.Thread(
            target=_fragmented_server,
            args=("127.0.0.1", port, response, chunk_sizes, ready),
            daemon=True)
        t.start()
        ready.wait(timeout=2)

        io = sdg.SDGLan()
        try:
            io.connect("127.0.0.1", port)
            result = io._recv_line()
        finally:
            io.close()
        t.join(timeout=3)

        expected = response.decode("ascii").strip()
        ok = result.strip() == expected
        record("1", ok, label,
               f"expected={expected!r}  got={result.strip()!r}" if not ok else "")

    # TC-1a: single chunk (baseline)
    run_tc("TC-1a single chunk",
           b"C1:BSWV WVTP,SINE,FRQ,1000\n", [])

    # TC-1b: \n arrives in a separate recv() — the original bug
    msg = b"C1:BSWV WVTP,SINE,FRQ,1000\n"
    run_tc("TC-1b newline split into own chunk",
           msg, [len(msg) - 1])

    # TC-1c: byte-by-byte fragmentation
    msg = b"*IDN? Siglent,SDG2042X\n"
    run_tc("TC-1c byte-by-byte",
           msg, [1] * (len(msg) - 1))

    # TC-1d: two unequal halves
    msg = b"C1:OUTP ON\n"
    run_tc("TC-1d split at midpoint",
           msg, [len(msg) // 2])

    # TC-1e: empty response (server closes immediately — must not hang)
    port  = _next_port()
    ready = threading.Event()
    def empty_server():
        srv = socket.create_server(("127.0.0.1", port)); srv.settimeout(5)
        ready.set()
        conn, _ = srv.accept(); conn.close(); srv.close()
    t = threading.Thread(target=empty_server, daemon=True); t.start()
    ready.wait(timeout=2)
    io = sdg.SDGLan()
    try:
        io.connect("127.0.0.1", port)
        result = io._recv_line()
    finally:
        io.close()
    t.join(timeout=3)
    record("1", result == "", "TC-1e empty response returns empty string",
           f"got={result!r}" if result != "" else "")


# ═════════════════════════════════════════════════════════════════════════════
# ISSUE #6 — Screenshot path uses stale disk config
# ═════════════════════════════════════════════════════════════════════════════
def test_issue_6():
    section("Issue #6 — on_screenshot() uses in-memory cfg, not disk config")

    with tempfile.TemporaryDirectory() as tmp:
        dir_disk   = os.path.join(tmp, "dir_disk");    os.makedirs(dir_disk)
        dir_memory = os.path.join(tmp, "dir_memory");  os.makedirs(dir_memory)

        # Write a config file pointing at dir_disk
        cfg_path = os.path.join(tmp, "SDG2042x.config")
        with open(cfg_path, "w") as f:
            f.write(f"SCREENSHOT_DIR={dir_disk}\nPRESET_DIR={tmp}\n")

        captured = {}

        # Patch load_config to return disk dir (simulates on-disk state)
        _orig_load = sdg.load_config
        sdg.load_config = lambda p: {"PRESET_DIR": tmp, "SCREENSHOT_DIR": dir_disk}

        # Patch ScreenshotWorker: capture outdir, don't actually run
        class FakeWorker:
            def __init__(self, io, outdir, parent=None):
                captured["outdir"] = outdir
            def isRunning(self): return False
            def start(self): pass
            done = type("sig", (), {"connect": lambda s, f: None})()
        sdg.ScreenshotWorker = FakeWorker

        # Minimal stand-in for Main instance with in-memory cfg = dir_memory
        class FakeMain:
            cfg     = {"SCREENSHOT_DIR": dir_memory, "PRESET_DIR": tmp}
            i       = type("io", (), {"sock": object()})()
            _scr_worker = None
            log     = type("log", (), {"appendPlainText": lambda s, t: None})()
            def logln(self, s): pass
            def setCursor(self, c): pass
            def unsetCursor(self): pass
            def _on_screenshot_done(self, *a): pass
            btn_scr = type("btn", (), {
                "setEnabled": lambda s, v: None,
                "isRunning":  lambda s: False})()

        obj = FakeMain()
        sdg.Main.on_screenshot(obj)

        sdg.load_config = _orig_load

        used = captured.get("outdir", "NOT_CAPTURED")

        # TC-6a: worker must receive the in-memory dir, not the disk dir
        record("6",
               used == dir_memory,
               "TC-6a: worker receives self.cfg path, not load_config() path",
               f"disk cfg  = {dir_disk}\n"
               f"self.cfg  = {dir_memory}\n"
               f"received  = {used}")

        # TC-6b: inverse sanity — must NOT be the disk path
        record("6",
               used != dir_disk,
               "TC-6b: worker path differs from stale disk path",
               f"got {used!r}, must not be {dir_disk!r}" if used == dir_disk else "")


# ═════════════════════════════════════════════════════════════════════════════
# ISSUE #2 — ARB download blocks GUI thread (structural AST check)
# ═════════════════════════════════════════════════════════════════════════════
def test_issue_2():
    section("Issue #2 — ARB download moved to QThread (structural)")

    tree = ast.parse(SRC)
    fn_names = {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)}

    # TC-2a: ARBDownloadWorker class exists
    worker_nodes = [n for n in ast.walk(tree)
                    if isinstance(n, ast.ClassDef) and n.name == "ARBDownloadWorker"]
    record("2", bool(worker_nodes), "TC-2a: class ARBDownloadWorker exists")

    if worker_nodes:
        worker = worker_nodes[0]
        bases  = [getattr(b, "attr", getattr(b, "id", "")) for b in worker.bases]
        # TC-2b: inherits QThread
        record("2", "QThread" in bases,
               "TC-2b: ARBDownloadWorker inherits QThread",
               f"bases found: {bases}" if "QThread" not in bases else "")

        # TC-2c: has a run() method
        run_methods = [n for n in ast.walk(worker)
                       if isinstance(n, ast.FunctionDef) and n.name == "run"]
        record("2", bool(run_methods), "TC-2c: ARBDownloadWorker.run() defined")

        if run_methods:
            run_src = ast.get_source_segment(SRC, run_methods[0]) or ""
            # TC-2d: recv is inside the worker's run(), i.e. the work is off-thread
            record("2", "recv" in run_src,
                   "TC-2d: recv() call inside ARBDownloadWorker.run()",
                   "No recv() found in run() body" if "recv" not in run_src else "")

    # TC-2e: on_arb_download in Main has NO while loop (blocking recv removed)
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == "on_arb_download":
            has_while = any(isinstance(n, ast.While) for n in ast.walk(node))
            record("2", not has_while,
                   "TC-2e: on_arb_download() contains no blocking while loop",
                   "while loop still present — BUG NOT FIXED" if has_while else "")

            # TC-2f: calls .start() to launch the thread
            call_attrs = [getattr(n.func, "attr", "")
                          for n in ast.walk(node) if isinstance(n, ast.Call)]
            record("2", "start" in call_attrs,
                   "TC-2f: on_arb_download() calls worker.start()",
                   f"call attrs found: {call_attrs}" if "start" not in call_attrs else "")

    # TC-2g: result handler exists
    record("2", "_on_arb_download_done" in fn_names,
           "TC-2g: _on_arb_download_done() handler defined")

    # TC-2h: done signal connected to handler
    record("2", "_on_arb_download_done" in SRC and "done.connect" in SRC,
           "TC-2h: done signal connected to handler")

    print()
    print(DIM("  NOTE: live GUI responsiveness during download requires"))
    print(DIM("  a connected SDG2042X — run manually per test strategy."))


# ═════════════════════════════════════════════════════════════════════════════
# Summary
# ═════════════════════════════════════════════════════════════════════════════
def print_summary():
    print()
    print(HEAD("══════════════════════════════════════════════"))
    print(HEAD("  SUMMARY"))
    print(HEAD("══════════════════════════════════════════════"))

    total_ok = total_fail = 0
    issue_verdict = {}

    for issue, tcs in sorted(results.items()):
        ok_count   = sum(1 for ok, _ in tcs if ok)
        fail_count = sum(1 for ok, _ in tcs if not ok)
        total_ok   += ok_count
        total_fail += fail_count
        verdict = "PASS" if fail_count == 0 else "FAIL"
        issue_verdict[issue] = verdict
        v_str = OK(f"PASS ({ok_count}/{len(tcs)})") if verdict == "PASS" \
                else FAIL(f"FAIL ({fail_count} failed)")
        print(f"  Issue #{issue:2s}  {v_str}")

    print()
    overall = total_fail == 0
    total   = total_ok + total_fail
    print(f"  Total: {total_ok}/{total} passed")
    if overall:
        print(OK("  ✓ All tests passed — safe to close all four issues"))
    else:
        print(FAIL(f"  ✗ {total_fail} test(s) failed — do NOT close affected issues"))

    print()
    print("  Gitea close checklist:")
    for iss in ["1", "2", "6", "12"]:
        v = issue_verdict.get(iss, "NOT RUN")
        box = OK("✓ close") if v == "PASS" else FAIL("✗ keep open")
        print(f"    #{iss:2s}  {box}")
    print()

    return overall


# ═════════════════════════════════════════════════════════════════════════════
# Entry point
# ═════════════════════════════════════════════════════════════════════════════
if __name__ == "__main__":
    target = None
    if "--issue" in sys.argv:
        idx = sys.argv.index("--issue")
        if idx + 1 < len(sys.argv):
            target = sys.argv[idx + 1]

    print(HEAD(f"SDG2042X v0.2 — regression test suite"))
    print(DIM(f"Script under test: {SCRIPT_PATH}"))

    if target in (None, "12"): test_issue_12()
    if target in (None, "1"):  test_issue_1()
    if target in (None, "6"):  test_issue_6()
    if target in (None, "2"):  test_issue_2()

    ok = print_summary()
    sys.exit(0 if ok else 1)
