#!/usr/bin/env python3
"""Host tests for the Mac-side Wi-Fi provisioning helper.

Covers value encoding round-trips, command framing, response parsing, redaction,
the candidate-first verdict wording and the CLI's password handling. Runs
without a serial port or a terminal.
"""
from __future__ import annotations

import io
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "tools/mac"))

import wifi_ctl  # noqa: E402


class FakePort:
    """A serial-like object that replies with a canned framed response."""

    def __init__(self, response: str):
        self._chunks = [line + "\n" for line in response.splitlines()]
        self.written = b""

    def reset_input_buffer(self):
        pass

    def write(self, data: bytes):
        self.written += data

    def flush(self):
        pass

    def readline(self):
        if not self._chunks:
            return b""
        return self._chunks.pop(0).encode()


# ---- Value encoding --------------------------------------------------------

def test_encode_matches_firmware_rules():
    # Unreserved characters pass through unchanged.
    assert wifi_ctl.usb_encode("Home-WiFi_1.2~3") == "Home-WiFi_1.2~3"
    # Space is always %20; '+' is always %2B.
    assert wifi_ctl.usb_encode("a b") == "a%20b"
    assert wifi_ctl.usb_encode("a+b") == "a%2Bb"
    # Separators and the escape character are always encoded.
    assert wifi_ctl.usb_encode("a=b") == "a%3Db"
    assert wifi_ctl.usb_encode("a&b") == "a%26b"
    assert wifi_ctl.usb_encode("a%b") == "a%25b"
    # Uppercase hex.
    assert wifi_ctl.usb_encode("é") == "%C3%A9"


def test_encode_decode_round_trip():
    values = [
        "", "Home-WiFi", "My Home WiFi", "hello world", "a=b%c&d", "Powell Home",
        "p@ss#w/rd?", "+plus+", "quote\"and'apos", "tab\tchar", "中文网络",
        "café-☕", "100% pure", "a&b=c", "=", "&", "%%%",
    ]
    for value in values:
        assert wifi_ctl.usb_decode(wifi_ctl.usb_encode(value)) == value, value
        encoded = wifi_ctl.usb_encode(value)
        assert " " not in encoded and "&" not in encoded and "=" not in encoded


def test_decode_preserves_invalid_escape():
    assert wifi_ctl.usb_decode("a%ZZb") == "a%ZZb"
    assert wifi_ctl.usb_decode("trailing%") == "trailing%"


# ---- Command framing -------------------------------------------------------

def test_build_command_add_encodes_values():
    line = wifi_ctl.build_command("wifi.add", ssid="Powell Home", password="hello world")
    assert line == "WFP/1 wifi.add ssid=Powell%20Home password=hello%20world\n"
    # A value containing separators cannot split the token.
    line = wifi_ctl.build_command("wifi.add", ssid="a=b&c", password="x%y")
    assert line == "WFP/1 wifi.add ssid=a%3Db%26c password=x%25y\n"


def test_build_command_simple_verbs():
    assert wifi_ctl.build_command("wifi.list") == "WFP/1 wifi.list\n"
    assert wifi_ctl.build_command("wifi.status") == "WFP/1 wifi.status\n"
    assert wifi_ctl.build_command("wifi.remove", ssid="My Office") == \
        "WFP/1 wifi.remove ssid=My%20Office\n"


# ---- Response parsing ------------------------------------------------------

def test_parse_status():
    result = wifi_ctl.parse_response([
        "WFP/1 OK wifi.status state=CONNECTED ssid=Home-WiFi profiles=3\n",
        "WFP/1 END\n",
    ])
    assert result["ok"] and result["verb"] == "wifi.status" and result["end"]
    assert result["fields"]["state"] == "CONNECTED"
    assert result["fields"]["ssid"] == "Home-WiFi"


def test_parse_status_decodes_utf8_ssid():
    result = wifi_ctl.parse_response([
        "WFP/1 OK wifi.status state=CONNECTED ssid=%E4%B8%AD%E6%96%87 profiles=1\n",
        "WFP/1 END\n",
    ])
    assert result["fields"]["ssid"] == "中文"


def test_parse_list_decodes_ssid():
    result = wifi_ctl.parse_response([
        "WFP/1 OK wifi.list\n",
        "WFP/1 DATA index=0 ssid=My%20Home%20WiFi proven=1\n",
        "WFP/1 DATA index=1 ssid=a%3Db%26c proven=0\n",
        "WFP/1 END\n",
    ])
    assert result["ok"] and len(result["data"]) == 2
    assert result["data"][0]["ssid"] == "My Home WiFi" and result["data"][0]["proven"] == "1"
    assert result["data"][1]["ssid"] == "a=b&c"


def test_parse_add_verdict():
    result = wifi_ctl.parse_response([
        "WFP/1 OK wifi.add ssid=Powell%20Home result=VERIFIED\n",
        "WFP/1 END\n",
    ])
    assert result["ok"] and result["result"] == "VERIFIED"
    assert result["fields"]["ssid"] == "Powell Home"


def test_parse_error():
    result = wifi_ctl.parse_response([
        "WFP/1 ERROR wifi.add reason=ssid-required\n",
        "WFP/1 END\n",
    ])
    assert not result["ok"] and result["end"] and "ssid-required" in result["error"]


def test_parse_incomplete():
    result = wifi_ctl.parse_response(["WFP/1 OK wifi.list\n"])
    assert result["ok"] and not result["end"]


# ---- Redaction and verdict wording -----------------------------------------

def test_redaction_removes_password():
    assert wifi_ctl.redact("password=s3cret") == "password=***"
    assert "s3cret" not in wifi_ctl.redact("WFP/1 wifi.add ssid=x password=s3cret")
    assert wifi_ctl.redact("no secrets here") == "no secrets here"


def test_add_verdict_wording_is_honest():
    def verdict(value):
        return wifi_ctl.format_result({"end": True, "ok": value == "VERIFIED",
                                       "verb": "wifi.add", "result": value})
    assert "PASS" in verdict("VERIFIED")
    assert "FAIL" in verdict("REJECTED")
    # Unavailable must never read as verified.
    unavailable = verdict("UNAVAILABLE")
    assert "unavailable" in unavailable.lower() and "pass" not in unavailable.lower()
    assert "storage" in verdict("STORAGE_ERROR").lower()
    assert "invalid" in verdict("INVALID").lower()


def test_add_never_prints_password(capsys):
    port = FakePort("WFP/1 OK wifi.add ssid=New-Net result=VERIFIED\nWFP/1 END")
    result = wifi_ctl.send_command(port, "wifi.add", ssid="New-Net", password="top-secret")
    assert result["ok"] and result["end"]
    # The command actually carried the encoded password to the device.
    assert b"password=top-secret" in port.written
    # ...but nothing the helper renders exposes it.
    assert "top-secret" not in wifi_ctl.format_result(result)


def test_format_list_is_human_readable():
    result = wifi_ctl.parse_response([
        "WFP/1 OK wifi.list\n",
        "WFP/1 DATA index=0 ssid=Home-WiFi proven=1\n",
        "WFP/1 DATA index=1 ssid=Office proven=0\n",
        "WFP/1 END\n",
    ])
    text = wifi_ctl.format_result(result)
    assert "1. Home-WiFi (last connected)" in text
    assert "2. Office" in text


def test_format_missing_end_is_not_success():
    result = wifi_ctl.parse_response(["WFP/1 OK wifi.list\n"])
    assert "No response" in wifi_ctl.format_result(result)


# ---- CLI: password never on argv -------------------------------------------

def test_cli_has_no_password_flag():
    parser_names = set()
    import argparse
    original = argparse.ArgumentParser.add_argument

    def capture(self, *names, **kwargs):
        parser_names.update(names)
        return original(self, *names, **kwargs)

    argparse.ArgumentParser.add_argument = capture
    original_stderr = sys.stderr
    sys.stderr = io.StringIO()  # argparse writes its usage error here
    try:
        try:
            wifi_ctl.cli(["/dev/null", "add", "--ssid", "x", "--password", "y"])
        except SystemExit:
            pass
    finally:
        sys.stderr = original_stderr
        argparse.ArgumentParser.add_argument = original
    assert "--password" not in parser_names


def test_cli_add_prompts_with_getpass(monkeypatch, capsys):
    """The password comes from getpass, never from argv."""
    port = FakePort("WFP/1 OK wifi.add ssid=New-Net result=VERIFIED\nWFP/1 END")
    captured = {}

    class Holder:
        def __enter__(self):
            return port

        def __exit__(self, *args):
            return False

    def fake_getpass(prompt=""):
        captured["prompt"] = prompt
        return "prompted-secret"

    monkeypatch.setattr(wifi_ctl, "open_port", lambda path: Holder())
    monkeypatch.setattr(wifi_ctl.getpass, "getpass", fake_getpass)
    code = wifi_ctl.cli(["/dev/null", "add", "--ssid", "New-Net"])
    assert code == 0
    assert captured.get("prompt") == "Password: "
    # The prompted password reached the device, and nothing printed it.
    assert b"password=prompted-secret" in port.written
    out = capsys.readouterr().out
    assert "prompted-secret" not in out
    assert "--password" not in out


def test_cli_add_has_timeout_headroom_over_firmware():
    """The host wait must exceed the firmware candidate budget with headroom."""
    assert wifi_ctl.ADD_TIMEOUT_S > wifi_ctl.FIRMWARE_CANDIDATE_TIMEOUT_S
    # Enough room for disconnect + connect + event dispatch + serial response.
    assert wifi_ctl.ADD_TIMEOUT_S - wifi_ctl.FIRMWARE_CANDIDATE_TIMEOUT_S >= 5.0


def test_add_effective_timeout_exceeds_firmware(monkeypatch):
    """The timeout actually passed to read_response for add is the larger one."""
    recorded = {}

    class Holder:
        def __init__(self):
            self._port = FakePort("WFP/1 OK wifi.add ssid=New-Net result=VERIFIED\nWFP/1 END")

        def __enter__(self):
            return self._port

        def __exit__(self, *args):
            return False

    real_send = wifi_ctl.send_command

    def spy_send(p, verb, *, ssid=None, password=None, timeout=wifi_ctl.DEFAULT_TIMEOUT):
        recorded["timeout"] = timeout
        return real_send(p, verb, ssid=ssid, password=password, timeout=timeout)

    monkeypatch.setattr(wifi_ctl, "open_port", lambda path: Holder())
    monkeypatch.setattr(wifi_ctl, "send_command", spy_send)
    monkeypatch.setattr(wifi_ctl.getpass, "getpass", lambda prompt="": "pw")
    assert wifi_ctl.cli(["/dev/null", "add", "--ssid", "New-Net"]) == 0
    assert recorded["timeout"] > wifi_ctl.FIRMWARE_CANDIDATE_TIMEOUT_S
    assert recorded["timeout"] == wifi_ctl.ADD_TIMEOUT_S
    # A user-supplied larger timeout is still honoured.
    recorded.clear()
    assert wifi_ctl.cli(["/dev/null", "add", "--ssid", "New-Net", "--timeout", "40"]) == 0
    assert recorded["timeout"] == 40.0


def test_cli_add_reports_rejected(monkeypatch, capsys):
    port = FakePort("WFP/1 ERROR wifi.add ssid=Bad result=REJECTED\nWFP/1 END")

    class Holder:
        def __enter__(self):
            return port

        def __exit__(self, *args):
            return False

    monkeypatch.setattr(wifi_ctl, "open_port", lambda path: Holder())
    monkeypatch.setattr(wifi_ctl.getpass, "getpass", lambda prompt="": "wrong")
    code = wifi_ctl.cli(["/dev/null", "add", "--ssid", "Bad"])
    assert code == 1
    assert "FAIL" in capsys.readouterr().out


def test_cli_rejects_missing_ssid(monkeypatch):
    class NoOpen:
        def __enter__(self):
            raise AssertionError("port must not be opened when SSID is missing")

        def __exit__(self, *args):
            return False

    monkeypatch.setattr(wifi_ctl, "open_port", lambda path: NoOpen())
    assert wifi_ctl.cli(["/dev/null", "add"]) == 2
    assert wifi_ctl.cli(["/dev/null", "remove"]) == 2


def test_cli_status_roundtrip(monkeypatch, capsys):
    port = FakePort("WFP/1 OK wifi.status state=OFFLINE ssid=- profiles=0\nWFP/1 END")

    class Holder:
        def __enter__(self):
            return port

        def __exit__(self, *args):
            return False

    monkeypatch.setattr(wifi_ctl, "open_port", lambda path: Holder())
    code = wifi_ctl.cli(["/dev/null", "status"])
    assert code == 0
    assert "state=OFFLINE" in capsys.readouterr().out


def main() -> int:
    tests = [value for name, value in sorted(globals().items())
             if name.startswith("test_") and callable(value)]

    class Capsys:
        def __init__(self):
            self._out = io.StringIO()
            self._err = io.StringIO()

        def readouterr(self):
            out, err = self._out.getvalue(), self._err.getvalue()
            self._out = io.StringIO()
            self._err = io.StringIO()
            return type("R", (), {"out": out, "err": err})()

    class Monkeypatch:
        def __init__(self):
            self._undo = []

        def setattr(self, obj, name, value):
            self._undo.append((obj, name, getattr(obj, name)))
            setattr(obj, name, value)

        def teardown(self):
            for obj, name, value in reversed(self._undo):
                setattr(obj, name, value)
            self._undo.clear()

    failures = 0
    report = []
    for test in tests:
        params = test.__code__.co_varnames[: test.__code__.co_argcount]
        capsys = Capsys()
        monkeypatch = Monkeypatch()
        kwargs = {}
        if "capsys" in params:
            kwargs["capsys"] = capsys
        if "monkeypatch" in params:
            kwargs["monkeypatch"] = monkeypatch
        original_stdout, original_stderr = sys.stdout, sys.stderr
        if "capsys" in params:
            sys.stdout, sys.stderr = capsys._out, capsys._err
        try:
            test(**kwargs)
            report.append(f"ok   {test.__name__}")
        except AssertionError as error:
            failures += 1
            report.append(f"FAIL {test.__name__}: {error}")
        finally:
            sys.stdout, sys.stderr = original_stdout, original_stderr
            monkeypatch.teardown()
    for line in report:
        print(line)
    if failures:
        print(f"{failures} failure(s)", file=sys.stderr)
        return 1
    print("Mac Wi-Fi helper: encoding round-trip, verdicts, redaction, no-argv password, "
          "timeout headroom PASS")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())