Update embedded app debugging workflow
This commit is contained in:
zoukai committed
1 parent
d50be29246
commit
38807e5f97
13 files changed
+846
-122
No files matched your search
@@ -0,0 +1,243 @@
|
||||
#!/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()
|
||||
Reference in new issue
Block a user