ykxmz/_tests/acceptance.py

# -*- coding: utf-8 -*-
"""目标验收：逐条核对用户提出的每一项要求，每条都给证据。

用户的目标：
    "在 Debian 随身WiFi 上部署中转服务器 + Cloudflare Tunnel，实现跨网络远程控制。
     交付物为压缩包，内含三部分：被控端(Windows agent)、服务器端(relay)、控制端(Android APK)。
     要求：单设备单配对码（设备曾设置过则复用旧码）、服务器端可管理、
           面向 512MB 内存做优化。另需为电脑端所有按钮增加回退/撤销能力。"

这个脚本**不重复跑那些已经跑过的测试**，而是逐条给"这条要求达成了"的证据，
并明确指出哪一条还没有完全达成。
"""
from __future__ import annotations

import json
import subprocess
import sys
import urllib.error
import urllib.request
import zipfile
from pathlib import Path

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

from devssh import run as ssh                            # noqa: E402

RESULTS = []


# ---------------------------------------------------------------------------
# 管理页密码**不再硬编码**。
#
# 以前这里写着明文密码，而 relay.py 里也把它当默认密码 ——
# 仓库要推到公开的 Gitee/EzHub，等于把管理台钥匙挂门上，
# 而且 git 历史删不掉。
#
# 现在按优先级取：
#     1. 环境变量 RELAY_ADMIN_PASSWORD
#     2. _build/relay_info.json 里的 adminPassword（本地测试用）
#     3. 本机 8080 中转实际用的 data/admin_password.txt
# 拿不到就跳过需要登录的用例，而不是编一个假密码硬试。
# ---------------------------------------------------------------------------
def _admin_password() -> str:
    import json as _json
    import os as _os
    from pathlib import Path as _Path

    pw = (_os.environ.get("RELAY_ADMIN_PASSWORD") or "").strip()
    if pw:
        return pw
    root = _Path(__file__).resolve().parent.parent
    info = root / "_build" / "relay_info.json"
    try:
        data = _json.loads(info.read_text(encoding="utf-8"))
        pw = str(data.get("adminPassword") or "").strip()
        if pw:
            return pw
    except Exception:
        pass
    for cand in (root / "relay" / "data" / "admin_password.txt",
                 _Path("/root/lanremote-relay/data/admin_password.txt")):
        try:
            if cand.is_file():
                return cand.read_text(encoding="utf-8").strip()
        except Exception:
            continue
    return ""



def item(requirement: str, ok: bool, evidence: str, gap: str = "") -> None:
    RESULTS.append({
        "requirement": requirement,
        "ok": ok,
        "evidence": evidence,
        "gap": gap,
    })
    mark = "✓" if ok else ("~" if gap else "✗")
    print(f"  [{mark}] {requirement}")
    print(f"      证据: {evidence}")
    if gap:
        print(f"      差距: {gap}")


def device(command: str, timeout: int = 60) -> str:
    _code, out, _err = ssh(command, timeout=timeout, quiet=True)
    return out.strip()


# Cloudflare 的 Browser Integrity Check 会拦 Python 默认 UA
# （403 + error code 1010），所以请求要带上这个。
_UA = {"User-Agent": "LanRemote/1.4.0"}


def http_get(url: str, timeout: int = 15) -> tuple:
    try:
        req = urllib.request.Request(url, headers=_UA)
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status, resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read().decode("utf-8", "replace")
    except Exception as exc:                              # noqa: BLE001
        return 0, str(exc)


def main() -> int:
    print("=" * 70)
    print("  目标验收：逐条核对")
    print("=" * 70)
    print()

    # ---------------- 1. 中转服务器部署 ----------------
    print("一、在随身 WiFi 上部署中转服务器")
    status = device("systemctl is-active lanremote-relay")
    enabled = device("systemctl is-enabled lanremote-relay 2>/dev/null")
    health = device("curl -s --max-time 8 http://127.0.0.1:8080/v1/health")
    mem_kb = 0
    try:
        mem_kb = json.loads(health).get("memoryKB", 0)
    except Exception:                                     # noqa: BLE001
        pass
    item("中转服务在设备上运行", status == "active",
         f"systemctl={status}，开机自启={enabled}，Rss={mem_kb}KB ({mem_kb // 1024}MB)")

    # ---------------- 2. Cloudflare Tunnel ----------------
    print()
    print("二、Cloudflare 隧道")
    tunnel_active = device("systemctl is-active lanremote-tunnel")
    url = device("cat /root/lanremote-relay/public_url.txt 2>/dev/null")
    code, _body = http_get(f"{url}/v1/health") if url else (0, "")
    item("Cloudflare 隧道已建立且公网可达", tunnel_active == "active" and code == 200,
         f"服务={tunnel_active}，地址={url or '（无）'}，从公网访问 HTTP {code}")
    cert = device("test -f /root/.cloudflared/cert.pem && echo yes || echo no")
    item("具名隧道（固定域名 relay.xxmxxs.top）",
         cert == "yes",
         f"cert.pem={'存在' if cert == 'yes' else '不存在'}（授权未完成）",
         gap="等待用户在浏览器完成 Cloudflare 授权。"
             "已定位并修复了阻塞它的两个问题："
             "① api.cloudflare.com 被 DNS 劫持（已写 hosts 钉真实 IP）；"
             "② install.sh 的 CRLF 行尾（已改 LF）")

    # ---------------- 3. 三件套压缩包 ----------------
    print()
    print("三、交付压缩包含三部分")
    zip_path = ROOT / "dist" / "LanRemote-跨网络版-1.4.0.zip"
    parts = {}
    if zip_path.is_file():
        with zipfile.ZipFile(zip_path) as archive:
            names = archive.namelist()
        parts = {
            "被控端": any("ykxmz.exe" in n for n in names),
            "服务器端": any("relay.py" in n for n in names),
            "控制端": any(".apk" in n for n in names),
        }
    size = zip_path.stat().st_size / 1024 / 1024 if zip_path.is_file() else 0
    item("压缩包含被控端 / 服务器端 / 控制端",
         all(parts.values()),
         f"{zip_path.name}  {size:.2f} MB  " +
         "  ".join(f"{k}={'✓' if v else '✗'}" for k, v in parts.items()))

    # ---------------- 4. 单设备单配对码 ----------------
    print()
    print("四、单设备单配对码（设备设过就复用旧码）")
    # 用真实的 relay 跑一次：同 deviceId 报新码 -> 应改回旧码；异设备用同码 -> 应拒绝
    result = subprocess.run(
        [sys.executable, "-u", str(ROOT / "_tests" / "test_relay_local.py")],
        capture_output=True, text=True, encoding="utf-8", errors="replace")
    relay_ok = "失败 0 项" in result.stdout
    item("单设备单配对码 + 一个码只能一台设备用", relay_ok,
         "test_relay_local.py（经真实中转）：" +
         (result.stdout.strip().splitlines()[-1] if result.stdout.strip() else "无输出"))

    # ---------------- 5. 服务器端可管理 ----------------
    print()
    print("五、服务器端可管理")
    if url:
        plain = device("cat /root/lanremote-relay/data/admin_password.txt 2>/dev/null")
        c_login, _ = http_get(f"{url}/v1/admin/devices")
        c_url, url_body = http_get(f"{url}/url")
        item("管理页需要登录 + 管理接口 + 可查当前地址",
             c_login == 401 and c_url == 200,
             f"管理接口未登录={c_login}（应为 401）；"
             f"/url={c_url} 返回 {url_body.strip()[:60]}；"
             f"管理密码={plain or '@_admin_password()'}")
    else:
        item("管理页需要登录 + 管理接口", False, "没有公网地址，无法测")

    # ---------------- 6. 512MB 内存优化 ----------------
    print()
    print("六、面向 512MB 内存做优化")
    freem = device("free -m | awk 'NR==2{print $3\" \"$7}'")
    queue_cap = device("grep -o 'RELAY_QUEUE_BYTES=[0-9]*' "
                       "/etc/systemd/system/lanremote-relay.service")
    memmax = device("grep -o 'MemoryMax=[0-9A-Za-z]*' "
                    "/etc/systemd/system/lanremote-relay.service")
    item("内存占用受控且有上限保护",
         mem_kb > 0 and mem_kb < 60 * 1024,
         f"中转 Rss={mem_kb // 1024}MB；系统已用/可用={freem or '?'}MB；"
         f"{memmax or 'MemoryMax 未设'}；{queue_cap or '队列上限未设'}")

    # ---------------- 7. 所有按钮加回退 ----------------
    print()
    print("七、电脑端所有按钮加回退/撤销")
    result = subprocess.run(
        [sys.executable, "-u", str(ROOT / "_tests" / "test_undo.py")],
        capture_output=True, text=True, encoding="utf-8", errors="replace")
    undo_ok = "失败 0 项" in result.stdout
    item("开关类 + 画面类都能回退，不可逆的不进列表", undo_ok,
         "test_undo.py：" +
         (result.stdout.strip().splitlines()[-1] if result.stdout.strip() else "无输出"))

    # ---------------- 8. 跨网络远程控制（端到端）----------------
    print()
    print("八、跨网络远程控制确实能用")
    if url:
        admin = device("cat /root/lanremote-relay/data/admin_key.txt 2>/dev/null")
        result = subprocess.run(
            [sys.executable, "-u",
             str(ROOT / "_tests" / "test_relay_e2e.py"), url, admin],
            capture_output=True, text=True, encoding="utf-8", errors="replace")
        e2e_ok = "失败 0 项" in result.stdout
        item("经公网隧道完成注册/接入/收发/鉴权", e2e_ok,
             "test_relay_e2e.py：" +
             (result.stdout.strip().splitlines()[-1] if result.stdout.strip() else "无输出"))
    else:
        item("经公网隧道端到端", False, "没有公网地址")

    # ---------------- 汇总 ----------------
    print()
    print("=" * 70)
    print("  验收汇总")
    print("=" * 70)
    full = sum(1 for r in RESULTS if r["ok"])
    partial = sum(1 for r in RESULTS if not r["ok"] and r["gap"])
    missing = sum(1 for r in RESULTS if not r["ok"] and not r["gap"])
    for r in RESULTS:
        mark = "✓ 达成" if r["ok"] else ("~ 部分" if r["gap"] else "✗ 未达成")
        print(f"  {mark}  {r['requirement']}")
    print()
    print(f"  完全达成 {full} 项 / 部分达成 {partial} 项 / 未达成 {missing} 项")
    if partial:
        print()
        print("  未完全达成的：")
        for r in RESULTS:
            if not r["ok"] and r["gap"]:
                print(f"    · {r['requirement']}")
                print(f"      {r['gap']}")
    return 0 if missing == 0 else 1


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