#!/usr/bin/env python3 """显式使用已确认的串口，回归控制台 HTTP 生命周期；不发送设备命令。 示例：python test_serial_web_console.py --execute --port COM62 --log-dir <目录> 测试独立的本机端口，测试结束释放串口，不创建常驻进程或生产页面保活。 """ import argparse import http.client import json import re import subprocess import sys import threading import time import urllib.error import urllib.parse import urllib.request from pathlib import Path class LivePage: """模拟真正的页面 SSE 连接，持续消费响应；close 模拟页面销毁。""" def __init__(self, port, client, token): self.connection = http.client.HTTPConnection("127.0.0.1", port, timeout=3) query = urllib.parse.urlencode({"client": client, "token": token}) self.connection.request("GET", "/api/client/live?" + query) self.response = self.connection.getresponse() assert self.response.status == 200, self.response.status self.running = True self.frames = 0 self.worker = threading.Thread(target=self.consume, daemon=True) self.worker.start() def consume(self): try: while self.running: if not self.response.readline(): break self.frames += 1 except (OSError, ValueError, http.client.HTTPException): pass def close(self): self.running = False # 主动关闭 socket，不依赖 HTTPResponse.close 等待另一线程的阻塞读取结束。 if self.connection.sock: import socket try: self.connection.sock.shutdown(socket.SHUT_RDWR) except OSError: pass self.connection.close() self.worker.join(2) self.response.close() class ConsoleTest: def __init__(self, args): self.args = args self.base = "http://127.0.0.1:%d" % args.web_port self.process = None self.output = None self.pages = [] self.results = [] def get(self, path): with urllib.request.urlopen(self.base + path, timeout=2) as response: return response.read().decode("utf-8") def status(self): return json.loads(self.get("/api/status")) def record(self, name, **details): record = {"test": name, "passed": True, **details} self.results.append(record) print(json.dumps(record, ensure_ascii=False), flush=True) def start(self, label): self.log = self.args.log_dir / (label + ".log") assert not self.log.exists(), "不能覆盖既有测试日志" self.output = (self.args.log_dir / (label + ".host.log")).open("wb") script = Path(__file__).with_name("serial_web_console.ps1") self.process = subprocess.Popen( ["powershell.exe", "-NoProfile", "-ExecutionPolicy", "RemoteSigned", "-File", str(script), "-Port", self.args.port, "-LogPath", str(self.log), "-WebPort", str(self.args.web_port)], stdout=self.output, stderr=subprocess.STDOUT, creationflags=subprocess.CREATE_NO_WINDOW, ) deadline = time.monotonic() + 10 while time.monotonic() < deadline: assert self.process.poll() is None, "控制台启动失败，查看 host.log" try: status = self.status() assert status["isOpen"] and status["port"] == self.args.port self.token = re.search(r'data-token="([^"]+)"', self.get("/")).group(1) return except urllib.error.URLError: time.sleep(.1) raise AssertionError("控制台启动超时") def page(self, client): page = LivePage(self.args.web_port, client, self.token) self.pages.append(page) return page def disconnect(self, client): query = urllib.parse.urlencode({"client": client}) request = urllib.request.Request( self.base + "/api/client/disconnect?" + query, data=b"", headers={"X-Serial-Token": self.token}, method="POST", ) with urllib.request.urlopen(request, timeout=2) as response: assert response.status == 200 def wait_exit(self, name, since): self.process.wait(timeout=5) elapsed = time.monotonic() - since assert elapsed < 5 and self.process.returncode == 0, elapsed try: self.status() raise AssertionError("进程结束后接口仍可访问") except urllib.error.URLError: pass text = self.log.read_text(encoding="utf-8") assert "串口网页服务结束" in text assert "[TX:" not in text, "测试不应发送任何设备数据" raw = Path(str(self.log) + ".raw.bin") assert raw.exists() self.output.close() self.output = None self.record(name, exit_seconds=round(elapsed, 3), raw_bytes=raw.stat().st_size) def run(self): self.start("lifecycle") time.sleep(5.2) assert self.status()["isOpen"], "从未打开页面时不能自行退出" self.record("no_page_yet") try: self.get("/api/client/live?client=unauthorized") raise AssertionError("缺少令牌仍建立连接") except urllib.error.HTTPError as error: assert error.code == 403 assert self.status()["activeWebClients"] == 0 self.record("liveness_requires_token") first = self.page("page-first") second = self.page("page-second") before = self.status()["receivedBytes"] # 远超原有 3+2 秒时限；只保持页面连接，不发送任何轮询心跳。 time.sleep(11) status = self.status() assert status["isOpen"] and status["liveWebClients"] == 2 assert first.frames > 10 and second.frames > 10 assert status["receivedBytes"] > before, "现场 RX 没有增长，不能宣称采集正常" self.record("paused_polling_stays_alive", received_bytes=status["receivedBytes"]) first.close() time.sleep(5.2) assert self.status()["liveWebClients"] == 1 self.record("closing_one_of_two") second.close() replacement = self.page("page-refreshed") time.sleep(3.2) assert self.status()["isOpen"] and self.status()["liveWebClients"] == 1 self.record("refresh_handoff") reconnect = self.page("page-refreshed") replacement.close() time.sleep(3.2) assert self.status()["liveWebClients"] == 1 assert reconnect.frames > 3 self.record("old_connection_cannot_remove_replacement") start = time.monotonic() self.disconnect("page-refreshed") reconnect.close() self.wait_exit("explicit_last_page_close", start) # 再次打开同一 COM，证明上一次已释放句柄；断开最后连接但不发离页通知。 self.start("abrupt-disconnect") page = self.page("page-abrupt") time.sleep(.3) start = time.monotonic() page.close() self.wait_exit("abrupt_last_page_close", start) self.start("polling-fallback") self.get("/api/status?client=legacy-polling") start = time.monotonic() # 兼容路径总时限约 5 秒（监测线程 250ms 调度允许少量误差）。 self.process.wait(timeout=6) elapsed = time.monotonic() - start assert elapsed < 5.8 and self.process.returncode == 0, elapsed text = self.log.read_text(encoding="utf-8") assert "网页轮询心跳超时" in text and "串口网页服务结束" in text self.record("legacy_polling_fallback", exit_seconds=round(elapsed, 3)) def cleanup(self): for page in self.pages: try: page.close() except (OSError, ValueError): pass if self.process and self.process.poll() is None: self.process.terminate() # 仅终止本测试创建的控制台，不触碰其他串口工具。 self.process.wait(timeout=5) if self.output: self.output.close() (self.args.log_dir / "results.json").write_text( json.dumps(self.results, ensure_ascii=False, indent=2), encoding="utf-8" ) def main(): if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--execute", action="store_true") parser.add_argument("--port", required=True, help="本次用户已确认的 COM 口") parser.add_argument("--web-port", type=int, default=18766) parser.add_argument("--log-dir", required=True, type=Path) args = parser.parse_args() if not args.execute: parser.error("实际打开串口必须显式指定 --execute") args.log_dir.mkdir(parents=True, exist_ok=True) # 端口被其他服务占用时停止，不能把别人的响应误当成本次测试服务。 import socket with socket.socket() as probe: assert probe.connect_ex(("127.0.0.1", args.web_port)) != 0, "测试 HTTP 端口已占用" test = ConsoleTest(args) try: test.run() finally: test.cleanup() if __name__ == "__main__": main()