244 lines
9.2 KiB
Python
244 lines
9.2 KiB
Python
#!/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()
|