# -*- coding: utf-8 -*-
"""
tool_bash 实时输出（增量流）单测 —— 纯函数级，无 UI 依赖

本轮改动核心：communicate(timeout=1) 阻塞式收集
            → stdout/stderr reader 线程 + 队列 + 主循环抽干回调 on_update
本测试锁死「改动没有破坏任何既有语义」+「输出真的是实时的」。

覆盖：
  T1    增量流：on_update 被多次调用（不是结束时一次性给）
  T2    实时性：第一块输出到达时刻 << 总耗时（证明是流式而非收尾补发）
  T3    上下文文本格式完全不变（$ cmd / stdout / [stderr] / [exit N] (Ns)）
  T4    stderr 实时流带 [stderr] 分隔标记，且最终文本仍有 [stderr] 段
  T5    每 1 秒读秒回调 on_timer 仍然工作
  T6    超时：杀进程树 + 「命令超时（>Ns）已终止」+ 实际耗时远小于命令时长
  T7    中止：AbortSignal.aborted → 「操作已中止」
  T8    50KB 截断仍然生效
  T9    退出非 0 → is_error，且 exit code 写入 details
  T10   空命令 → 参数校验错误
  T11   无输出命令：不产生 on_update，但读秒正常
  T12   最终文本 = 全部 stdout 拼接（不丢不重）

运行: PYTHONIOENCODING=utf-8 python tests/test_bash_stream.py
"""
import os
import re
import sys
import time
import tempfile
import shutil

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

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)


_TMP = tempfile.mkdtemp(prefix="hocode_bashstream_")
PY = sys.executable


def run(script=None, timeout=None, abort_after=None, shell_script=None):
    """把脚本写成临时文件再执行（避开 cmd 引号地狱）。
    返回 (result, updates, first_t, dur, timer_calls)"""
    path = None
    if script is not None:
        path = os.path.join(_TMP, f"s_{int(time.time() * 1000000) % 10**9}.py")
        with open(path, "w", encoding="utf-8") as f:
            f.write(script)
        cmd = f'"{PY}" -u "{path}"'
    else:
        cmd = shell_script

    updates = []
    first_t = [None]
    t0 = time.time()

    def on_update(text):
        if first_t[0] is None:
            first_t[0] = time.time() - t0
        updates.append(text)

    timers = []

    def on_timer(elapsed, total):
        timers.append((elapsed, total))

    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()

    args = {"command": cmd}
    if timeout is not None:
        args["timeout"] = timeout
    res = tool_bash("call_test", args, sig, on_update,
                    {"cwd": _TMP, "shell": True, "on_timer": on_timer})
    dur = time.time() - t0
    return res, updates, first_t[0], dur, timers


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))


# ======================================================================
try:
    # ---------- T1/T2/T3/T12: 流式 + 上下文格式 ----------
    res, ups, first_t, dur, timers = run(
        "import sys, time\n"
        "for i in range(3):\n"
        "    print('LINE%d' % i, flush=True)\n"
        "    time.sleep(0.5)\n", timeout=30)
    txt = text_of(res)
    check("T1.1 增量流被多次回调（>=3 块）", len(ups) >= 3, f"updates={len(ups)} {ups!r}")
    check("T1.2 每块都是本次输出的一部分", all(u.strip() for u in ups), repr(ups[:3]))
    check("T2.1 第一块在 1s 内到达（实时，不是收尾补发）",
          first_t is not None and first_t < 1.0, f"first_t={first_t}")
    check("T2.2 第一块明显早于总耗时",
          first_t is not None and (dur - first_t) > 0.6, f"first={first_t} dur={dur:.2f}")
    check("T3.1 上下文文本以 $ 命令开头", txt.startswith("$ "), repr(txt[:60]))
    check("T3.2 含 [exit 0] 与耗时",
          re.search(r"\[exit 0\] \(\d+\.\d+s\)", txt) is not None, repr(txt[-60:]))
    check("T12.1 三行输出全部进入最终文本，不丢不重",
          all(f"LINE{i}" in txt for i in range(3)) and txt.count("LINE0") == 1, repr(txt))
    check("T3.3 is_error=False", res.is_error is False)
    check("T3.4 details 带 exit_code", res.details.get("exit_code") == 0, str(res.details))

    # ---------- T4: stderr 实时 + 最终段 ----------
    res, ups, first_t, dur, timers = run(
        "import sys\nprint('OUT1', flush=True)\nprint('ERR1', file=sys.stderr, flush=True)\n",
        timeout=30)
    txt = text_of(res)
    joined = "".join(ups)
    check("T4.1 实时流里有 [stderr] 分隔标记", "[stderr]" in joined, repr(joined))
    check("T4.2 最终文本含 [stderr] 段与内容",
          "[stderr]" in txt and "ERR1" in txt, repr(txt))
    check("T4.3 stdout 内容也在最终文本里", "OUT1" in txt, repr(txt))

    # ---------- T5/T11: 读秒 ----------
    res, ups, first_t, dur, timers = run(
        "import time\ntime.sleep(2.6)\nprint('done', flush=True)\n", timeout=30)
    check("T5.1 长命令期间读秒回调（>=2 次）", len(timers) >= 2, f"timers={timers}")
    check("T5.2 读秒总量正确", all(t[1] == 30 for t in timers), str(timers))

    res, ups, first_t, dur, timers = run(
        "import time\ntime.sleep(1.5)\n", timeout=30)
    check("T11.1 无输出命令不产生 on_update", len(ups) == 0, repr(ups))
    check("T11.2 无输出命令仍有读秒", len(timers) >= 1, str(timers))

    # ---------- T6: 超时杀树 ----------
    res, ups, first_t, dur, timers = run(
        "import time\nprint('start', flush=True)\ntime.sleep(30)\n", timeout=2)
    txt = text_of(res)
    check("T6.1 超时文案", "命令超时（>2s）已终止" in txt, repr(txt))
    check("T6.2 真的提前返回（耗时应 <10s）", dur < 10, f"dur={dur:.2f}")
    check("T6.3 超时结果是 is_error", res.is_error is True)

    # ---------- T7: 中止 ----------
    res, ups, first_t, dur, timers = run(
        "import time\ntime.sleep(30)\n", timeout=60, abort_after=1.5)
    txt = text_of(res)
    check("T7.1 中止文案", "操作已中止" in txt, repr(txt))
    check("T7.2 中止后迅速返回（<8s）", dur < 8, f"dur={dur:.2f}")

    # ---------- T8: 50KB 截断 ----------
    res, ups, first_t, dur, timers = run(
        "import sys\n"
        "line = 'X' * 100 + '\\n'\n"
        "for _ in range(1200):\n"
        "    sys.stdout.write(line)\n"
        "sys.stdout.flush()\n", timeout=30)
    txt = text_of(res)
    check("T8.1 超过 50KB 触发截断标记", "[输出超过 50KB 已截断]" in txt, repr(txt[-80:]))
    check("T8.2 截断后字节数 <= 50KB+余量",
          len(txt.encode("utf-8")) <= 50 * 1024 + 200, str(len(txt.encode("utf-8"))))

    # ---------- T9: 非 0 退出 ----------
    res, ups, first_t, dur, timers = run("import sys\nsys.exit(3)\n", timeout=30)
    txt = text_of(res)
    check("T9.1 exit 3 写入文本", "[exit 3]" in txt, repr(txt))
    check("T9.2 is_error=True", res.is_error is True)
    check("T9.3 details.exit_code=3", res.details.get("exit_code") == 3, str(res.details))

    # ---------- T10: 空命令 ----------
    res = tool_bash("c", {"command": "   "}, AbortSignal(), None,
                    {"cwd": _TMP, "shell": True})
    check("T10.1 空命令被拒", "command 不能为空" in text_of(res), repr(text_of(res)))

    # ---------- T13: 秒级短命令仍正常（回归：不因线程化而变慢/丢输出） ----------
    res, ups, first_t, dur, timers = run(shell_script="echo hello_from_cmd")
    txt = text_of(res)
    check("T13.1 短命令输出正确", "hello_from_cmd" in txt, repr(txt))
    check("T13.2 短命令耗时 <3s", dur < 3, f"dur={dur:.2f}")
    check("T13.3 短命令实时流也拿到了输出", "hello_from_cmd" in "".join(ups), repr(ups))

finally:
    shutil.rmtree(_TMP, ignore_errors=True)

failed = [n for n, ok in RESULTS if not ok]
print(f"\n===== {len(RESULTS) - len(failed)}/{len(RESULTS)} PASS =====", flush=True)
print("ALL PASS" if not failed else f"FAILED: {failed}", flush=True)
sys.exit(0 if not failed else 1)