"""Behavior tests for the Windows desktop-exe integrity gate (#69179).

The desktop self-update chain (Desktop → hermes-setup --update →
``hermes update`` → ``hermes desktop --build-only`` → relaunch) rebuilds
Hermes.exe on the end user's machine. Before this gate, "build succeeded" was
just "the file exists", so a truncated PE (corrupt cached Electron zip), a
non-PE file, or a wrong-architecture tree shipped as the new app — Windows
then refuses to launch it with "This app can't run on your computer"
(此应用无法在你的电脑上运行) and the user has no working install left.

These tests exercise PE validation and rejection before publication. Windows
host/architecture probes stay native; the Linux callback witness checks staging.
"""

from __future__ import annotations

import argparse
import struct
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from hermes_cli import main as cli_main
from hermes_cli import main_desktop
from hermes_platform.host import facts

PE_AMD64 = 0x8664
PE_ARM64 = 0xAA64
PE_I386 = 0x014C


@pytest.fixture(autouse=True)
def _clear_host_fact_caches():
    facts.clear_caches()
    yield
    facts.clear_caches()


def make_pe(path: Path, machine: int = PE_AMD64, *, truncate_to: int | None = None) -> Path:
    """Write a minimal, structurally-complete PE file with one section.

    Layout: DOS header (e_lfanew=0x80) → PE signature + COFF header at 0x80 →
    one 40-byte section entry whose raw data spans [0x200, 0x400). Total file
    size 0x400 unless ``truncate_to`` cuts it short (the corrupt-download
    shape).
    """
    buf = bytearray(0x400)
    buf[0:2] = b"MZ"
    struct.pack_into("<I", buf, 0x3C, 0x80)
    buf[0x80:0x84] = b"PE\x00\x00"
    # COFF: machine, nsections=1, timestamp, symtab ptr, nsyms, opt-hdr size=0, chars
    struct.pack_into("<HHIIIHH", buf, 0x84, machine, 1, 0, 0, 0, 0, 0x0002)
    # Section table entry at 0x80 + 24 = 0x98; SizeOfRawData @+16, PointerToRawData @+20
    section_off = 0x98
    struct.pack_into("<II", buf, section_off + 16, 0x200, 0x200)
    data = bytes(buf if truncate_to is None else buf[:truncate_to])
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    return path


# ─── _parse_pe_machine ──────────────────────────────────────────────────────






# ─── _expected_windows_pe_machines ──────────────────────────────────────────




# ─── _windows_native_machine ────────────────────────────────────────────────

# MACHINE_ATTRIBUTES.UserEnabled — mirrors hermes_cli.main.
_USER_ENABLED = 0x00000001


class _SettableFunc:
    """Callable that accepts ctypes ``.restype`` / ``.argtypes`` assignment."""

    def __init__(self, fn):
        self._fn = fn
        self.restype = None
        self.argtypes = None

    def __call__(self, *args, **kwargs):
        return self._fn(*args, **kwargs)


class _FakeKernel32:
    """Stands in for kernel32 so the native-arch probes run on any CI OS."""

    def __init__(
        self,
        native_pe_machine: int,
        *,
        wow64_ok: bool = True,
        user_runnable: set | None = None,
    ):
        self._native_pe = native_pe_machine
        self._wow64_ok = wow64_ok
        self._user_runnable = user_runnable
        self.GetCurrentProcess = _SettableFunc(lambda: -1)
        self.IsWow64Process2 = _SettableFunc(self._is_wow64_process2)
        self.GetMachineTypeAttributes = _SettableFunc(
            self._get_machine_type_attributes
        )

    def _is_wow64_process2(self, _handle, process_ref, native_ref):
        if not self._wow64_ok:
            return 0
        process_ref._obj.value = PE_AMD64  # emulated x64 process
        native_ref._obj.value = self._native_pe
        return 1

    def _get_machine_type_attributes(self, machine, attributes_ref):
        if self._user_runnable is None:
            raise AttributeError("GetMachineTypeAttributes unavailable in this fake")
        if machine not in self._user_runnable:
            attributes_ref._obj.value = 0
            return 0
        attributes_ref._obj.value = _USER_ENABLED
        return 0


def _fake_windll(
    native_pe_machine: int,
    *,
    wow64_ok: bool = True,
    user_runnable: set | None = None,
):
    kernel32 = _FakeKernel32(
        native_pe_machine,
        wow64_ok=wow64_ok,
        user_runnable=user_runnable,
    )

    def _windll(name, *args, **kwargs):
        assert "kernel32" in str(name).lower()
        return kernel32

    return _windll


@pytest.mark.platforms("windows")
def test_native_machine_reports_os_arch_not_process_arch():
    """The #69179 WoA regression: x64 Python under ARM64 emulation must report
    ARM64 (the OS), not AMD64 (the process) — otherwise the integrity gate
    rejects the correct ARM64 rebuild.

    ``platforms("windows")``: the probe under test is a ``ctypes.WinDLL("kernel32")``
    call to ``IsWow64Process2``. A patched ``sys.platform`` only got the branch
    entered — there is no kernel32 to bind on Linux, so nothing below the
    branch (the HANDLE typing that #71218 was about) was ever executed.
    """
    import ctypes

    # WinDLL only exists on Windows; create=True so Linux/macOS CI can stub it.
    with patch.object(ctypes, "WinDLL", _fake_windll(PE_ARM64), create=True), \
         patch("platform.machine", return_value="AMD64"):
        assert main_desktop._windows_native_machine() == "ARM64"


@pytest.mark.platforms("windows")
def test_expected_machines_prefers_user_runnable_api_over_arch_name(monkeypatch):
    """GetMachineTypeAttributes answers "can this host load PE machine X?"
    directly, so a WoA host that reports AMD64 everywhere else still accepts an
    ARM64 exe.

    ``platforms("windows")``: ``GetMachineTypeAttributes`` is a real kernel32 export
    the fake host could not provide.
    """
    import ctypes

    monkeypatch.setenv("PROCESSOR_ARCHITECTURE", "AMD64")
    monkeypatch.delenv("PROCESSOR_ARCHITEW6432", raising=False)
    with patch.object(
        ctypes,
        "WinDLL",
        _fake_windll(
            PE_ARM64, wow64_ok=False, user_runnable={PE_ARM64, PE_AMD64}
        ),
        create=True,
    ), patch("platform.machine", return_value="AMD64"):
        assert main_desktop._expected_windows_pe_machines() == {PE_ARM64, PE_AMD64}








# ─── _desktop_exe_integrity_error ───────────────────────────────────────────




# ─── _desktop_packaged_executable arch preference (win32) ───────────────────






# ─── staged integrity gate ─────────────────────────────────────────────────


@pytest.mark.platforms("linux")
def test_corrupt_staged_app_keeps_live_bytes_without_backup_recovery(tmp_path, monkeypatch, capsys):
    """Exercise publication and PE parsing, not Windows host/architecture discovery."""
    desktop_dir = tmp_path / "apps" / "desktop"
    live_exe = desktop_dir / "release" / "linux-unpacked" / "hermes"
    make_pe(live_exe)
    live_bytes = live_exe.read_bytes()
    assert main_desktop._parse_pe_machine(live_exe) == PE_AMD64
    # The host-native layout makes discovery/swap real; PE is just fixture data.
    staging = main_desktop._desktop_staging_dir(desktop_dir)
    staged_exe = make_pe(staging / "linux-unpacked" / "hermes", truncate_to=0x300)
    staged_bytes = staged_exe.read_bytes()
    backup_exe = make_pe(staging / "linux-unpacked.bak" / "hermes")
    old_diagnostic = staging / "linux-unpacked.corrupt" / "diagnostic"
    old_diagnostic.parent.mkdir()
    old_diagnostic.write_bytes(b"previous diagnostic")
    # Raw in-place pack recovery material is not owned by this transaction.
    live_backup = make_pe(live_exe.parent.with_name("linux-unpacked.bak") / "hermes")

    checked = []

    def check_pe(path):
        checked.append(path)
        try:
            main_desktop._parse_pe_machine(path)
        except ValueError as exc:
            return str(exc)
        return None

    discard = main_desktop._discard_desktop_staging

    def check_before_discard(path):
        assert path == staging
        assert staged_exe.read_bytes() == staged_bytes
        assert backup_exe.read_bytes() == live_bytes
        assert old_diagnostic.read_bytes() == b"previous diagnostic"
        discard(path)

    monkeypatch.setattr(main_desktop, "_discard_desktop_staging", check_before_discard)
    with patch.object(main_desktop.os, "rename", wraps=main_desktop.os.rename) as rename:
        with pytest.raises(RuntimeError, match="previous desktop app was left untouched"):
            main_desktop._promote_staged_desktop_app(desktop_dir, staging, integrity_check=check_pe)
        rename.assert_not_called()

    assert checked == [staged_exe]
    assert live_exe.read_bytes() == live_bytes
    assert live_backup.read_bytes() == live_bytes
    assert not staging.exists()
    assert not list((desktop_dir / "release").glob("*.previous"))
    out = capsys.readouterr().out
    assert "integrity check" in out
    assert "truncated executable" in out


@pytest.mark.platforms("windows")
def test_gate_fails_clearly_without_backup(tmp_path, capsys):
    """Exercise the default Windows PE check, without injecting a validator."""
    desktop_dir = tmp_path / "apps" / "desktop"
    staging = main_desktop._desktop_staging_dir(desktop_dir)
    exe = staging / "win-unpacked" / "Hermes.exe"
    exe.parent.mkdir(parents=True)
    exe.write_bytes(b"<html>proxy error</html>" + b" " * 600)

    with pytest.raises(RuntimeError, match="produced no launchable app"):
        main_desktop._promote_staged_desktop_app(desktop_dir, staging)

    assert not staging.exists()
    assert not (desktop_dir / "release").exists()
    out = capsys.readouterr().out
    assert "integrity check" in out
    assert "missing MZ header" in out


# ─── end-to-end: `hermes desktop --build-only` exits nonzero on corrupt exe ─


def _ns(**kw):
    defaults = dict(
        skip_build=False,
        build_only=True,
        force_build=False,
        source=False,
        fake_boot=False,
        ignore_existing=False,
        hermes_root=None,
        cwd=None,
    )
    defaults.update(kw)
    return argparse.Namespace(**defaults)


@pytest.mark.platforms("windows")
def test_build_only_fails_when_pack_produces_corrupt_exe(tmp_path, monkeypatch, capsys):
    """The updater chain's contract: a rebuild whose Hermes.exe cannot launch
    must exit nonzero (so hermes-setup's retry-once kicks in) and must leave
    the previous working build in place instead of installing the corrupt one.

    Stage-and-swap (#86443): the pack lands in a staging dir; the integrity
    gate runs on the STAGED exe and a failure discards staging without ever
    touching the live ``win-unpacked`` tree.

    ``platforms("windows")``: the whole chain is Windows-gated — ``win-unpacked``
    candidate discovery in ``_desktop_packaged_executable`` and the integrity
    gate itself both short-circuit off Windows.
    """
    root = tmp_path / "hermes-agent"
    desktop_dir = root / "apps" / "desktop"
    desktop_dir.mkdir(parents=True)
    (desktop_dir / "package.json").write_text("{}", encoding="utf-8")
    monkeypatch.setattr(cli_main, "PROJECT_ROOT", root)

    live_exe = desktop_dir / "release" / "win-unpacked" / "Hermes.exe"
    make_pe(live_exe, PE_AMD64)  # the previous, working app
    live_bytes = live_exe.read_bytes()

    def pack_into_staging(cmd, *args, **kwargs):
        assert kwargs["env"]["PATH"].startswith("C:\\pm-pinned-git\\cmd;")
        if cmd[1:3] != ["run", "builder"]:
            return subprocess.CompletedProcess(list(cmd), 0)
        # electron-builder honours -c.directories.output=<staging>; emulate a
        # pack that "succeeds" but writes a truncated exe there.
        out_flag = next((a for a in cmd if str(a).startswith("-c.directories.output=")), None)
        assert out_flag is not None, "pack must be redirected into a staging dir"
        staging = Path(str(out_flag).split("=", 1)[1])
        make_pe(staging / "win-unpacked" / "Hermes.exe", PE_AMD64, truncate_to=0x300)
        return subprocess.CompletedProcess(list(cmd), 0)

    def pinned_git(name, *, base_env):
        assert name == "git"
        return SimpleNamespace(env={**base_env, "PATH": "C:\\pm-pinned-git\\cmd;" + base_env["PATH"]})

    with patch("hermes_cli.main_desktop.shutil.which", return_value="/usr/bin/npm"), \
         patch("hermes_cli.source_build.source_build_env", return_value={"PATH": "/usr/bin"}), \
         patch("hermes_cli.source_build.prepare_source_dependencies"), \
         patch("pm.ensure", side_effect=pinned_git), \
         patch("hermes_cli.main_desktop._desktop_build_needed", return_value=True), \
         patch("hermes_cli.main_desktop._stop_desktop_processes_locking_build", return_value=[]), \
         patch("hermes_cli.main_desktop._windows_native_machine", return_value="AMD64"), \
         patch("hermes_cli.main_desktop.subprocess.run", side_effect=pack_into_staging), \
         pytest.raises(SystemExit) as exc:
        cli_main.cmd_gui(_ns())

    assert exc.value.code == 1
    # The previous working exe was never touched...
    assert live_exe.read_bytes() == live_bytes
    assert main_desktop._parse_pe_machine(live_exe) == PE_AMD64
    # ...the staged corrupt tree was discarded...
    assert not list(desktop_dir.glob(".staging-*"))

    out = capsys.readouterr().out
    assert "integrity check" in out