# -*- coding: utf-8 -*-
"""
P1-02 跨平台 shell 契约单测（纯函数级 + 真实进程级，无 UI 依赖）

覆盖：
  A. 平台参数：Windows `cmd.exe /d /c` argv / Linux `/bin/bash -lc` argv；
     Linux Popen 独立进程组（start_new_session），Windows 无该参数
  B. 提示词平台段：通用正文完全相同、只插入对应平台段、互不串段；
     load_system_prompt() 与文件内容一致（当前平台）
  C. 进程树（真实进程，父+孙）：
     - 当前平台：超时 → 父与孙都不存在
     - 当前平台：主动中止 → 父与孙都不存在
     - Linux 专属：独立进程组整组终止（Windows 上跳过）
  D. kill_process_tree 安全边界：已退出的进程 / 无独立组 → 不抛异常、不误杀

运行: PYTHONIOENCODING=utf-8 python tests/test_cross_platform_shell.py
"""
import os
import subprocess
import sys
import tempfile
import time

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from core import platform_shell  # noqa: E402
from core.agent.tools import tool_bash  # noqa: E402
from core.agent.types import AbortSignal  # noqa: E402

RESULTS = []


def check(name, cond, extra=""):
    RESULTS.append((name, bool(cond)))
    print(f"{'PASS' if cond else 'FAIL'}  {name}  {extra if not cond else ''}", flush=True)


def text_of(res):
    c = res.content
    if isinstance(c, str):
        return c
    return "".join(x.get("text", "") for x in c if isinstance(x, dict))


_TMP = tempfile.mkdtemp(prefix="haocode_crossshell_")
PY = sys.executable
IS_WIN = os.name == "nt"

# ======================================================================
# A. 平台参数（mock 平台，纯逻辑）
# ======================================================================
_orig_is_win = platform_shell.is_windows


def _mock_win(v):
    platform_shell.is_windows = lambda: v


try:
    _mock_win(True)
    check("A1 Windows 命令 = cmd.exe /d /s /c \"<cmd>\"",
          platform_shell.shell_command("echo hi")
          == 'cmd.exe /d /s /c "echo hi"',
          repr(platform_shell.shell_command("echo hi")))
    check("A2 Windows popen_flags 无进程组参数",
          platform_shell.popen_flags() == {}, repr(platform_shell.popen_flags()))

    _mock_win(False)
    check("A3 Linux argv = /bin/bash -lc <cmd>",
          platform_shell.shell_command("echo hi") == ["/bin/bash", "-lc", "echo hi"],
          repr(platform_shell.shell_command("echo hi")))
    check("A4 Linux popen_flags 独立进程组",
          platform_shell.popen_flags() == {"start_new_session": True},
          repr(platform_shell.popen_flags()))
finally:
    platform_shell.is_windows = _orig_is_win

# ======================================================================
# B. 提示词平台段
# ======================================================================
_PROMPT_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                            "SYSTEM_PROMPT.md")
with open(_PROMPT_FILE, "r", encoding="utf-8") as f:
    _PROMPT_TEXT = f.read().strip()

check("B1 通用正文含占位符",
      platform_shell.SHELL_SECTION_PLACEHOLDER in _PROMPT_TEXT)

_GENERIC_NO_LEAK = ("cmd.exe" not in _PROMPT_TEXT
                    and "/bin/bash" not in _PROMPT_TEXT)
check("B2 通用正文无平台泄漏（无 cmd.exe / 无 /bin/bash）", _GENERIC_NO_LEAK)

try:
    _mock_win(True)
    full_win = platform_shell.apply_platform_section(_PROMPT_TEXT)
    sec_win = platform_shell.shell_prompt_section()
    _mock_win(False)
    full_lin = platform_shell.apply_platform_section(_PROMPT_TEXT)
    sec_lin = platform_shell.shell_prompt_section()
finally:
    platform_shell.is_windows = _orig_is_win

check("B3 Windows 段含 cmd.exe、无 /bin/bash",
      "cmd.exe" in sec_win and "/bin/bash" not in sec_win)
check("B4 Linux 段含 /bin/bash -lc、无 cmd.exe",
      "/bin/bash -lc" in sec_lin and "cmd.exe" not in sec_lin)
check("B5 只有对应平台段被插入",
      full_win == _PROMPT_TEXT.replace(platform_shell.SHELL_SECTION_PLACEHOLDER, sec_win)
      and full_lin == _PROMPT_TEXT.replace(platform_shell.SHELL_SECTION_PLACEHOLDER, sec_lin))
check("B6 通用正文两平台完全相同",
      full_win.replace(sec_win, "§SEC§") == full_lin.replace(sec_lin, "§SEC§"))
check("B7 无占位符文本原样返回",
      platform_shell.apply_platform_section("无占位符的兜底文本") == "无占位符的兜底文本")

# load_system_prompt()（当前平台，真实文件）
from core import llm_engine  # noqa: E402  (PyQt6/openai 已装，无需 QApplication)
check("B8 load_system_prompt = 文件内容 + 当前平台段",
      llm_engine.load_system_prompt()
      == platform_shell.apply_platform_section(_PROMPT_TEXT))

# ======================================================================
# C. 进程树（真实进程：父 + 孙，心跳文件证明生死）
# ======================================================================
_CHILD = os.path.join(_TMP, "child.py")
with open(_CHILD, "w", encoding="utf-8") as f:
    f.write(
        "import sys, time\n"
        "hb = open(sys.argv[1], 'a', encoding='utf-8')\n"
        "t0 = time.time()\n"
        "while time.time() - t0 < 30:\n"
        "    hb.write(f'{time.time():.3f}\\n')\n"
        "    hb.flush()\n"
        "    time.sleep(0.2)\n"
        "open(sys.argv[2], 'w').write('done')\n"
    )


def _spawn_parent(hb_path, done_path, out_path):
    """写一个父进程脚本（启动孙进程后挂 30s），返回 bash 命令字符串。"""
    parent = os.path.join(_TMP, f"parent_{os.path.basename(hb_path)}.py")
    with open(parent, "w", encoding="utf-8") as f:
        f.write(
            "import subprocess, sys, time\n"
            "p = subprocess.Popen([sys.executable, "
            + repr(_CHILD) + ", sys.argv[1], sys.argv[2]], "
            "stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)\n"
            "time.sleep(30)\n"
        )
    cmd = f'"{PY}" -u "{parent}" "{hb_path}" "{done_path}"'
    return cmd


def _hb_last(hb_path):
    try:
        with open(hb_path, "r", encoding="utf-8") as f:
            lines = [l.strip() for l in f if l.strip()]
        return float(lines[-1]) if lines else None
    except FileNotFoundError:
        return None


def _tree_dead(hb_path, done_path, quiesce=1.2):
    """等待静默后判断：孙进程不再有心跳、且未跑完 → 进程树已死。"""
    time.sleep(quiesce)
    last = _hb_last(hb_path)
    return last is not None and (time.time() - last) > 0.6, os.path.exists(done_path)


def run_tree_case(tag, timeout=None, abort_after=None):
    hb = os.path.join(_TMP, f"hb_{tag}.txt")
    done = os.path.join(_TMP, f"done_{tag}.txt")
    out = os.path.join(_TMP, f"out_{tag}.log")
    cmd = _spawn_parent(hb, done, out)

    updates, timers = [], []
    sig = AbortSignal()
    if abort_after is not None:
        import threading
        def _aborter():
            time.sleep(abort_after)
            sig.abort("test")
        threading.Thread(target=_aborter, daemon=True).start()

    t0 = time.time()
    res = tool_bash(f"call_{tag}", {"command": cmd, "timeout": timeout or 60}, sig,
                    lambda t: updates.append(t),
                    {"cwd": _TMP, "on_timer": lambda e, tt: timers.append((e, tt))})
    dur = time.time() - t0
    return res, dur, hb, done, timers


# ---- C1: 超时 → 父与孙都不存在 ----
res, dur, hb, done, timers = run_tree_case("timeout", timeout=3)
txt = text_of(res)
dead, finished = _tree_dead(hb, done)
check("C1.1 超时返回错误结果", res.is_error and "超时" in txt, f"{txt!r}")
check("C1.2 超时快速返回（<15s，含 taskkill 缓冲）", dur < 15, f"dur={dur:.1f}s")
check("C1.3 超时后父+孙都不存在（心跳停止且未跑完）", dead and not finished,
      f"hb_last={_hb_last(hb)} finished={finished}")
check("C1.4 读秒滴答仍工作", len(timers) >= 2, f"timers={timers}")

# ---- C2: 主动中止 → 父与孙都不存在 ----
res, dur, hb, done, _timers2 = run_tree_case("abort", timeout=60, abort_after=1.5)
txt = text_of(res)
dead, finished = _tree_dead(hb, done)
check("C2.1 中止返回错误结果", res.is_error and "中止" in txt, f"{txt!r}")
check("C2.2 中止后父+孙都不存在（心跳停止且未跑完）", dead and not finished,
      f"hb_last={_hb_last(hb)} finished={finished}")

# ---- C3: Linux 独立进程组整组终止（Windows 跳过） ----
if not IS_WIN:
    proc = subprocess.Popen(
        platform_shell.shell_command("sleep 30"),
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        **platform_shell.popen_flags())
    time.sleep(0.3)
    try:
        pgid = os.getpgid(proc.pid)
        check("C3.1 start_new_session 生效（pgid == 子 pid）", pgid == proc.pid,
              f"pgid={pgid} pid={proc.pid}")
        platform_shell.kill_process_tree(proc, grace_s=1.0)
        time.sleep(0.5)
        check("C3.2 整组终止后子进程不存在",
              proc.poll() is not None, f"poll={proc.poll()}")
    except Exception as e:
        try:
            platform_shell.kill_process_tree(proc, grace_s=1.0)
        except Exception:
            pass
        check("C3 Linux 进程组用例", False, repr(e))
else:
    print("SKIP  C3 Linux 进程组用例（当前为 Windows，Windows 用例见 C1/C2）", flush=True)

# ======================================================================
# D. kill_process_tree 安全边界
# ======================================================================
_p = subprocess.Popen([PY, "-c", "pass"])
_p.wait(timeout=5)
try:
    platform_shell.kill_process_tree(_p, grace_s=0.5)
    check("D1 已退出进程：不抛异常", True)
except Exception as e:
    check("D1 已退出进程：不抛异常", False, repr(e))

try:
    platform_shell.kill_process_tree(None)
    check("D2 None 输入：不抛异常", True)
except Exception as e:
    check("D2 None 输入：不抛异常", False, repr(e))

# ======================================================================
print(f"\n===== {sum(1 for _, ok in RESULTS if ok)}/{len(RESULTS)} passed =====", flush=True)
sys.exit(0 if all(ok for _, ok in RESULTS) else 1)