# -*- coding: utf-8 -*-
"""
P1-04 平台截图能力单测（portal 用替身 subprocess，不依赖真实 portal / compositor）

覆盖：
  C1 capture_plan 路由（win32/x11 → overlay；wayland → portal；unknown → unavailable+明确日志）
  C2 detect_portal（非 Linux / 无 D-Bus 会话 / 无 gdbus / 齐备 → 四分支）
  C3 portal_screenshot_sync 成功路径（fake subprocess：Screenshot 返回解析 + FilePicked
     file:// URI 解码（含空格文件名）→ 现有附件流程可用的真实文件路径；校验 gdbus 命令行）
  C4 用户取消/授权被拒（只有 Finished 无 FilePicked → denied）
  C5 portal 缺失（call rc≠0 NotSupported → "portal 不可用" + 原因）
  C6 等待超时（显式预算内无信号 → timeout，不无限挂起）
  C7 PortalScreenshotWorker 信号回主线程（done(ok, path)）
  C8 ScreenCaptureOverlay offscreen 构造 + start() 空画面守卫不崩

运行: PYTHONIOENCODING=utf-8 python tests/test_screen_capture_platforms.py
"""
import contextlib
import io
import json
import os
import shutil
import sys
import tempfile
import time
import types
import urllib.parse

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

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")

from PyQt6.QtWidgets import QApplication  # noqa: E402

from ui.views.system_tools import desktop_session as ds  # noqa: E402
from ui.views.system_tools import portal_capture as pc  # noqa: E402
from ui.views.system_tools import screen_capture as sc  # noqa: E402

RESULTS = []
APP = QApplication.instance() or QApplication(sys.argv)


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)


# ======================================================================
# C1. capture_plan 路由
# ======================================================================
mode, msg = ds.capture_plan("win32")
check("C1.1 win32 → overlay", mode == "overlay" and msg is None, repr((mode, msg)))
mode, msg = ds.capture_plan("x11")
check("C1.2 x11 → overlay（原生 grabWindow 路径）", mode == "overlay" and msg is None,
      repr((mode, msg)))
mode, msg = ds.capture_plan("wayland")
check("C1.3 wayland → portal（不绕过 compositor）", mode == "portal" and msg is None,
      repr((mode, msg)))
mode, msg = ds.capture_plan("unknown")
check("C1.4 unknown → unavailable + 明确日志（主程序其余功能不受影响）",
      mode == "unavailable" and msg and "不可用" in msg, repr((mode, msg)))

# ======================================================================
# C2. detect_portal 四分支
# ======================================================================
_orig_sys = pc.sys
_orig_which = pc.shutil.which

try:
    pc.sys = types.SimpleNamespace(platform="win32")
    ok, reason = pc.detect_portal()
    check("C2.1 非 Linux → 不可用", ok is False and reason, repr((ok, reason)))

    pc.sys = types.SimpleNamespace(platform="linux")
    saved = {k: os.environ.get(k) for k in ("XDG_RUNTIME_DIR", "DBUS_SESSION_BUS_ADDRESS")}
    for k in saved:
        os.environ.pop(k, None)
    ok, reason = pc.detect_portal()
    check("C2.2 无 D-Bus 会话 → 不可用（明确原因）",
          ok is False and "D-Bus" in reason, repr((ok, reason)))
    os.environ["XDG_RUNTIME_DIR"] = "/run/user/1000"
    pc.shutil.which = lambda name: None
    ok, reason = pc.detect_portal()
    check("C2.3 无 gdbus CLI → 不可用（明确原因）",
          ok is False and "gdbus" in reason, repr((ok, reason)))
    pc.shutil.which = lambda name: "/usr/bin/gdbus"
    ok, reason = pc.detect_portal()
    check("C2.4 齐备 → 可用", ok is True and reason is None, repr((ok, reason)))
finally:
    pc.sys = _orig_sys
    pc.shutil.which = _orig_which
    for k, v in saved.items():
        if v is None:
            os.environ.pop(k, None)
        else:
            os.environ[k] = v


# ======================================================================
# C3–C7. portal_screenshot_sync（fake subprocess，模拟 Linux 环境）
# ======================================================================
_linux_sys = types.SimpleNamespace(platform="linux")


class FakeRun:
    """替身 subprocess.run：记录 argv，返回预先设定的 call 输出"""
    def __init__(self, rc=0, stdout="", stderr=""):
        self.rc = rc
        self.stdout = stdout
        self.stderr = stderr
        self.cmds = []

    def __call__(self, cmd, **kw):
        self.cmds.append(list(cmd))
        p = types.SimpleNamespace()
        p.returncode = self.rc
        p.stdout = self.stdout
        p.stderr = self.stderr
        return p


class FakePipe:
    """替身 monitor 的 stdout：按序输出 JSON 行；readline 可模拟阻塞"""
    def __init__(self, lines, delay_s=0.0):
        self._lines = list(lines)
        self._delay = delay_s
        self.closed = False

    def readline(self):
        if self._delay:
            time.sleep(self._delay)
            self._delay = 0
        if self._lines:
            return self._lines.pop(0) + "\n"
        # 模拟管道关闭（EOF）
        return ""


class FakePopen:
    def __init__(self, lines, delay_s=0.0):
        self._lines = lines
        self._delay = delay_s
        self.cmd = None
        self.killed = False

    def __call__(self, cmd, **kw):
        self.cmd = list(cmd)
        p = types.SimpleNamespace()
        p.stdout = FakePipe(self._lines, self._delay)
        p.stderr = io.StringIO()
        self._rc = None
        p.poll = lambda: self._rc
        p.kill = lambda: setattr(p, "killed", True) or setattr(self, "killed", True)
        p.wait = lambda timeout=None: 0
        return p


def _uri(path: str) -> str:
    return "file://" + urllib.parse.quote(path)


def _json(member, iface, body):
    return json.dumps({"interface": iface, "member": member, "body": body},
                      ensure_ascii=False)


# C3.1 成功路径（含空格文件名 → URI 解码）
tmpdir = tempfile.mkdtemp(prefix="haocode_p104_portal_")
shot = os.path.join(tmpdir, "shot test.png")
with open(shot, "wb") as f:
    f.write(b"\x89PNG\r\n\x1a\nfake")

pc.sys = _linux_sys   # C3–C7 模拟 Linux 宿主
_orig_which2 = pc.shutil.which
pc.shutil.which = lambda name: "/usr/bin/gdbus"   # 本机无 gdbus，替身探测
_saved_xdg = os.environ.get("XDG_RUNTIME_DIR")
os.environ["XDG_RUNTIME_DIR"] = "/run/user/1000"
run_fake = FakeRun(rc=0, stdout="('/org/freedesktop/portal/desktop/request/1000/haocode/7', <>)")
popen_fake = FakePopen([
    _json("FilePicked", "org.freedesktop.portal.FileChooser", [_uri(shot), {}]),
    _json("Finished", "org.freedesktop.portal.Request", [0]),
])
_orig_run, _orig_popen = pc.subprocess.run, pc.subprocess.Popen
pc.subprocess.run = run_fake
pc.subprocess.Popen = popen_fake
try:
    t0 = time.time()
    ok, result = pc.portal_screenshot_sync("haocode-shot", request_timeout_s=5, wait_budget_s=10)
    dt = time.time() - t0
finally:
    pc.subprocess.run = _orig_run
    pc.subprocess.Popen = _orig_popen

check("C3.1 成功：返回真实文件路径（URI 空格解码正确）",
      ok is True and result == shot and os.path.exists(result), repr((ok, result)))
check("C3.2 Screenshot gdbus 命令行正确（dest/path/method/parent=/）",
      run_fake.cmds and run_fake.cmds[0][:5] == ["gdbus", "call", "--session", "--dest",
                                                 pc.PORTAL_DEST]
      and "--method=org.freedesktop.portal.Screenshot.Screenshot" in run_fake.cmds[0]
      and "/" in run_fake.cmds[0], repr(run_fake.cmds))
check("C3.3 monitor 监听 request 对象路径",
      popen_fake.cmd and popen_fake.cmd[-2:] == ["--object-path",
                                                 "/org/freedesktop/portal/desktop/request/1000/haocode/7"],
      repr(popen_fake.cmd))
check("C3.4 全程在显式预算内完成（<10s）", dt < 10, f"{dt:.1f}s")

# C4. 用户取消/授权被拒（只有 Finished）
popen_fake = FakePopen([_json("Finished", "org.freedesktop.portal.Request", [0])])
pc.subprocess.run = run_fake
pc.subprocess.Popen = popen_fake
try:
    ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 10)
finally:
    pc.subprocess.run = _orig_run
    pc.subprocess.Popen = _orig_popen
check("C4.1 授权被拒 → 不伪造成功，明确原因",
      ok is False and "取消" in result, repr((ok, result)))

# C5. portal 缺失/方法不支持（call rc≠0）
run_fake2 = FakeRun(rc=1, stderr="Error: org.freedesktop.DBus.Error.NotSupported: "
                                 "Method not supported by portal")
pc.subprocess.run = run_fake2
pc.subprocess.Popen = FakePopen([])
try:
    ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 10)
finally:
    pc.subprocess.run = _orig_run
    pc.subprocess.Popen = _orig_popen
check("C5.1 portal 不支持 → 不可用 + 原因透出",
      ok is False and "portal 不可用" in result and "NotSupported" in result, repr((ok, result)))

# C6. 等待超时（monitor 首行前阻塞 > 预算）
popen_fake = FakePopen([_json("Finished", "org.freedesktop.portal.Request", [0])],
                       delay_s=3.0)
pc.subprocess.run = run_fake
pc.subprocess.Popen = popen_fake
try:
    t0 = time.time()
    ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 1.0)
    dt = time.time() - t0
finally:
    pc.subprocess.run = _orig_run
    pc.subprocess.Popen = _orig_popen
check("C6.1 超预算 → timeout（不无限挂起）",
      ok is False and "超时" in result and dt < 5, repr((ok, result, f"{dt:.1f}s")))

# ======================================================================
# C7. PortalScreenshotWorker 信号
# ======================================================================
popen_fake = FakePopen([
    _json("FilePicked", "org.freedesktop.portal.FileChooser", [_uri(shot), {}]),
    _json("Finished", "org.freedesktop.portal.Request", [0]),
])
pc.subprocess.run = run_fake
pc.subprocess.Popen = popen_fake
try:
    w = pc.PortalScreenshotWorker(wait_budget_s=10)
    done = {}
    w.done.connect(lambda ok, r: done.update(ok=ok, r=r))
    w.start()
    deadline = time.time() + 15
    while "ok" not in done and time.time() < deadline:
        APP.processEvents()
        time.sleep(0.02)
    w.wait(3000)
finally:
    pc.subprocess.run = _orig_run
    pc.subprocess.Popen = _orig_popen
    pc.shutil.which = _orig_which2
    if _saved_xdg is None:
        os.environ.pop("XDG_RUNTIME_DIR", None)
    else:
        os.environ["XDG_RUNTIME_DIR"] = _saved_xdg
    pc.sys = _orig_sys
check("C7.1 worker done 信号回主线程（ok + 路径）",
      done.get("ok") is True and done.get("r") == shot, repr(done))

# ======================================================================
# C8. 覆盖层 offscreen 构造 + start() 空画面守卫
# ======================================================================
ov = sc.ScreenCaptureOverlay()
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
    ov.start()   # offscreen：无屏幕或空画面都不得崩
shown = ov.isVisible()
check("C8.1 offscreen start() 不崩；空画面时不显示覆盖层且日志明确",
      (not shown and "空画面" in buf.getvalue()) or (shown and ov._full_pixmap is not None),
      f"shown={shown} log={buf.getvalue()[-160:]!r}")
if shown:
    ov.close()

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)