slmm/test_wedge_recovery.py

#!/usr/bin/env python3
"""
End-to-end test for wedge detection and recovery (app/recovery.py).

Spins up a fake NL-43 (control + FTP listeners speaking just enough of the
ASCII protocol) that can be "wedged" — control listener killed while FTP
stays up, exactly the failure signature from the Feb 2026 investigation.
A local HTTP server stands in for the relay webhook; hitting it "power
cycles" the fake device.

Runs against a throwaway SQLite DB in a temp directory — does not touch
data/slmm.db.

Usage:
    python3 test_wedge_recovery.py

Exit code 0 = all tests passed.
"""

import asyncio
import os
import sys
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

# --- Run from a temp dir so app.database creates a throwaway DB ------------
REPO_ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(REPO_ROOT))
WORKDIR = tempfile.mkdtemp(prefix="slmm-wedge-test-")
os.chdir(WORKDIR)

# Confirm fast in tests
os.environ.setdefault("WEDGE_CONFIRM_FAILURES", "2")

from app.database import Base, engine, SessionLocal  # noqa: E402
from app.models import NL43Config, NL43Status  # noqa: E402
from app.recovery import classify_failure, probe_tcp, recovery_manager  # noqa: E402

Base.metadata.create_all(bind=engine)

PASS = 0
FAIL = 0


def check(name: str, condition: bool, detail: str = ""):
    global PASS, FAIL
    if condition:
        PASS += 1
        print(f"  ✓ {name}")
    else:
        FAIL += 1
        print(f"  ✗ {name} {('— ' + detail) if detail else ''}")


# ---------------------------------------------------------------------------
# Fake NL-43 device
# ---------------------------------------------------------------------------

class FakeNL43:
    """Minimal NL-43: control port speaks the ASCII protocol, FTP port
    just accepts. Can be wedged (control listener killed, FTP alive)."""

    def __init__(self):
        self.control_server = None
        self.ftp_server = None
        self.control_port = None
        self.ftp_port = None
        self.measurement_state = "Stop"
        self.index = 1
        self.commands = []  # every command line received

    async def start(self):
        self.ftp_server = await asyncio.start_server(self._handle_ftp, "127.0.0.1", 0)
        self.ftp_port = self.ftp_server.sockets[0].getsockname()[1]
        await self.start_control()

    async def start_control(self):
        self.control_server = await asyncio.start_server(
            self._handle_control, "127.0.0.1", 0 if self.control_port is None else self.control_port
        )
        self.control_port = self.control_server.sockets[0].getsockname()[1]

    async def wedge(self):
        """Kill the control listener (existing + new connections die). FTP stays up."""
        if self.control_server:
            self.control_server.close()
            await self.control_server.wait_closed()
            self.control_server = None

    async def power_cycle(self, boot_delay: float = 2.0):
        """Simulate a power cycle: everything down, then back up after boot_delay.
        Power loss stops any running measurement."""
        await self.wedge()
        self.measurement_state = "Stop"
        await asyncio.sleep(boot_delay)
        await self.start_control()

    async def stop(self):
        for srv in (self.control_server, self.ftp_server):
            if srv:
                srv.close()
                await srv.wait_closed()
        self.control_server = None
        self.ftp_server = None

    async def _handle_ftp(self, reader, writer):
        try:
            writer.write(b"220 Connection Ready\r\n")
            await writer.drain()
            await reader.read(1024)
        except Exception:
            pass
        finally:
            writer.close()

    async def _handle_control(self, reader, writer):
        try:
            while True:
                line = await reader.readuntil(b"\n")
                cmd = line.decode(errors="ignore").strip()
                if not cmd:
                    continue
                self.commands.append(cmd)
                writer.write(self._respond(cmd))
                await writer.drain()
        except (asyncio.IncompleteReadError, ConnectionResetError):
            pass
        finally:
            try:
                writer.close()
            except Exception:
                pass

    def _respond(self, cmd: str) -> bytes:
        ok = b"R+0000\r\n"
        if cmd == "Measure?":
            return ok + f"{self.measurement_state}\r\n".encode()
        if cmd == "Measure,Start":
            self.measurement_state = "Start"
            return ok
        if cmd == "Measure,Stop":
            self.measurement_state = "Stop"
            return ok
        if cmd.startswith("Clock,"):
            return ok
        if cmd == "Store Name?":
            return ok + f"{self.index:04d}\r\n".encode()
        if cmd.startswith("Store Name,"):
            self.index = int(cmd.split(",")[1])
            return ok
        if cmd == "Overwrite?":
            return ok + b"None\r\n"
        if cmd == "DOD?":
            return ok + b"1,55.5,54.2,60.1,50.3,72.8\r\n"
        if cmd.startswith("Sleep Mode"):
            return ok + (b"Off\r\n" if cmd.endswith("?") else b"")
        return b"R+0001\r\n"  # command error


# ---------------------------------------------------------------------------
# Fake relay webhook
# ---------------------------------------------------------------------------

class FakeRelay:
    """HTTP server standing in for a relay controller. A request to /cycle
    power-cycles the fake device."""

    def __init__(self, device: FakeNL43, loop: asyncio.AbstractEventLoop):
        self.device = device
        self.loop = loop
        self.hits = 0
        relay = self

        class Handler(BaseHTTPRequestHandler):
            def _handle(self):
                relay.hits += 1
                # Schedule the device power cycle on the asyncio loop
                asyncio.run_coroutine_threadsafe(relay.device.power_cycle(boot_delay=2.0), relay.loop)
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"cycling")

            do_GET = _handle
            do_POST = _handle

            def log_message(self, *args):
                pass

        self.server = HTTPServer(("127.0.0.1", 0), Handler)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    @property
    def url(self):
        return f"http://127.0.0.1:{self.port}/cycle"

    def stop(self):
        self.server.shutdown()


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------

async def test_classifier(device: FakeNL43):
    print("\n[1] Failure classification")

    # Healthy device
    c = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
    check("healthy device classifies as ok", c["state"] == "ok", str(c))

    # Wedged: control refused, FTP alive — the definitive signature
    await device.wedge()
    c = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
    check("wedged device classifies as wedged", c["state"] == "wedged", str(c))
    check("wedge confidence is high (FTP alive)", c["confidence"] == "high", str(c))

    # Wedged without FTP probe available
    c = await classify_failure("127.0.0.1", device.control_port, None)
    check("wedge without FTP probe is medium confidence",
          c["state"] == "wedged" and c["confidence"] == "medium", str(c))

    # Offline: unroutable address times out (TEST-NET-1)
    c = await classify_failure("192.0.2.1", 2255, None)
    check("unreachable host classifies as offline", c["state"] == "offline", str(c))

    await device.start_control()
    probe = await probe_tcp("127.0.0.1", device.control_port)
    check("control port back open after un-wedge", probe == "open", probe)


async def test_webhook_recovery(device: FakeNL43, relay: FakeRelay):
    print("\n[2] End-to-end webhook recovery (wedge → relay cycle → reconnect → resume)")

    db = SessionLocal()
    cfg = NL43Config(
        unit_id="TEST-NL43",
        host="127.0.0.1",
        tcp_port=device.control_port,
        ftp_port=device.ftp_port,
        tcp_enabled=True,
        ftp_enabled=True,
        auto_recovery_enabled=True,
        reset_backend="webhook",
        reset_webhook_url=relay.url,
        reset_webhook_method="POST",
        recovery_boot_wait_seconds=10,   # validator min; device boots in 2s
        recovery_reconnect_timeout=60,
        recovery_auto_resume=True,
        recovery_max_attempts=3,
        recovery_window_minutes=360,
    )
    db.add(cfg)
    # Device was measuring before the wedge
    status = NL43Status(unit_id="TEST-NL43", measurement_state="Start", consecutive_failures=2)
    db.add(status)
    db.commit()

    # Wedge it
    device.measurement_state = "Start"
    await device.wedge()

    classification = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
    check("pre-recovery classification is wedged", classification["state"] == "wedged")

    started = recovery_manager.start_recovery(cfg, classification)
    check("recovery task started", started)

    # Wait for the recovery task to finish (boot wait 10s + commands ≈ 20s)
    deadline = time.time() + 90
    while recovery_manager.is_active("TEST-NL43") and time.time() < deadline:
        await asyncio.sleep(1)
    check("recovery task completed", not recovery_manager.is_active("TEST-NL43"))

    check("relay webhook was hit", relay.hits == 1, f"hits={relay.hits}")

    db.expire_all()
    status = db.query(NL43Status).filter_by(unit_id="TEST-NL43").first()
    check("recovery_state is idle", status.recovery_state == "idle", status.recovery_state)
    check("connection_state is ok", status.connection_state == "ok", status.connection_state)
    check("consecutive_failures reset", status.consecutive_failures == 0)
    check("last_recovery_result records success",
          status.last_recovery_result and "Recovered" in status.last_recovery_result,
          str(status.last_recovery_result))

    check("measurement resumed on device", device.measurement_state == "Start", device.measurement_state)
    check("device received Measure,Start", "Measure,Start" in device.commands)
    check("start cycle used overwrite protection", "Overwrite?" in device.commands)

    db.close()


async def test_gating(device: FakeNL43):
    print("\n[3] Recovery gating (confirmation threshold + attempt limit)")

    db = SessionLocal()
    cfg = NL43Config(
        unit_id="TEST-GATE",
        host="127.0.0.1",
        tcp_port=device.control_port,
        ftp_port=device.ftp_port,
        tcp_enabled=True,
        auto_recovery_enabled=True,
        reset_backend="webhook",
        reset_webhook_url="http://127.0.0.1:1/unreachable",  # fails fast
        recovery_max_attempts=1,
        recovery_window_minutes=360,
    )
    db.add(cfg)
    status = NL43Status(unit_id="TEST-GATE", consecutive_failures=1)
    db.add(status)
    db.commit()

    classification = {"state": "wedged", "confidence": "high", "tcp_probe": "refused", "ftp_probe": "open"}

    # Below confirmation threshold (1 < 2) — no recovery
    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
    check("below threshold does not trigger", not started)

    # At threshold — triggers (and fails fast on the dead webhook)
    status.consecutive_failures = 2
    db.commit()
    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
    check("at threshold triggers recovery", started)

    deadline = time.time() + 30
    while recovery_manager.is_active("TEST-GATE") and time.time() < deadline:
        await asyncio.sleep(0.5)

    db.expire_all()
    status = db.query(NL43Status).filter_by(unit_id="TEST-GATE").first()
    check("failed webhook marks recovery failed", status.recovery_state == "failed", status.recovery_state)
    check("failure recorded in result",
          status.last_recovery_result and "FAILED" in status.last_recovery_result,
          str(status.last_recovery_result))

    # Attempt limit (max 1 per window) — second attempt blocked
    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
    check("attempt limit blocks repeat recovery", not started)

    # Disabled flag blocks everything
    cfg.auto_recovery_enabled = False
    db.commit()
    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
    check("auto_recovery_enabled=False blocks recovery", not started)

    db.close()


async def main():
    print(f"Wedge recovery test — temp workdir: {WORKDIR}")

    device = FakeNL43()
    await device.start()
    relay = FakeRelay(device, asyncio.get_running_loop())
    print(f"Fake NL-43: control=127.0.0.1:{device.control_port}, ftp=127.0.0.1:{device.ftp_port}")
    print(f"Fake relay webhook: {relay.url}")

    try:
        await test_classifier(device)
        await test_webhook_recovery(device, relay)
        await test_gating(device)
    finally:
        relay.stop()
        await device.stop()

    print(f"\n{'='*50}\nResults: {PASS} passed, {FAIL} failed")
    return FAIL == 0


if __name__ == "__main__":
    ok = asyncio.run(main())
    sys.exit(0 if ok else 1)