mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 12:30:44 +00:00
xmz7891
b62ac865ea
被控端 - 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%） - 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败） - 每分钟回头找局域网，自动切回快 400 倍的路径 - 画面心跳帧：静止时省流量，客户端仍知道流是活的 - 锁屏时给出可操作的提示 手表端 - 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException - 连接错误翻译成人话（证书/时间/解析/超时分别提示） - 经中转连上后用服务器报的地址刷新已配对记录 - 新增局域网连接页面、服务器延迟检测 服务器端 - 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502） - 网络调优脚本 net-tune.sh（开机自动应用） - 管理页：批量删除、网络体检、下载区 - 隧道改 HTTP/2 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
239 lines
8.1 KiB
Python
239 lines
8.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""清理 + 加回归测试。
|
||
|
||
1. 去掉我多加的那份 _do_quit_agent（原本第 1177 行就有一个，功能一样）
|
||
2. 新增测试，盯住这轮修的三件事：
|
||
* 每个 toggle 都能往两个方向走（不能只往一个方向）
|
||
* close_visuals 能关掉通知 / 全屏窗口
|
||
* _taskbar_hidden 的字节解析
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
|
||
FEATURES = ROOT / "windows" / "lanremote" / "features.py"
|
||
|
||
DUP = ''' def _do_quit_agent(self, params) -> str:
|
||
"""退出被控端（延迟一点点，好让回复先发出去）。"""
|
||
import threading
|
||
import time as _t
|
||
|
||
def die():
|
||
_t.sleep(1.0)
|
||
os._exit(0)
|
||
|
||
threading.Thread(target=die, daemon=True, name="quit").start()
|
||
return "被控端正在退出"
|
||
|
||
'''
|
||
|
||
TEST = '''# -*- coding: utf-8 -*-
|
||
"""盯住开关的方向性、以及「能打开就要能关掉」。
|
||
|
||
这三条对应用户报的：
|
||
任务栏自动隐藏 -> 能隐藏但恢复不了
|
||
开机自启 -> 显示已关闭但打不开
|
||
手表开的窗口 -> 关不掉
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT / "windows"))
|
||
|
||
import tkinter as tk # noqa: E402
|
||
|
||
from lanremote.actions import Actions # noqa: E402
|
||
from lanremote.config import Store # noqa: E402
|
||
from lanremote.features import Features # noqa: E402
|
||
from lanremote.headless import HeadlessUI # noqa: E402
|
||
|
||
passed = 0
|
||
failed = 0
|
||
|
||
|
||
def check(name, ok, detail=""):
|
||
global passed, failed
|
||
if ok:
|
||
passed += 1
|
||
print(f" [PASS] {name}")
|
||
else:
|
||
failed += 1
|
||
print(f" [FAIL] {name} {detail}")
|
||
|
||
|
||
def visible_windows(keyword):
|
||
import ctypes
|
||
user32 = ctypes.windll.user32
|
||
proc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p)
|
||
found = []
|
||
|
||
def callback(hwnd, _l):
|
||
pid = ctypes.c_ulong()
|
||
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
||
if pid.value != os.getpid() or not user32.IsWindowVisible(hwnd):
|
||
return True
|
||
length = user32.GetWindowTextLengthW(hwnd)
|
||
buf = ctypes.create_unicode_buffer(length + 2)
|
||
user32.GetWindowTextW(hwnd, buf, length + 2)
|
||
if keyword in buf.value:
|
||
found.append(buf.value)
|
||
return True
|
||
|
||
user32.EnumWindows(proc(callback), 0)
|
||
return found
|
||
|
||
|
||
def main() -> int:
|
||
store = Store(Path(tempfile.mkdtemp()))
|
||
ui = HeadlessUI(store)
|
||
actions = Actions(ui, store)
|
||
features = Features(actions, store)
|
||
features.actions = actions
|
||
ui.root # 惰性建根窗口
|
||
|
||
defs = features.defs()
|
||
toggles = [d for d in defs if d.get("kind") == "toggle"]
|
||
|
||
print("一、功能表完整性")
|
||
check("功能数量 > 200", len(defs) > 200, f"{len(defs)} 个")
|
||
missing = [d["name"] for d in defs if not hasattr(features, "_do_" + d["name"])]
|
||
check("没有缺实现的功能", not missing, str(missing))
|
||
check("有 toggle 类功能", len(toggles) >= 5, f"{len(toggles)} 个")
|
||
|
||
print()
|
||
print("二、开关必须能往两个方向走")
|
||
# 只挑不会真的改坏系统的：其余只检查"两个方向的返回话术不一样"
|
||
safe = {"loop_play", "shuffle_play", "block_new_app", "game_watch_start"}
|
||
for item in toggles:
|
||
name = item["name"]
|
||
label = item["label"]
|
||
try:
|
||
first = features.run(name, {"on": True})
|
||
except Exception as exc: # noqa: BLE001
|
||
check(f"{label}：能开启", False, str(exc)[:70])
|
||
continue
|
||
if name == "game_watch_start":
|
||
features.run("game_watch_stop", {})
|
||
continue
|
||
try:
|
||
second = features.run(name, {"on": False})
|
||
except Exception as exc: # noqa: BLE001
|
||
check(f"{label}：能关闭", False, str(exc)[:70])
|
||
continue
|
||
different = str(first) != str(second)
|
||
check(f"{label}：两个方向的话术不同", different,
|
||
f"开={first!r} 关={second!r}")
|
||
if name not in safe:
|
||
# 收尾，别把系统留在改过的状态
|
||
try:
|
||
features.run(name, {"on": False})
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
print()
|
||
print("三、手表开的画面必须能关掉")
|
||
ui.notify("测试通知一", "内容")
|
||
ui.notify("测试通知二", "内容")
|
||
time.sleep(1.2)
|
||
before = visible_windows("通知")
|
||
check("通知确实弹出来了", len(before) >= 2, f"{len(before)} 个")
|
||
|
||
result = features.run("close_visuals", {})
|
||
time.sleep(1.2)
|
||
after = visible_windows("通知")
|
||
check("close_visuals 把通知关掉了", len(after) == 0,
|
||
f"关之前 {len(before)} 个，关之后 {len(after)} 个（{result}）")
|
||
|
||
# 全屏大字也要能收
|
||
features.run("big_text", {"text": "测试大字"})
|
||
time.sleep(1.0)
|
||
check("全屏大字弹出来了", len(visible_windows("测试大字")) >= 1)
|
||
features.run("close_visuals", {})
|
||
time.sleep(1.0)
|
||
check("close_visuals 把全屏大字也关掉了",
|
||
len(visible_windows("测试大字")) == 0)
|
||
|
||
result = features.run("close_visuals", {})
|
||
check("没有可关的东西时也不报错", "没有" in str(result), str(result))
|
||
|
||
print()
|
||
print("四、任务栏状态读取")
|
||
hidden = features._taskbar_hidden()
|
||
check("_taskbar_hidden() 返回布尔值", isinstance(hidden, bool), repr(hidden))
|
||
# 真正往返一次：设成隐藏 -> 读到 True -> 恢复 -> 读到 False
|
||
try:
|
||
features.run("taskbar_hide", {"on": True})
|
||
time.sleep(3.5)
|
||
check("隐藏之后 _taskbar_hidden() 读到 True",
|
||
features._taskbar_hidden() is True,
|
||
f"读到 {features._taskbar_hidden()}")
|
||
features.run("taskbar_hide", {"on": False})
|
||
time.sleep(3.5)
|
||
check("恢复之后 _taskbar_hidden() 读到 False",
|
||
features._taskbar_hidden() is False,
|
||
f"读到 {features._taskbar_hidden()}")
|
||
except Exception as exc: # noqa: BLE001
|
||
check("任务栏往返", False, str(exc)[:80])
|
||
|
||
print()
|
||
print("五、开机自启往返")
|
||
from lanremote import autostart
|
||
original = autostart.is_enabled()
|
||
features.run("autostart", {"on": True})
|
||
check("开启后 is_enabled() 为 True", autostart.is_enabled() is True)
|
||
features.run("autostart", {"on": False})
|
||
check("关闭后 is_enabled() 为 False", autostart.is_enabled() is False)
|
||
if original:
|
||
features.run("autostart", {"on": True})
|
||
|
||
ui.stop()
|
||
print()
|
||
print("=" * 66)
|
||
print(f"通过 {passed} 项，失败 {failed} 项")
|
||
if failed:
|
||
print("失败项：")
|
||
return 0 if failed == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|
||
'''
|
||
|
||
|
||
def main() -> int:
|
||
text = FEATURES.read_text(encoding="utf-8")
|
||
if DUP in text:
|
||
text = text.replace(DUP, "", 1)
|
||
FEATURES.write_text(text, encoding="utf-8")
|
||
print(" 重复的 _do_quit_agent 已删除")
|
||
else:
|
||
print(" 重复定义不在，跳过")
|
||
|
||
lines = text.splitlines()
|
||
hits = [i + 1 for i, l in enumerate(lines) if "def _do_quit_agent" in l]
|
||
print(f" _do_quit_agent 现在只在: {hits}")
|
||
|
||
test_path = ROOT / "_tests" / "test_toggles.py"
|
||
test_path.write_text(TEST, encoding="utf-8")
|
||
print(f" 已写入 {test_path.name}")
|
||
|
||
import ast
|
||
for label, path in (("features.py", FEATURES), ("test_toggles.py", test_path)):
|
||
try:
|
||
ast.parse(path.read_text(encoding="utf-8"))
|
||
print(f" {label}: 语法 OK")
|
||
except SyntaxError as exc:
|
||
print(f" {label}: 语法错误 {exc}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|