ykxmz/_tests/test_admin_netcheck.py

# -*- coding: utf-8 -*-
"""验证管理页的"网络体检"接口 /v1/admin/netcheck。"""
from __future__ import annotations

import http.cookiejar
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
UA = {"User-Agent": "LanRemote/1.4.0"}
PORT = 8140
BASE = f"http://127.0.0.1:{PORT}"

passed = 0
failed = 0


def check(name, ok, detail=""):
    global passed, failed
    if ok:
        passed += 1
        print("  [PASS] " + name)
    else:
        failed += 1
        print("  [FAIL] " + name + "  " + str(detail))


def main() -> int:
    work = Path(tempfile.mkdtemp(prefix="netcheck_"))
    relay = None
    try:
        data = work / "d"
        data.mkdir(parents=True, exist_ok=True)
        env = dict(os.environ)
        env.update({"RELAY_PORT": str(PORT), "RELAY_BIND": "127.0.0.1",
                    "RELAY_DATA": str(data),
                    "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
        relay = subprocess.Popen(
            [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        up = False
        for _ in range(40):
            time.sleep(0.5)
            try:
                with urllib.request.urlopen(urllib.request.Request(
                        BASE + "/v1/health", headers=UA), timeout=2) as resp:
                    if resp.status == 200:
                        up = True
                        break
            except Exception:                             # noqa: BLE001
                continue
        check("本地中转起来了", up)
        if not up:
            return 1

        # 造两台设备，其中一台带 lanIps
        for index, lan in ((1, ["192.168.10.103"]), (2, [])):
            body = json.dumps({
                "deviceId": f"pc-{index}", "code": f"90000{index}",
                "name": f"电脑{index}", "version": "1.4.0",
                "lanIps": lan, "port": 8765,
            }).encode()
            urllib.request.urlopen(urllib.request.Request(
                BASE + "/v1/agent/register", data=body, method="POST",
                headers={**UA, "Content-Type": "application/json"}),
                timeout=10).read()

        print()
        print("一、没登录不能调")
        try:
            code = urllib.request.urlopen(urllib.request.Request(
                BASE + "/v1/admin/netcheck", headers=UA), timeout=8).status
        except urllib.error.HTTPError as exc:
            code = exc.code
        check("未登录被拒（401）", code == 401, code)

        print()
        print("二、登录后能拿到体检信息")
        jar = http.cookiejar.CookieJar()
        opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(jar))
        form = urllib.parse.urlencode({"username": "xmz7891", "password": "@REVAedu120830"}).encode()
        req = urllib.request.Request(BASE + "/admin/login", data=form,
                                     method="POST", headers=UA)
        req.add_header("Content-Type", "application/x-www-form-urlencoded")
        try:
            opener.open(req, timeout=15).read()
        except urllib.error.HTTPError:
            pass

        with opener.open(urllib.request.Request(
                BASE + "/v1/admin/netcheck", headers=UA), timeout=10) as resp:
            info = json.loads(resp.read().decode("utf-8"))
        print(f"       返回: {json.dumps(info, ensure_ascii=False)[:180]}")

        check("ok=true", info.get("ok") is True, info)
        check("有 deviceCount 信息（devices）", "devices" in info, info)
        check("设备数正确（2）", info.get("devices") == 2, info.get("devices"))
        check("有 tunnelMetrics 标记", "tunnelMetrics" in info, info)
        check("有内存信息", "memoryKB" in info, info)
        check("有公网地址字段", "publicUrl" in info, info)
        check("统计了带局域网地址的设备",
              info.get("lanDevices") == 1, info.get("lanDevices"))
        # 本机一般没跑 cloudflared，所以这里应该是 False —— 关键是别崩
        check("读不到隧道指标时不崩（tunnelMetrics=False）",
              info.get("tunnelMetrics") is False, info.get("tunnelMetrics"))
        # 关键：本地没有 cloudflared 时，读不到是正常的，
        # 但**不能是因为代码写错**（比如少 import）。
        # 之前就是漏了 import urllib.request，报 AttributeError，
        # 而读不到时不崩这条断言照样通过 —— 断言得太松了。
        err = str(info.get("tunnelError") or "")
        check("读不到的原因不是代码错误（不能是 AttributeError/NameError）",
              err not in ("AttributeError", "NameError", "TypeError"),
              err)

        print()
        print("三、管理页上有这张卡片")
        html = opener.open(urllib.request.Request(BASE + "/zhongzhuanzhan", headers=UA),
                           timeout=10).read().decode("utf-8")
        check("有网络体检面板", 'id=netPanel' in html, "")
        check("有「测一下」按钮", 'id=netBtn' in html, "")
        check("JS 里调 netcheck 接口", "/v1/admin/netcheck" in html, "")
        check("会显示隧道连接数", "tunnelConnections" in html, "")

        print()
        print("=" * 58)
        print(f"通过 {passed} 项，失败 {failed} 项")
        return 0 if failed == 0 else 1
    finally:
        if relay is not None:
            try:
                relay.terminate()
                relay.wait(timeout=8)
            except Exception:                             # noqa: BLE001
                try:
                    relay.kill()
                except Exception:                     # noqa: BLE001
                    pass
        time.sleep(0.5)
        shutil.rmtree(work, ignore_errors=True)


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