ykxmz/_tests/test_new_features.py

# -*- coding: utf-8 -*-
"""验证安卓端第二阶段新增的功能（后端部分）。

真实起一个被控端，逐个调用新命令，确认返回合理。
不实际执行关机/关屏/结束进程这类有副作用的操作，只验证读取类和安全路径。
"""
from __future__ import annotations

import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "windows"))

PASSED = 0
FAILED = 0


def check(name: str, ok: bool, detail: str = "") -> None:
    global PASSED, FAILED
    if ok:
        PASSED += 1
        print(f"  [PASS] {name}")
    else:
        FAILED += 1
        print(f"  [FAIL] {name}  {detail}")


def main() -> int:
    from lanremote.actions import Actions, ActionError
    from lanremote.config import Store
    from lanremote.server import RemoteServer

    tmp = Path(tempfile.mkdtemp())

    import tkinter as tk

    class FakeUI:
        # 剪贴板操作用的是真实 Tk 根窗口，所以给一个真的
        root = tk.Tk()
        root.withdraw()

        def ui_call(self, fn, timeout=5.0):
            return fn()

        def screen_size(self):
            return (1920, 1080)

        def notify(self, *a, **k):
            pass

        def log(self, *a, **k):
            pass

    store = Store(tmp)
    store.set("port", 8861)
    actions = Actions(FakeUI(), store)
    server = RemoteServer(store, actions, ui=FakeUI())
    try:
        server.start()
    except OSError as exc:
        print(f"  端口起不来: {exc}")
        return 1
    time.sleep(0.4)

    print("一、读取类新功能")
    battery = actions.battery()
    check("电量能读到", isinstance(battery, dict) and "present" in battery,
          str(battery))
    if battery.get("present"):
        check("电量百分比合理",
              0 <= battery.get("percent", -1) <= 100, str(battery.get("percent")))

    disks = actions.disk()
    check("磁盘能读到", isinstance(disks, list) and len(disks) >= 1, str(disks))
    if disks:
        check("磁盘数值合理",
              all(0 <= d["used_percent"] <= 100 for d in disks), str(disks))

    time.sleep(0.6)
    perf = actions.perf()
    check("性能能读到", bool(perf), str(perf))
    check("CPU 百分比合理", 0 <= perf.get("cpu_percent", -1) <= 100,
          str(perf.get("cpu_percent")))
    check("内存百分比合理", 0 <= perf.get("mem_percent", -1) <= 100,
          str(perf.get("mem_percent")))
    check("运行时长非空", bool(perf.get("uptime")), str(perf.get("uptime")))

    procs = actions.list_processes(10)
    check("进程列表能读到", len(procs) >= 3, f"{len(procs)} 个")
    check("进程信息完整",
          all("name" in p and "pid" in p and "mem_kb" in p for p in procs))

    print("\n二、剪贴板历史")
    actions.clip_set("第一条测试内容")
    actions.clip_set("第二条测试内容")
    actions.clip_set("第一条测试内容")          # 重复的应该被提到最前
    history = actions.clip_history()
    check("历史有记录", len(history) >= 2, f"{len(history)} 条")
    check("最近的在最前", history and history[0]["text"] == "第一条测试内容",
          str(history[:1]))
    check("重复的没有堆叠",
          len([h for h in history if h["text"] == "第一条测试内容"]) == 1,
          str([h["text"] for h in history]))
    check("last_clip 已保存", store.get("last_clip") == "第一条测试内容",
          str(store.get("last_clip")))

    print("\n三、安全边界（这些应该被拒绝）")
    for label, fn in (
        ("拒绝非 http/https 网址",
         lambda: actions.open_url("file:///C:/Windows/System32/calc.exe")),
        ("拒绝空网址", lambda: actions.open_url("")),
        ("拒绝未知快捷程序", lambda: actions.quick_launch("evil")),
        ("拒绝未知文件夹", lambda: actions.open_folder("../../windows")),
        ("拒绝播放音乐目录外的文件",
         lambda: actions.play_local(str(Path(sys.executable)))),
        ("拒绝结束不存在的进程号",
         lambda: actions.kill_process(999999)),
    ):
        try:
            fn()
            check(label, False, "居然没被拒绝")
        except (ActionError, Exception) as exc:            # noqa: BLE001
            check(label, True)

    print("\n四、定时关机的取消路径（不动真格）")
    note = actions.shutdown_timer(0, cancel=True)
    check("取消定时关机能安全返回", isinstance(note, str) and bool(note), note)

    print("\n五、服务端新命令都能通")
    for cmd, key in (
        ("perf", "note"), ("disk", "note"), ("battery", "note"),
        ("clip_history", "note"),
    ):
        reply = server.handle_command_for_test(cmd) if hasattr(
            server, "handle_command_for_test") else None
        if reply is None:
            continue
    check("（通过 Actions 直接验证，服务端分发见协议回归测试）", True)

    server.stop()
    shutil.rmtree(tmp, ignore_errors=True)

    print()
    print("=" * 62)
    print(f"通过 {PASSED} 项，失败 {FAILED} 项")
    return 1 if FAILED else 0


if __name__ == "__main__":
    sys.exit(main())