ykxmz/_build/test_response_headers.py

# -*- coding: utf-8 -*-
r'''加一个"响应头"检查，并把这次的坑记下来。

踩的坑：
    重做管理页时，_send() 的 content_type 默认是 application/json，
    我调它的时候没传 text/html —— 结果浏览器把整页 HTML
    当纯文本显示，用户看到的是一屏源码。

    更糟的是**我当时没发现**：验证脚本只看了 HTTP 200 和字节数，
    两个都对，就以为没问题了。**200 不等于页面能看。**

所以这个检查必须看响应头，不能只看状态码。
顺便也发现 HEAD 请求返回 501（没实现 do_HEAD），一起补了。
'''
import os
from __future__ import annotations

import ast
import json
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
PORT = 8944
PASS = "@（从 RELAY_ADMIN_PASSWORD 环境变量读）"


def wait_port(seconds: float = 12.0) -> bool:
    deadline = time.time() + seconds
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(
                    "http://127.0.0.1:%d/v1/health" % PORT, timeout=2):
                return True
        except Exception:                                 # noqa: BLE001
            time.sleep(0.5)
    return False


def head(path: str, cookie: str = "") -> dict:
    """发一个 HEAD，取回响应头。"""
    req = urllib.request.Request("http://127.0.0.1:%d%s" % (PORT, path),
                                 method="HEAD")
    if cookie:
        req.add_header("Cookie", cookie)
    try:
        with urllib.request.urlopen(req, timeout=8) as resp:
            return {"status": resp.status,
                    "ctype": resp.headers.get("Content-Type", ""),
                    "len": resp.headers.get("Content-Length", "")}
    except urllib.error.HTTPError as exc:
        return {"status": exc.code,
                "ctype": exc.headers.get("Content-Type", ""), "len": ""}
    except Exception as exc:                              # noqa: BLE001
        return {"status": 0, "ctype": "", "len": str(exc)}


def login() -> str:
    """登录拿 cookie。

    用 http.cookiejar 而不是手工读 Set-Cookie ——
    /admin/login 会 302 跳转，urllib 默认跟着跳，
    Set-Cookie 头在第一个响应上就被丢掉了，手工读永远是空。
    cookiejar 会把中间那次的 cookie 存下来。
    """
    import http.cookiejar
    jar = http.cookiejar.CookieJar()
    opener = urllib.request.build_opener(
        urllib.request.HTTPCookieProcessor(jar))
    body = ("username=xmz7891&password=" + PASS).encode()
    req = urllib.request.Request(
        "http://127.0.0.1:%d/admin/login" % PORT, data=body, method="POST")
    req.add_header("Content-Type", "application/x-www-form-urlencoded")
    try:
        opener.open(req, timeout=8).read()
    except Exception:                                     # noqa: BLE001
        pass
    for cookie in jar:
        if cookie.name == "lr_admin":
            return "%s=%s" % (cookie.name, cookie.value)
    return ""


def main() -> int:
    env = dict(__import__("os").environ)
    env["RELAY_PORT"] = str(PORT)
    env["RELAY_ADMIN_PASSWORD"] = PASS
    data = ROOT / "_build" / "_hdr_data"
    data.mkdir(parents=True, exist_ok=True)
    env["RELAY_DATA"] = str(data)

    # 中转把管理密码放在 DATA_DIR/data/admin_password.txt，
    # 没有这个文件就用内置默认值。为了让测试不依赖默认值，
    # 这里显式写一份，保证登录能成。
    auth_dir = data / "data"
    auth_dir.mkdir(parents=True, exist_ok=True)
    (auth_dir / "admin_password.txt").write_text(PASS, encoding="utf-8")

    # /shot/ 路由是从 relay.py 所在目录找 site_shots/。
    # 仓库里那张图在 _build/site_shots/，这里拷一份过去，
    # 否则那条断言会因为"文件不存在"而误报。
    shot_src = ROOT / "_build" / "site_shots" / "pc-main.png"
    shot_dst = ROOT / "relay" / "site_shots" / "pc-main.png"
    if shot_src.is_file():
        shot_dst.parent.mkdir(parents=True, exist_ok=True)
        if not shot_dst.is_file():
            shot_dst.write_bytes(shot_src.read_bytes())

    print("  启动中转（端口 %d）" % PORT)
    proc = subprocess.Popen(
        [sys.executable, "relay.py"], cwd=str(ROOT / "relay"), env=env,
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    passed = failed = 0

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

    try:
        if not wait_port():
            print("  [失败] 中转没起来")
            return 1

        print()
        print("  一、HEAD 请求要能用（原来一律 501）")
        for path in ("/", "/v1/health", "/favicon.ico"):
            info = head(path)
            check("HEAD %s 不是 501" % path, info["status"] != 501,
                  info["status"])

        print()
        print("  二、Content-Type 必须是 text/html（否则浏览器显示源码）")
        site = head("/")
        check("官网 Content-Type 是 text/html",
              "text/html" in site["ctype"], site["ctype"])

        cookie = login()
        check("登录拿到 cookie", bool(cookie), cookie or "（没拿到）")
        admin = head("/zhongzhuanzhan", cookie)
        check("管理页 Content-Type 是 text/html",
              "text/html" in admin["ctype"],
              admin["ctype"] or "（空）")
        check("管理页不是 application/json",
              "json" not in admin["ctype"].lower(), admin["ctype"])

        print()
        print("  二b、HEAD 下载文件要报对长度（不能是 0）")
        # 这条路原来漏了：HEAD 一个 APK 会返回 Content-Length: 0
        # + text/plain，下载器据此认为文件是空的。
        dldir = ROOT / "relay" / "downloads"
        dldir.mkdir(parents=True, exist_ok=True)
        probe = dldir / "probe-1.8.0.apk"
        probe.write_bytes(b"PK\\x03\\x04" + b"x" * 4096)
        info = head("/download/probe-1.8.0.apk")
        check("HEAD 下载返回 200", info["status"] == 200, info["status"])
        check("HEAD 下载 Content-Length 不为 0",
              info["len"] not in ("", "0"), info["len"])
        check("HEAD 下载类型是 android 包",
              "android" in info["ctype"].lower()
              or "octet-stream" in info["ctype"].lower(), info["ctype"])

        print()
        print("  三、其它资源的类型")
        icon = head("/favicon.ico")
        check("favicon 是 image/*", "image" in icon["ctype"], icon["ctype"])
        if shot_dst.is_file():
            shot = head("/shot/pc-main.png")
            check("截图是 image/*", "image" in shot["ctype"], shot["ctype"])
        else:
            print("    [跳过] 截图类型（本地没有 site_shots/pc-main.png）")

    finally:
        proc.terminate()
        try:
            proc.wait(timeout=8)
        except subprocess.TimeoutExpired:
            proc.kill()

    print()
    print("=" * 54)
    print("  通过 %d 项，失败 %d 项" % (passed, failed))
    return 0 if failed == 0 else 1


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