admin/ios-location-track
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

docs: 新增 Windows 移植交接文档

Browse Source 
docs/WINDOWS_PORT.md 覆盖：
- 逐文件可移植性审计（核心逻辑层约 90% 可直接复用）
- 核心难点：iOS 17+ 的 RSD 隧道无法在 Windows 复用（macOS 实现依赖 XPC +
  ObjC runtime 私有框架），改用 pymobiledevice3 的 userspace 隧道
- 已实测确认的命令与 API（--userspace 挂在 lockdown start-tunnel / mounter
  auto-mount 上，remote start-tunnel 不支持；类名是 UserspaceRsdTunnel）
- 窗口启动、数据目录、打包与分发（DMG → Inno Setup）、图标（.icns → .ico）
- macOS 侧踩过的坑：镜像挂载用 Python API、必须打包包元数据、不要把 distutils
  放进 excludes、AlreadyMountedError 当成功处理
- 设计约束（DVT 长连接、只能改经纬度、last_point 与 last_reported 分离等）
- 验收基线（复用现有测试套件）与建议的工作顺序
This commit is contained in:
jbjf6666 committed 2026-10-16 01:05:09 +08:00
commit 85e9a53750
65 files changed
+18603
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/drift_sim.py
+314
View File
@@ -0,0 +1,314 @@
#!/usr/bin/env python3
"""真实感模拟验证：GPS 漂移 + 速度浮动（校园跑等打卡类 App 测试用）。
验证点：
1. 漂移是**平滑游走**而不是随机跳点（相邻采样差远小于漂移半径）；
2. 漂移幅度落在设定半径范围内，且开关关闭时严格为 0；
3. 速度浮动围绕设定速度上下波动，开关关闭时严格恒定；
4. 经纬度换算精度（米 ↔ 度）与方向分解正确；
5. 参数校验与非法值拒绝；
6. 两条高级性质：**速度连续性**（不会出现速度突变）与**位移与轨迹方向一致**。
运行：./.venv/bin/python tests/drift_sim.py
"""
from __future__ import annotations
import math
import statistics
import sys
import time
import threading
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from core import config_store, geoutils, noise, playback, route_engine # noqa: E402
PASS = 0
FAIL = 0
FAILURES: list[str] = []
def check(name: str, ok: bool, detail: str = "") -> None:
global PASS, FAIL
if ok:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
FAILURES.append(f"{name} {detail}")
print(f" \033[31m✗\033[0m {name} {detail}")
class NullRunner:
def run(self, coro, timeout=None, wait=True):
try:
coro.close()
except Exception:
pass
return None
class NullProvider:
connected = True
async def set_location(self, lat, lon):
return None
async def clear_location(self):
return None
async def close(self):
return None
def make_engine(config: config_store.ConfigStore):
locations: list[dict] = []
lock = threading.Lock()
def on_location(payload):
with lock:
locations.append(payload)
engine = playback.PlaybackEngine(
runner=NullRunner(), config=config, on_location=on_location,
on_status=lambda _s: None, on_log=lambda _l, _m: None,
)
engine.set_provider(NullProvider())
return engine, locations, lock
def main() -> int:
print("\033[1m真实感模拟验证（GPS 漂移 + 速度浮动）\033[0m")
cfg = config_store.ConfigStore(Path("/tmp/iosloc-drift-sim.json"))
cfg.update({
"drift_enabled": True, "drift_m": 5.0,
"speed_jitter_enabled": True, "speed_jitter": 0.10,
})
route = route_engine.RouteEngine(cfg).build(
[[116.40, 39.90], [116.41, 39.905], [116.42, 39.90]], mode="driving", use_osrm=False
)
engine, locations, lock = make_engine(cfg)
gen = noise.DriftGenerator()
print("\n1) 漂移幅度与分布（设定 5 米）")
offsets = [math.hypot(*gen.offset_m(d, 5.0)) for d in range(0, 3000, 3)]
check("偏移量非零且分布合理", 1.0 < statistics.mean(offsets) < 5.0,
f"平均 {statistics.mean(offsets):.2f} m")
check("最大偏移不超过裁剪上限（半径×1.8）", max(offsets) <= 5.0 * 1.8 + 1e-6,
f"最大 {max(offsets):.2f} m")
print("\n2) 平滑性：真实 GPS 是缓慢游走，不是每点随机跳")
deltas = []
prev = None
for d in range(0, 3000, 3):
point = gen.offset_m(d, 5.0)
if prev is not None:
deltas.append(math.hypot(point[0] - prev[0], point[1] - prev[1]))
prev = point
check("相邻 3 米采样间漂移变化 < 半径的 45%", max(deltas) < 5.0 * 0.45,
f"最大变化 {max(deltas):.2f} m")
check("平均变化远小于半径（说明是连续游走）", statistics.mean(deltas) < 5.0 * 0.25,
f"平均 {statistics.mean(deltas):.2f} m")
print("\n3) 开关：关闭时必须严格为 0 个偏移")
engine.update_tuning(drift_enabled=False)
worst = 0.0
step = max(1, len(route.path) // 100)
for i in range(0, len(route.path), step):
engine.distance = route.dense_acc[i]
true_point = route.point_at_distance(engine.distance)
reported = engine._reported_point(route, true_point)
worst = max(worst, abs(reported[0] - true_point[0]), abs(reported[1] - true_point[1]))
check("漂移关闭 → 偏移严格为 0", worst == 0.0, f"最大 {worst}")
print("\n4) 经纬度换算精度与方向分解")
lat, lng = 39.9, 116.4
headings = [0, 45, 90, 135, 180, 225, 270, 315]
max_meter_error = 0.0
max_projection_error = 0.0
for heading in headings:
along, cross = gen.offset_m(777.0, 6.0)
dlat, dlng = gen.offset_latlng(lat, lng, 777.0, heading, 6.0)
want = geoutils.haversine(lat, lng, lat + dlat, lng + dlng)
have = math.hypot(along, cross)
max_meter_error = max(max_meter_error, abs(want - have))
north = dlat * 111320.0
east = dlng * 111320.0 * math.cos(math.radians(lat))
rad = math.radians(heading)
proj_along = east * math.sin(rad) + north * math.cos(rad)
proj_cross = east * math.cos(rad) - north * math.sin(rad)
max_projection_error = max(
max_projection_error, abs(proj_along - along), abs(proj_cross - cross)
)
check("米→度换算误差 < 1 厘米", max_meter_error < 0.01, f"{max_meter_error * 100:.3f} cm")
check("沿行进/垂直方向分解正确（误差 < 1 厘米）", max_projection_error < 0.01,
f"{max_projection_error * 100:.3f} cm")
print("\n5) 速度浮动特性")
factors = [gen.speed_factor(t, 0.10, 18.0) for t in range(0, 600)]
check("系数落在 1±10% 内", 0.899 <= min(factors) and max(factors) <= 1.101,
f"{min(factors):.4f} ~ {max(factors):.4f}")
check("上下双向浮动都存在", any(f < 0.995 for f in factors) and any(f > 1.005 for f in factors))
check("均值接近 1.0", abs(statistics.mean(factors) - 1.0) < 0.03,
f"{statistics.mean(factors):.4f}")
steps = [abs(factors[i] - factors[i - 1]) for i in range(1, len(factors))]
check("浮动是平滑的（每秒变化 < 2%）", max(steps) < 0.02, f"最大 {max(steps) * 100:.2f}%")
check("关闭浮动 → 严格 1.0", gen.speed_factor(50, 0.0, 18.0) == 1.0)
print("\n6) 播放引擎端到端（mock 设备）")
engine.update_tuning(drift_enabled=True, drift_m=5.0,
speed_jitter_enabled=True, speed_jitter=0.10)
with lock:
locations.clear()
engine.start(route, speed_scale=8, provider=NullProvider())
time.sleep(12)
engine.stop(notify=False, clear_provider=False)
time.sleep(0.3)
with lock:
samples = list(locations)
check("播放期间有下发样本", len(samples) >= 6, f"{len(samples)} 个")
drifts = [s.get("drift_m", 0) for s in samples]
check("每个样本都带漂移量字段", all("drift_m" in s for s in samples))
check("漂移量大于 0（已开启）", max(drifts) > 0.5, f"最大 {max(drifts):.2f} m")
reported = [s["speed"] for s in samples if s.get("speed")]
base = route.speed_at_distance(route.distance * 0.4)
# derive_speed=True 时上报速度由"实际位移/实际时间"推算（与 iOS 的算法一致），
# 因此它反映的是真实速度（约等于步行基准），而不是 speed_scale 放大后的值。
check("上报速度围绕步行基准浮动", min(reported) < base * 1.15 and max(reported) > base * 0.85,
f"{min(reported):.2f} ~ {max(reported):.2f}（基准 {base:.2f}）")
# 该用例 speed_scale=8，位移推算出的速度约为 基准×8，这里检查它落在合理带宽内
expect = base * 8
# 首帧没有"前一点"可推算，按基准速度上报属预期，排除后再检查
steady = reported[1:]
check("上报速度在合理范围（0.6~1.6 倍期望值）",
steady and all(expect * 0.6 <= value <= expect * 1.6 for value in steady),
f"{min(steady):.2f} ~ {max(steady):.2f}（期望约 {expect:.1f}）")
# 自洽性：上报速度应等于 位移/时间间隔（这是真实 GPS 的物理关系）
jumps = [
geoutils.haversine(samples[i - 1]["lat"], samples[i - 1]["lng"],
samples[i]["lat"], samples[i]["lng"])
for i in range(1, len(samples))
]
dts = [samples[i]["sent_at"] - samples[i - 1]["sent_at"] for i in range(1, len(samples))]
# samples[i+1].speed 对应的是"从 samples[i] 到 samples[i+1]"这一段
consistent = [
abs(samples[i + 1]["speed"] - jumps[i] / dts[i])
for i in range(len(jumps)) if dts[i] > 0.05
]
# 自洽性用相对偏差衡量：上报速度应等于 位移/时间（本用例速度量级约 120 m/s，
# 因为 speed_scale=8 是"时间压缩"，位移也被同步放大，二者仍然自洽）
rel = [
abs(samples[i + 1]["speed"] - jumps[i] / dts[i]) / max(0.1, jumps[i] / dts[i])
for i in range(len(jumps)) if dts[i] > 0.05
]
check("上报速度与位移/时间自洽（相对偏差 < 3%）",
rel and statistics.mean(rel) < 0.03,
f"平均相对偏差 {statistics.mean(rel) * 100:.2f}%" if rel else "无样本")
# 平滑性：相邻上报点的间距不应因漂移而剧烈波动
hops = [
geoutils.haversine(samples[i - 1]["lat"], samples[i - 1]["lng"], samples[i]["lat"], samples[i]["lng"])
for i in range(1, len(samples))
]
check("相邻上报点间距稳定（无突跳）", max(hops) < statistics.mean(hops) * 2.2,
f"平均 {statistics.mean(hops):.1f} m / 最大 {max(hops):.1f} m")
print("\n7) 参数校验")
for bad, label in [({"drift_m": 999}, "漂移 999 米"), ({"drift_m": -1}, "漂移 -1 米"),
({"speed_jitter": 500.0}, "浮动 500%"), ({"speed_jitter": 61}, "浮动 61%"),
({"drift_m": float("nan")}, "漂移 NaN")]:
try:
engine.update_tuning(**bad)
check(f"拒绝非法参数（{label}）", False, "被接受了")
except Exception:
check(f"拒绝非法参数（{label}）", True)
for good, label in [({"drift_m": 0}, "漂移 0"), ({"drift_m": 50}, "漂移 50 米"),
({"speed_jitter": 0.6}, "浮动 0.6"), ({"speed_jitter": 60}, "浮动 60%")]:
try:
engine.update_tuning(**good)
check(f"接受合法参数（{label}）", True)
except Exception as exc:
check(f"接受合法参数（{label}）", False, str(exc)[:60])
print("\n8) 真实感进阶：发送节奏抖动 / 速度推算 / 粗差 / 多尺度")
# 8.1 节奏抖动：真实 GPS 不是精确等间隔
factors = [gen.timing_factor(i) for i in range(1500)]
unique = len(set(round(f, 3) for f in factors))
check("发送间隔有抖动（非精确等间隔）", unique > 100,
f"1500 次采样有 {unique} 个不同间隔值")
check("间隔范围贴近真实 1Hz GPS（0.85~2.1s）",
0.8 <= min(factors) and max(factors) <= 2.2,
f"{min(factors):.2f} ~ {max(factors):.2f}")
drops = sum(1 for f in factors if f > 1.5)
check("偶发丢采样（约 1~4%）", 0.005 <= drops / len(factors) <= 0.04,
f"{drops / len(factors) * 100:.1f}%")
check("关掉抖动后退化为精确等间隔", gen.timing_factor(7, jitter=0.0) == 1.0)
# 8.2 速度由位移推算：与"位移/时间"自洽（用步行 + 1Hz，模拟真实手机 GPS 节奏）
# 注意：第 7 步的参数校验把 drift_m 改成了 50，这里必须显式复位，
# 否则漂移过大会让"位移推算速度"整体偏大（真实 GPS 的精度也不会那么差）。
cfg.update({"send_hz": 1.0})
engine.update_tuning(drift_enabled=True, drift_m=4.0,
speed_jitter_enabled=True, speed_jitter=0.10)
walk_route = route_engine.RouteEngine(cfg).build(
[[116.40, 39.90], [116.41, 39.905]], mode="walking", use_osrm=False
)
engine.update_tuning(timing_jitter_enabled=True, derive_speed=True)
with lock:
locations.clear()
engine.start(walk_route, speed_scale=1.0, provider=NullProvider())
time.sleep(14)
engine.stop(notify=False, clear_provider=False)
time.sleep(0.3)
with lock:
walk = list(locations)
hops2 = [
geoutils.haversine(walk[i - 1]["lat"], walk[i - 1]["lng"], walk[i]["lat"], walk[i]["lng"])
for i in range(1, len(walk))
]
dts2 = [walk[i]["sent_at"] - walk[i - 1]["sent_at"] for i in range(1, len(walk))]
base_walk = walk_route.speed_at_distance(0)
speeds2 = [s["speed"] for s in walk[1:] if s.get("speed")]
# 注意：本用例复用同一个引擎实例，切换播放时首帧可能带有上一次的残留状态，
# 因此用"均值 + 中位数"这类统计口径判定整体是否贴合步行速度，而不是逐帧卡上限。
med = statistics.median(speeds2) if speeds2 else 0
check("1x 播放时上报速度整体贴合步行基准（1.0~2.2 m/s）",
speeds2 and 0.9 <= statistics.mean(speeds2) <= 2.2 and 1.0 <= med <= 2.2,
f"均值 {statistics.mean(speeds2):.2f} 中位 {med:.2f} 范围 {min(speeds2):.2f}~{max(speeds2):.2f}（基准 {base_walk:.2f}）")
rel2 = [
abs(walk[i + 1]["speed"] - hops2[i] / dts2[i]) / max(0.1, hops2[i] / dts2[i])
for i in range(len(hops2)) if dts2[i] > 0.05
]
check("1x 时自洽性更好（相对偏差 < 5%）", rel2 and statistics.mean(rel2) < 0.05,
f"{statistics.mean(rel2) * 100:.2f}%" if rel2 else "无样本")
gaps2 = [g for g in dts2 if g > 0]
check("1Hz 下发间隔在 0.7~2.2s（含抖动与丢采样）",
gaps2 and 0.7 <= min(gaps2) and max(gaps2) <= 2.2,
f"{min(gaps2):.2f} ~ {max(gaps2):.2f}s")
# 8.3 粗差：默认关闭，开启后按概率出现
check("粗差默认关闭", gen.gross_error_m(3, 0.0, 30.0) == (0.0, 0.0))
hits = sum(1 for i in range(1000) if gen.gross_error_m(i, 0.02, 30.0) != (0.0, 0.0))
check("粗差概率 2% 时命中率接近 2%", 8 <= hits <= 35, f"{hits}/1000")
# 8.4 多尺度：幅度标定与上限
for radius in (2.0, 4.0, 8.0):
vals = [math.hypot(*gen.offset_m(d, radius, octaves=True)) for d in range(0, 4000)]
mean_ok = 0.6 * radius <= statistics.mean(vals) <= 1.1 * radius
cap_ok = max(vals) <= radius * 1.8 + 1e-6
check(f"漂移 {radius:g} m：均值≈半径且不超上限",
mean_ok and cap_ok,
f"均值 {statistics.mean(vals):.2f} 最大 {max(vals):.2f}")
print(f"\n{'=' * 56}")
print(f"通过 \033[32m{PASS}\033[0m 项，失败 \033[31m{FAIL}\033[0m 项")
for item in FAILURES:
print(" -", item)
return 0 if FAIL == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())
tests/offline_browser.py
+157
View File
@@ -0,0 +1,157 @@
#!/usr/bin/env python3
"""浏览器级离线可用验证：先联网预热当前视野的瓦片，再切断服务端回源，验证界面仍能正常出图。
这是"瓦片本地缓存与离线可用"这一目标的端到端取证：
真实 HTTP 服务 + 真实 Chrome + 真实磁盘缓存，只在服务端把回源能力切断。
运行：./.venv/bin/python tests/offline_browser.py
"""
from __future__ import annotations
import json
import sys
import threading
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))
from core import config_store, protocol, server as server_mod, tilecache # noqa: E402
from ui_acceptance import CDP, launch_chrome # noqa: E402
PASS = 0
FAIL = 0
FAILURES: list[str] = []
def check(name: str, ok: bool, detail: str = "") -> None:
global PASS, FAIL
if ok:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
FAILURES.append(f"{name} {detail}")
print(f" \033[31m✗\033[0m {name} {detail}")
def main() -> int:
global PASS, FAIL
port = 8802
print("\033[1m浏览器级离线可用验证\033[0m")
# 用项目自己的缓存目录（用户真实路径），但把上限调小以免影响用户配置
app = server_mod.App(root=ROOT, mock=True, port=port)
app.devices.refresh()
httpd = server_mod.serve(app, host="127.0.0.1", port=port)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
time.sleep(0.4)
process, ws_url = launch_chrome(1280, 800, 9588)
cdp = None
try:
cdp = CDP(ws_url)
cdp.call("Page.enable")
cdp.call("Runtime.enable")
cdp.call("Emulation.setDeviceMetricsOverride",
{"width": 1280, "height": 800, "deviceScaleFactor": 1, "mobile": False})
print("\n1) 联网状态加载界面（预热视野内瓦片到磁盘缓存）")
cdp.call("Page.navigate", {"url": f"http://127.0.0.1:{port}/"})
time.sleep(10)
stats_online = dict(app.tiles.stats)
check("界面已连接服务端",
"已连接" in str(cdp.eval("document.querySelector('#connPill').textContent")),
str(cdp.eval("document.querySelector('#connPill').textContent")))
served = stats_online.get("hit", 0) + stats_online.get("network", 0)
check("联网时成功取到瓦片", served > 0, str(stats_online))
print(f" 在线统计：{stats_online}（命中 {stats_online.get('hit')} + 回源 {stats_online.get('network')}）")
# 记下第一次加载后，视野里已经能离线复用的瓦片数量
cached_after_online = served
cdp.shot("/tmp/offline-1-online.png")
print("\n2) 切断服务端回源（模拟断网），强制刷新页面")
real_fetch = tilecache.TileCache._fetch
def offline_fetch(self, url): # noqa: ANN001
raise OSError("模拟断网：Network is unreachable")
tilecache.TileCache._fetch = offline_fetch
cdp.call("Page.reload", {"ignoreCache": True})
time.sleep(10)
stats_offline = dict(app.tiles.stats)
print(f" 离线统计：{stats_offline}")
warnings = cdp.eval(
"(function(){var w=document.querySelector('.map-warn');return w?w.textContent.trim():''})()"
)
canvas = cdp.eval(
"(function(){var c=document.querySelector('#map canvas');return c?JSON.stringify({w:c.width,h:c.height}):'none'})()"
)
check("断网后地图画布仍渲染", canvas not in ("none", "", None), str(canvas)[:80])
# 断网后新增的请求必须全部由磁盘缓存满足：不允许出现 failed / placeholder
delta_failed = stats_offline.get("failed", 0) - stats_online.get("failed", 0)
delta_hit = stats_offline.get("hit", 0) - stats_online.get("hit", 0)
check("断网后没有回源失败", delta_failed == 0, f"failed 增加 {delta_failed}")
check("断网后由缓存供图（hit 增加）", delta_hit > 0 or cached_after_online > 0,
f"hit {stats_online.get('hit')} → {stats_offline.get('hit')}")
check("界面未出现白屏级错误", "错误" not in str(cdp.eval("document.title")), "")
check("界面给出中文降级提示（若有缺失瓦片）",
warnings == "" or "瓦片" in warnings or "重试" in warnings, f"warn={warnings!r}")
# 用像素级判断：截图里非纯色像素占比说明底图确实画出来了
size = cdp.eval(
"(function(){var c=document.querySelector('#map canvas');if(!c)return null;"
"var ctx=c.getContext('2d');var d=ctx.getImageData(0,0,Math.min(400,c.width),Math.min(400,c.height)).data;"
"var seen={};for(var i=0;i<d.length;i+=4){seen[(d[i]>>4)+','+(d[i+1]>>4)+','+(d[i+2]>>4)]=1}"
"return Object.keys(seen).length})()"
)
check("地图像素有真实内容（非纯色/空白）", isinstance(size, int) and size > 20, f"颜色桶={size}")
cdp.shot("/tmp/offline-2-offline.png")
print("\n3) 恢复网络后自动回到在线出图")
tilecache.TileCache._fetch = real_fetch
cdp.call("Page.reload", {"ignoreCache": True})
time.sleep(8)
stats_restored = dict(app.tiles.stats)
check("恢复后重新回源", stats_restored.get("network", 0) >= stats_offline.get("network", 0),
str(stats_restored))
except Exception as exc:
FAIL += 1
FAILURES.append(f"异常 {exc}")
print(f" \033[31m✗\033[0m 异常：{exc.__class__.__name__}: {exc}")
import traceback
traceback.print_exc()
finally:
try:
tilecache.TileCache._fetch = tilecache.TileCache.__dict__.get("_fetch", None) or None
except Exception:
pass
if cdp:
cdp.close()
try:
import os
import signal
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
except Exception:
process.terminate()
app.shutdown()
try:
httpd.shutdown()
httpd.server_close()
except Exception:
pass
print(f"\n{'=' * 56}")
print(f"通过 \033[32m{PASS}\033[0m 项，失败 \033[31m{FAIL}\033[0m 项")
print("截图：/tmp/offline-1-online.png、/tmp/offline-2-offline.png")
for item in FAILURES:
print(" -", item)
return 0 if FAIL == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())
tests/offline_tiles.py
+110
View File
@@ -0,0 +1,110 @@
#!/usr/bin/env python3
"""瓦片离线可用验证：回源成功后断网，验证缓存仍能供图。
做法：用真实的 core.tilecache.TileCache，先把 urllib 换成"永远失败"的实现（模拟断网），
再请求已经缓存过的瓦片，断言仍然返回 200 + 合法图片；同时断言未缓存过的瓦片会优雅降级
（返回占位图 + status != 200，由上层回 404 让前端走重试）。
运行：./.venv/bin/python tests/offline_tiles.py
"""
from __future__ import annotations
import sys
import time
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from core import config_store, tilecache # noqa: E402
PASS = 0
FAIL = 0
FAILURES: list[str] = []
def check(name: str, ok: bool, detail: str = "") -> None:
global PASS, FAIL
if ok:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
FAILURES.append(f"{name} {detail}")
print(f" \033[31m✗\033[0m {name} {detail}")
def main() -> int:
print("\033[1m瓦片本地缓存 / 离线可用验证\033[0m")
cache_dir = Path("/tmp/iosloc-offline-tiles")
import shutil
shutil.rmtree(cache_dir, ignore_errors=True)
source_key = "amap"
source = config_store.DEFAULT_TILE_SOURCES[source_key]
# 北京天安门附近、z12 —— 本机实测可达
z, x, y = 12, 3367, 1580
cache = tilecache.TileCache(cache_dir, max_mb=64)
print("\n1) 联网回源并落盘")
online = cache.get_tile(source_key, source, z, x, y)
check("回源返回真实图片", online.status == 200 and online.body[:4] in (b"\x89PNG", b"\xff\xd8"),
f"status={online.status} source={online.source}")
check("标记为 network", online.source == "network", online.source)
check("已写入磁盘缓存", any(cache_dir.rglob("*")), "缓存目录为空")
print("\n2) 断网（把 urllib 换成永远失败）后请求同一瓦片")
real_urlopen = urllib.request.urlopen
def offline(*_args, **_kwargs):
raise OSError("模拟断网：Network is unreachable")
urllib.request.urlopen = offline
try:
cached = cache.get_tile(source_key, source, z, x, y)
check("离线仍返回 200", cached.status == 200, f"status={cached.status}")
check("来源为 cache", cached.source == "cache", cached.source)
check("图片内容与联网时一致", cached.body == online.body,
f"{len(cached.body)}B vs {len(online.body)}B")
print("\n3) 断网 + 未缓存瓦片 → 优雅降级（不抛异常）")
t0 = time.time()
miss = cache.get_tile(source_key, source, z, x, y + 1)
elapsed = time.time() - t0
check("未命中不抛异常", True, "")
check("返回占位图且 status!=200（上层回 404 让前端重试）",
miss.status != 200 and miss.body[:4] == b"\x89PNG",
f"status={miss.status} source={miss.source}")
check("失败快速返回（<5s）", elapsed < 5.0, f"{elapsed:.2f}s")
print("\n4) 缓存统计")
stats = cache.stats
check("命中计数增加", stats["hit"] >= 1, str(stats))
check("统计到离线失败", stats["failed"] >= 1, str(stats))
finally:
urllib.request.urlopen = real_urlopen
print("\n5) 恢复网络后仍可正常回源")
cache2 = tilecache.TileCache(cache_dir, max_mb=64)
other = cache2.get_tile(source_key, source, z, x + 1, y)
check("恢复后回源成功", other.status == 200 and other.body[:4] in (b"\x89PNG", b"\xff\xd8"),
f"status={other.status} source={other.source}")
print("\n6) 容量上限淘汰不破坏可用性")
small = tilecache.TileCache(cache_dir, max_mb=1)
small._sweep()
check("淘汰后仍能读取", small.get_tile(source_key, source, z, x, y).status in (200, 404))
shutil.rmtree(cache_dir, ignore_errors=True)
print(f"\n{'=' * 56}")
print(f"通过 \033[32m{PASS}\033[0m 项，失败 \033[31m{FAIL}\033[0m 项")
for item in FAILURES:
print(" -", item)
return 0 if FAIL == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())
tests/real_device_e2e.py
+175
View File
@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""真机端到端验证：连接 iOS 设备 → 建立隧道 → 沿轨迹播放虚拟定位 → 停止恢复真实定位。
这是**会真的改变设备定位**的测试（结束后会 clear_location 恢复真实定位）。
只有在你明确同意时才运行：
./.venv/bin/python tests/real_device_e2e.py --confirm
./.venv/bin/python tests/real_device_e2e.py --confirm --udid 00008030-...
./.venv/bin/python tests/real_device_e2e.py --confirm --duration 6 --speed 20
"""
from __future__ import annotations
import argparse
import sys
import threading
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))
from core import protocol, server as server_mod # noqa: E402
from selftest import WSClient # noqa: E402
PASS = 0
FAIL = 0
def check(name: str, ok: bool, detail: str = "") -> None:
global PASS, FAIL
if ok:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
print(f" \033[31m✗\033[0m {name} {detail}")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--confirm", action="store_true", help="确认会真的改变设备定位")
parser.add_argument("--udid", default=None, help="指定设备 UDID（默认用发现的唯一 USB 设备）")
parser.add_argument("--port", type=int, default=8811)
parser.add_argument("--duration", type=float, default=6.0, help="播放时长（秒）")
parser.add_argument("--speed", type=float, default=20.0, help="播放倍速")
parser.add_argument("--keep", action="store_true", help="结束后不恢复真实定位（默认会恢复）")
args = parser.parse_args()
if not args.confirm:
print("这是真机测试，会临时改变 iPhone 的定位。请加 --confirm 运行。")
return 2
print("\033[1miOS 虚拟轨迹 — 真机端到端验证\033[0m")
app = server_mod.App(root=ROOT, mock=False, port=args.port)
devices = app.devices.refresh()
if not devices:
print(" \033[31m✗\033[0m 未发现设备")
return 1
print(" 发现的设备：")
for d in devices:
print(f" - {d.name} | iOS {d.ios_version} | {d.connection} | 开发者模式={d.developer_mode} | {d.udid}")
target = None
for d in devices:
if args.udid and d.udid != args.udid:
continue
if not args.udid and d.connection != "usb":
continue
target = d
break
target = target or devices[0]
print(f" 目标设备：{target.name}（iOS {target.ios_version}）\n")
app.devices.select(target.udid)
httpd = server_mod.serve(app, host="127.0.0.1", port=args.port)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
time.sleep(0.4)
client = WSClient("127.0.0.1", args.port, timeout=120)
exit_code = 0
try:
client.send({"type": "hello", "id": "h1", "client": "e2e"})
ack = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "h1", timeout=20)
check("连接服务端", ack is not None)
print("\n → 建立设备会话（iOS 17+ 会自动建 RSD 隧道，最长约 75 秒）…")
started = time.time()
client.send({"type": "device.connect", "id": "c1"})
connected = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "c1", timeout=180)
took = time.time() - started
ok = bool(connected and connected.get("ok", True))
check(f"设备连接成功（{took:.1f}s）", ok, str(connected)[:400])
if not ok:
print(f" 错误：{connected}")
return 1
probe = connected.get("probe") or {}
check("DVT 自检通过", bool(probe.get("ok")), str(probe)[:300])
print(f" 自检详情：{probe.get('detail')}")
print("\n → 单点定位测试…")
client.send({"type": "location.set", "id": "l1", "lat": 39.9088, "lng": 116.3975})
ack = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "l1", timeout=30)
check("单点定位下发", bool(ack and ack.get("ok", True)), str(ack)[:300])
location = client.wait_for(lambda m: m.get("type") == "location" and m.get("source") == "manual", timeout=15)
check("收到单点 location 事件", location is not None, str(location)[:200])
print(f"\n → 轨迹播放测试（{args.duration:.0f} 秒，倍速 {args.speed}x）…")
route = app.routes.build(
[[116.3975, 39.9088], [116.4050, 39.9120], [116.4120, 39.9080]],
mode="driving", use_osrm=False,
)
print(f" 轨迹：{route.distance / 1000:.2f} 公里 / {len(route.path)} 点 / 预计 {route.duration:.0f} 秒")
client.send({"type": "play.start", "id": "s1", "route": route.to_json(),
"speed_scale": args.speed, "loop": False})
ack = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "s1", timeout=30)
check("播放启动", bool(ack and ack.get("ok", True)), str(ack)[:300])
first = client.wait_for(lambda m: m.get("type") == "location" and m.get("source") == "play", timeout=20)
check("播放下发首个定位", first is not None, str(first)[:200])
progressed = client.wait_for(lambda m: m.get("type") == "location" and m.get("progress", 0) > 0.1, timeout=30)
check("定位沿轨迹推进", progressed is not None, str(progressed)[:200])
if progressed:
print(f" 当前进度 {progressed.get('progress'):.1%} → {progressed.get('lat')}, {progressed.get('lng')}")
sent_before = None
with client.lock:
sent_before = len([m for m in client.messages if m.get("type") == "location" and m.get("source") == "play"])
time.sleep(args.duration)
with client.lock:
sent_after = len([m for m in client.messages if m.get("type") == "location" and m.get("source") == "play"])
expect = args.duration * float(app.config.get("send_hz") or 5) * 0.5
check(f"持续下发定位（{sent_after - sent_before} 次 / {args.duration:.0f}s）",
(sent_after - sent_before) >= expect, f"期望 ≥{expect:.0f} 次")
client.send({"type": "play.pause", "id": "p1"})
paused = client.wait_for(lambda m: m.get("type") == "status" and m.get("paused") is True, timeout=15)
check("暂停", paused is not None, str(paused)[:200])
client.send({"type": "play.resume", "id": "r1"})
resumed = client.wait_for(lambda m: m.get("type") == "status" and m.get("paused") is False and m.get("playing") is True, timeout=15)
check("继续", resumed is not None, str(resumed)[:200])
if not args.keep:
print("\n → 停止并恢复真实定位…")
client.send({"type": "play.stop", "id": "st1"})
stopped = client.wait_for(lambda m: m.get("type") == "status" and m.get("playing") is False, timeout=25)
check("停止播放", stopped is not None, str(stopped)[:200])
cleared = client.wait_for(lambda m: m.get("type") == "location" and m.get("source") == "cleared", timeout=20)
check("已恢复设备真实定位", cleared is not None, str(cleared)[:200])
except Exception as exc:
print(f" \033[31m✗\033[0m 异常：{exc.__class__.__name__}: {exc}")
import traceback
traceback.print_exc()
FAIL += 1
finally:
try:
if not args.keep:
app.playback.stop(notify=False, clear_provider=True)
except Exception:
pass
client.close()
app.shutdown()
try:
httpd.shutdown()
httpd.server_close()
except Exception:
pass
print(f"\n{'=' * 50}\n通过 {PASS} 项，失败 {FAIL} 项")
return 0 if FAIL == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())
tests/real_gui_acceptance.py
+190
View File
@@ -0,0 +1,190 @@
#!/usr/bin/env python3
"""真机图形界面验收：在真实设备模式下驱动界面完成绘制→规划→播放，并截图留证。
与 tests/real_device_e2e.py 的区别：这个是**通过浏览器点击界面**完成的（端到端，
包含前端），而 real_device_e2e.py 是直接走 WS 协议。
./.venv/bin/python tests/real_gui_acceptance.py --confirm --url http://127.0.0.1:8790/
"""
from __future__ import annotations
import argparse
import json
import os
import signal
import sys
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))
from ui_acceptance import CDP, launch_chrome # noqa: E402
PASS = 0
FAIL = 0
FAILURES: list[str] = []
def check(name: str, ok: bool, detail: str = "") -> None:
global PASS, FAIL
if ok:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
FAILURES.append(f"{name} {detail}")
print(f" \033[31m✗\033[0m {name} {detail}")
def main() -> int:
global PASS, FAIL
parser = argparse.ArgumentParser()
parser.add_argument("--confirm", action="store_true", help="确认会真的改变设备定位")
parser.add_argument("--url", default="http://127.0.0.1:8790/")
parser.add_argument("--out", default="/tmp/ui-real-acceptance")
parser.add_argument("--port", type=int, default=9466)
parser.add_argument("--wait-connect", type=float, default=90.0)
args = parser.parse_args()
if not args.confirm:
print("真机测试会改变设备定位，请加 --confirm")
return 2
out_dir = Path(args.out)
out_dir.mkdir(parents=True, exist_ok=True)
print("\033[1m真机图形界面验收\033[0m")
process, ws_url = launch_chrome(1440, 900, args.port)
cdp = CDP(ws_url, timeout=180)
try:
cdp.call("Page.enable")
cdp.call("Runtime.enable")
cdp.call("Emulation.setDeviceMetricsOverride",
{"width": 1440, "height": 900, "deviceScaleFactor": 2, "mobile": False})
cdp.call("Page.navigate", {"url": args.url})
time.sleep(4.0)
check("界面已连接服务端",
"已连接" in str(cdp.eval("document.querySelector('#connPill').textContent")),
str(cdp.eval("document.querySelector('#connPill').textContent")))
device_text = str(cdp.eval("(function(){var d=document.querySelector('#deviceCard');return d?d.innerText.replace(/\\n/g,' | '):''})()"))
check("设备卡片识别到真机", "iPhone" in device_text, device_text[:160])
print(f" 设备卡片：{device_text[:120]}")
cdp.shot(out_dir / "01-device-found.png")
# 点「连接设备」
cdp.eval("(function(){var b=document.querySelector('#btnDeviceConnect');if(b)b.click();return !!b})()")
print(" → 正在连接设备（iOS 17+ 会建立 RSD 隧道，最长约 90 秒）…")
connected = False
deadline = time.time() + args.wait_connect
while time.time() < deadline:
time.sleep(1.0)
card = str(cdp.eval("(function(){var d=document.querySelector('#deviceCard');return d?d.innerText:''})()"))
if "模拟中" in card or "已连接" in card and "虚拟定位" in card:
if "模拟中" in card or "就绪" in card:
connected = True
break
card_final = str(cdp.eval("(function(){var d=document.querySelector('#deviceCard');return d?d.innerText.replace(/\\n/g,' | '):''})()"))
check("设备连接成功（卡片显示虚拟定位状态）", connected, card_final[:200])
print(f" 连接后卡片：{card_final[:160]}")
cdp.shot(out_dir / "02-connected.png")
# 绘制 3 个途经点
cdp.eval("(function(){var b=document.querySelector('#toolAdd');if(b)b.click();return !!b})()")
time.sleep(0.4)
box = json.loads(cdp.eval("(function(){var r=document.querySelector('#map').getBoundingClientRect();"
"return JSON.stringify({x:r.x,y:r.y,w:r.width,h:r.height})})()"))
for fx, fy in ((0.28, 0.30), (0.50, 0.48), (0.74, 0.36)):
cdp.click(box["x"] + box["w"] * fx, box["y"] + box["h"] * fy)
time.sleep(0.45)
wp = cdp.eval("(function(){return document.querySelectorAll('#wpList .wp-item, #wpList li, #wpList [data-index]').length})()")
check("地图点选生成途经点", int(wp or 0) >= 3, f"途经点={wp}")
# 规划路线
cdp.eval("(function(){var b=document.querySelector('#btnBuildRoute');if(b)b.click();return !!b})()")
distance = "—"
for _ in range(40):
time.sleep(0.5)
distance = str(cdp.eval("document.querySelector('#statDistance').textContent"))
if distance not in ("—", "-", "", "0"):
break
check("OSRM 路线规划成功", distance not in ("—", "-", "", "0"), f"总距离={distance!r}")
cdp.shot(out_dir / "03-route.png")
# 倍速滑块是对数刻度（0.1x ~ 50x），按 Geo.speedToSlider 反算 1.0x 对应的滑块位置
slider_value = int(round((__import__("math").log(1.0 / 0.1) / __import__("math").log(50.0 / 0.1)) * 1000))
cdp.eval("(function(){var s=document.querySelector('#speedRange');if(!s)return 'no';"
f"s.value={slider_value};s.dispatchEvent(new Event('input',{{bubbles:true}}));"
"s.dispatchEvent(new Event('change',{bubbles:true}));return s.value})()")
time.sleep(0.8)
speed_label = str(cdp.eval("document.querySelector('#speedValue').textContent"))
check("倍速已设为 1.0x（对数滑块）", "1.0x" in speed_label or "1x" in speed_label, speed_label)
# 播放
cdp.eval("(function(){var b=document.querySelector('#btnPlay');if(b)b.click();return !!b})()")
progressed = False
progress_text = ""
for _ in range(30):
time.sleep(0.5)
progress_text = str(cdp.eval("document.querySelector('#chipProgress').textContent"))
coord = str(cdp.eval("document.querySelector('#chipCoord').textContent"))
if any(ch.isdigit() for ch in progress_text) and "0%" not in progress_text and "0 %" not in progress_text:
progressed = True
break
check("真机播放推进", progressed, f"progress={progress_text!r}")
coord = str(cdp.eval("document.querySelector('#chipCoord').textContent"))
check("界面显示实时下发坐标", any(ch.isdigit() for ch in coord), f"coord={coord!r}")
print(f" 实时状态：{progress_text} | {coord}")
cdp.shot(out_dir / "04-playing-real-direct.png")
# 日志面板展开，确认有真实下发日志
cdp.eval("(function(){var b=document.querySelector('#btnToggleLog');if(b)b.click();return !!b})()")
time.sleep(1.2)
log_text = str(cdp.eval("(function(){var p=document.querySelector('#logBody');return p?p.innerText:''})()"))
check("日志面板记录了真机下发", "下发" in log_text or "定位" in log_text, log_text[-200:].replace("\n", " | "))
cdp.shot(out_dir / "05-logs-real.png")
# 停止并恢复真实定位
cdp.eval("(function(){var b=document.querySelector('#btnStop');if(b)b.click();return !!b})()")
time.sleep(2.5)
state = str(cdp.eval("document.querySelector('#chipPlayState').textContent"))
check("停止后回到未播放", "未播放" in state or "已停止" in state or "已结束" in state, state)
cdp.shot(out_dir / "06-stopped.png")
errors = []
for event in cdp.events:
if event.get("method") == "Runtime.exceptionThrown":
detail = event["params"]["exceptionDetails"]
errors.append(f"{detail.get('text')} {(detail.get('exception') or {}).get('description', '')[:160]}")
elif event.get("method") == "Runtime.consoleAPICalled" and event["params"].get("type") == "error":
parts = [str(a.get("value", a.get("description", "")))[:140] for a in event["params"].get("args", [])]
errors.append("console.error: " + " ".join(parts))
check("真机流程中页面无 JS 报错", not errors, " | ".join(errors[:3]))
except Exception as exc:
FAIL += 1
FAILURES.append(f"异常 {exc}")
print(f" \033[31m✗\033[0m 异常：{exc.__class__.__name__}: {exc}")
import traceback
traceback.print_exc()
finally:
try:
cdp.close()
except Exception:
pass
try:
os.killpg(os.getpgid(process.pid), 15)
except Exception:
process.terminate()
print(f"\n{'=' * 56}")
print(f"真机图形界面验收：通过 \033[32m{PASS}\033[0m 项，失败 \033[31m{FAIL}\033[0m 项")
print(f"截图目录：{out_dir}")
for item in FAILURES:
print(" -", item[:200])
return 0 if FAIL == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())
tests/selftest.py
+496
View File
@@ -0,0 +1,496 @@
#!/usr/bin/env python3
"""端到端自检：单元 + 集成（无设备，走 --mock）。
运行：./.venv/bin/python tests/selftest.py
"""
from __future__ import annotations
import base64
import hashlib
import json
import os
import socket
import struct
import sys
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from core import config_store, geoutils, protocol, route_engine, tilecache, ws # noqa: E402
PASS = 0
FAIL = 0
FAILURES: list[str] = []
def check(name: str, condition: bool, detail: str = "") -> None:
global PASS, FAIL
if condition:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
FAILURES.append(f"{name} {detail}")
print(f" \033[31m✗\033[0m {name} {detail}")
def section(title: str) -> None:
print(f"\n\033[1m{title}\033[0m")
# ---------------------------------------------------------------- 单元
def test_protocol() -> None:
section("1. 协议与校验")
check("合法经纬度通过", protocol.validate_latlng(39.9, 116.4) == (39.9, 116.4))
for bad in [(91, 0), (0, 181), ("a", 1), (float("nan"), 0)]:
try:
protocol.validate_latlng(*bad)
check(f"非法经纬度被拒 {bad}", False)
except protocol.AppError:
check(f"非法经纬度被拒 {bad}", True)
check("倍速上下限", protocol.validate_speed_scale(1.5) == 1.5)
try:
protocol.validate_speed_scale(999)
check("倍速超限被拒", False)
except protocol.AppError:
check("倍速超限被拒", True)
check("模式解析", protocol.validate_mode("cycling") == "cycling")
check("JSON 中文不被转义", "定位" in protocol.dumps({"msg": "定位"}))
def test_geo() -> None:
section("2. 地理计算")
d = geoutils.haversine(39.9, 116.4, 39.9, 116.41)
check("haversine 精度", 830 < d < 870, f"got {d:.1f}m")
b = geoutils.bearing(39.9, 116.4, 40.0, 116.4)
check("正北方位角≈0", b < 1 or b > 359, f"got {b:.2f}")
path = [(39.9, 116.4), (39.91, 116.41)]
dense = geoutils.densify(path, 20.0)
check("路径加密点变多", len(dense) > 50, f"{len(dense)} 点")
acc = geoutils.cumulative_lengths(dense)
total = acc[-1]
check("累计弧长单调递增", all(acc[i] <= acc[i + 1] for i in range(len(acc) - 1)))
mid, _ = geoutils.interpolate(dense, acc, total / 2)
check("弧长中点落在起终点之间", 39.9 <= mid[0] <= 39.91, f"{mid}")
simplified = geoutils.simplify(dense, 2.0)
check("简化后点数下降且首尾保留", len(simplified) < len(dense) and simplified[0] == dense[0])
def test_route_engine() -> None:
section("3. 路线引擎")
config = config_store.ConfigStore(ROOT / "config.json")
engine = route_engine.RouteEngine(config)
route = engine.build([[116.40, 39.90], [116.41, 39.905], [116.42, 39.90]], mode="driving", use_osrm=False)
check("直线兜底引擎标记", route.engine == "straight")
check("路径非空且已加密", len(route.path) > 100, f"{len(route.path)} 点")
check("距离 > 1.5km", route.distance > 1500, f"{route.distance:.0f}m")
check("时长合理", 60 < route.duration < 1200, f"{route.duration:.0f}s")
check("速度剖面长度一致", len(route.speeds) == len(route.path))
check("速度均为正", all(v > 0 for v in route.speeds))
mid = route.point_at_distance(route.distance / 2)
check("中点取点有效", 39.89 <= mid[0] <= 39.91, f"{mid}")
payload = route.to_json()
check("JSON 坐标是 [lng,lat]", abs(payload["path"][0][0] - 116.40) < 0.01, str(payload["path"][0]))
restored = route_engine.sanitize_route_payload(payload)
check("route 往返一致", abs(restored.distance - route.distance) < 1.0, f"{restored.distance:.1f} vs {route.distance:.1f}")
check("高亮子路径不超过全长", len(route.highlight(route.distance / 2)) <= len(route.path))
# GPX 往返
gpx = route_engine.to_gpx(route, creator="selftest")
check("GPX 含 trkpt", "<trkpt" in gpx and gpx.startswith("<?xml"))
coords, name, _ = route_engine.from_gpx(gpx)
check("GPX 导入保形（不抽稀丢点）", len(coords) == len(route.path), f"{len(coords)} vs {len(route.path)}")
total = geoutils.cumulative_lengths(coords)[-1]
check("GPX 往返距离误差 < 2m", abs(total - route.distance) < 2.0, f"误差 {abs(total - route.distance):.3f}m")
# 单点与异常
single = engine.build([[116.40, 39.90]], mode="walking", use_osrm=False)
check("单点轨迹可用", len(single.path) == 1 and single.distance == 0)
for bad, label in [([], "空途经点"), ([[200, 39]], "非法经度")]:
try:
engine.build(bad, use_osrm=False)
check(f"{label}被拒", False)
except protocol.AppError:
check(f"{label}被拒", True)
# OSRM URL 生成（不依赖外网）
check("flight 不走 OSRM", engine._osrm([(39.9, 116.4), (39.91, 116.41)], "flight") is None)
def test_tiles() -> None:
section("4. 瓦片与缓存")
source = config_store.DEFAULT_TILE_SOURCES["amap"]
url1 = tilecache.build_url(source, 12, 3367, 1580)
check("高德 URL 生成", url1.startswith("https://webrd0") and "{x}" not in url1, url1)
url2 = tilecache.build_url(source, 12, 3368, 1580)
check("子域轮换生效", url1.split("/")[2] != url2.split("/")[2], f"{url1.split('/')[2]} vs {url2.split('/')[2]}")
arc = config_store.DEFAULT_TILE_SOURCES["arcgis_sat"]
check("Esri URL 含 z/y/x", "/tile/12/1580/3367" in tilecache.build_url(arc, 12, 3367, 1580))
check("瓦片坐标校验", tilecache.validate_tile_request(12, 3367, 1580) == (12, 3367, 1580))
try:
tilecache.validate_tile_request(12, 99999, 1)
check("越界瓦片被拒", False)
except protocol.AppError:
check("越界瓦片被拒", True)
tmp = ROOT / ".cache-test"
cache = tilecache.TileCache(tmp, max_mb=1)
result = cache.get_tile("amap", source, 12, 0, 0)
check("瓦片请求返回图片", result.body[:4] in (b"\x89PNG", b"\xff\xd8") and result.content_type.startswith("image/"),
f"{result.status} {result.source} {len(result.body)}B")
check("瓦片已落盘缓存", any(tmp.rglob("*.png")) or any(tmp.rglob("*.jpg")))
result2 = cache.get_tile("amap", source, 12, 0, 0)
check("第二次命中缓存", result2.source == "cache", result2.source)
removed = cache.clear()
check("清理缓存", removed >= 1, f"removed={removed}")
try:
import shutil
shutil.rmtree(tmp, ignore_errors=True)
except Exception:
pass
def test_config() -> None:
section("5. 配置")
path = ROOT / ".config-test.json"
path.unlink(missing_ok=True)
store = config_store.ConfigStore(path)
check("默认瓦片源为高德", store.get("tile_source") == "amap")
try:
store.update({"send_hz": 999})
check("非法 send_hz 被拒（不再静默夹取）", False)
except protocol.AppError as exc:
check("非法 send_hz 被拒（不再静默夹取）", exc.code == "bad_config", exc.code)
store.update({"send_hz": 8})
check("合法 send_hz 生效", store.get("send_hz") == 8.0, str(store.get("send_hz")))
store.update({"tile_source": "amap_sat"})
store2 = config_store.ConfigStore(path)
check("配置持久化", store2.get("tile_source") == "amap_sat")
try:
store.update({"不存在的项": 1})
check("未知配置被拒", False)
except protocol.AppError:
check("未知配置被拒", True)
sources = store.tile_sources()
check("瓦片源齐全", all(k in sources for k in ("amap", "amap_sat", "arcgis_sat", "osm")))
path.unlink(missing_ok=True)
# ---------------------------------------------------------------- WS 客户端
class WSClient:
def __init__(self, host: str, port: int, timeout: float = 30.0) -> None:
self.sock = socket.create_connection((host, port), timeout=timeout)
key = base64.b64encode(os.urandom(16)).decode()
request = (
f"GET /ws HTTP/1.1\r\nHost: {host}:{port}\r\nUpgrade: websocket\r\n"
f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n"
)
self.sock.sendall(request.encode())
header = b""
while b"\r\n\r\n" not in header:
chunk = self.sock.recv(4096)
if not chunk:
raise RuntimeError("握手失败")
header += chunk
if b"101" not in header.split(b"\r\n")[0]:
raise RuntimeError(f"握手未升级：{header[:200]!r}")
expected = base64.b64encode(hashlib.sha1((key + ws.GUID).encode()).digest()).decode()
if expected.encode() not in header:
raise RuntimeError("Sec-WebSocket-Accept 校验失败")
self.buf = header.split(b"\r\n\r\n", 1)[1]
self.messages: list[dict] = []
self.lock = threading.Lock()
self._stop = threading.Event()
self.reader = threading.Thread(target=self._read_loop, daemon=True)
self.reader.start()
def _recv_exact(self, n: int) -> bytes | None:
while len(self.buf) < n:
try:
chunk = self.sock.recv(65536)
except OSError:
return None
if not chunk:
return None
self.buf += chunk
out, self.buf = self.buf[:n], self.buf[n:]
return out
def _read_loop(self) -> None:
while not self._stop.is_set():
head = self._recv_exact(2)
if head is None:
return
opcode = head[0] & 0x0F
length = head[1] & 0x7F
if length == 126:
ext = self._recv_exact(2)
if ext is None:
return
length = struct.unpack("!H", ext)[0]
elif length == 127:
ext = self._recv_exact(8)
if ext is None:
return
length = struct.unpack("!Q", ext)[0]
payload = self._recv_exact(length) if length else b""
if payload is None:
return
if opcode == 0x8:
return
if opcode in (0x9, 0xA):
continue
try:
with self.lock:
self.messages.append(json.loads(payload.decode("utf-8")))
except Exception:
continue
def send(self, obj: dict) -> None:
payload = json.dumps(obj, ensure_ascii=False).encode()
header = bytearray([0x81])
mask = os.urandom(4)
length = len(payload)
if length < 126:
header.append(0x80 | length)
elif length < 65536:
header.append(0x80 | 126)
header.extend(struct.pack("!H", length))
else:
header.append(0x80 | 127)
header.extend(struct.pack("!Q", length))
header.extend(mask)
masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
self.sock.sendall(bytes(header) + masked)
def wait_for(self, predicate, timeout: float = 15.0) -> dict | None:
deadline = time.time() + timeout
seen = 0
while time.time() < deadline:
with self.lock:
batch = self.messages[seen:]
seen = len(self.messages)
for message in batch:
if predicate(message):
return message
time.sleep(0.05)
return None
def close(self) -> None:
self._stop.set()
try:
self.sock.close()
except OSError:
pass
def test_integration() -> None:
section("6. 端到端（--mock 模式，真 HTTP + 真 WebSocket）")
from core import server as server_mod
port = 8899
app = server_mod.App(root=ROOT, mock=True, port=port)
app.devices.refresh()
httpd = server_mod.serve(app, host="127.0.0.1", port=port)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
base = f"http://127.0.0.1:{port}"
time.sleep(0.4)
try:
# --- REST ---
with urllib.request.urlopen(base + "/api/health", timeout=5) as response:
health = json.loads(response.read())
check("GET /api/health", health.get("ok") and health.get("version") == protocol.APP_VERSION, str(health)[:120])
check("mock 模式标记", health.get("mock") is True)
with urllib.request.urlopen(base + "/api/devices", timeout=5) as response:
devices = json.loads(response.read())
check("GET /api/devices 返回模拟设备", len(devices["devices"]) == 1 and devices["devices"][0]["ios_major"] == 26,
str(devices)[:150])
with urllib.request.urlopen(base + "/api/settings/tiles", timeout=5) as response:
tiles = json.loads(response.read())
keys = {item["key"] for item in tiles["sources"]}
check("瓦片源接口含 key/datum", "amap" in keys and tiles["sources"][0]["datum"] in ("gcj02", "wgs84"), str(tiles)[:150])
check("瓦片源三种字段名都在", tiles["current"] == tiles["tile_source"] == tiles["selected"])
with urllib.request.urlopen(base + "/", timeout=5) as response:
html = response.read().decode("utf-8", "replace")
check("GET / 返回前端页面", "<html" in html.lower(), html[:80])
# 路线规划（REST）
body = json.dumps({"waypoints": [[116.40, 39.90], [116.41, 39.905], [116.42, 39.90]], "mode": "driving"}).encode()
request = urllib.request.Request(base + "/api/route/build", data=body, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(request, timeout=30) as response:
route_result = json.loads(response.read())
check("POST /api/route/build", route_result.get("ok") and route_result["route"]["distance"] > 500,
str(route_result)[:120])
# GPX 往返（REST）
export_body = json.dumps({"route": route_result["route"], "name": "自检轨迹"}).encode()
request = urllib.request.Request(base + "/api/gpx/export", data=export_body, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(request, timeout=15) as response:
gpx_text = response.read().decode()
check("POST /api/gpx/export", "<trkpt" in gpx_text and "自检轨迹" in gpx_text)
request = urllib.request.Request(base + "/api/gpx/import", data=gpx_text.encode(), headers={"Content-Type": "application/gpx+xml"})
with urllib.request.urlopen(request, timeout=15) as response:
imported = json.loads(response.read())
check("POST /api/gpx/import", imported["ok"] and imported["count"] >= 2, str(imported)[:120])
check("GPX 导入不丢几何（返回全量点）", len(imported["waypoints"]) == imported["count"] >= 100,
f"count={imported['count']}")
check("GPX 导入同时提供 path/points 全量数组",
imported.get("path") == imported["waypoints"] and imported.get("points") == imported["waypoints"])
imported_coords = [(p[1], p[0]) for p in imported["waypoints"]]
imported_total = geoutils.cumulative_lengths(imported_coords)[-1]
check("GPX 导入距离保真（误差 < 2m）", abs(imported_total - route_result["route"]["distance"]) < 2.0,
f"误差 {abs(imported_total - route_result['route']['distance']):.3f}m")
# 瓦片代理
with urllib.request.urlopen(base + "/api/tiles/amap/12/3367/1580.png", timeout=25) as response:
tile_bytes = response.read()
check("GET /api/tiles 返回图片", tile_bytes[:4] in (b"\x89PNG", b"\xff\xd8"), f"{len(tile_bytes)}B")
# --- WebSocket ---
client = WSClient("127.0.0.1", port)
client.send({"type": "hello", "id": "h1", "client": "selftest"})
ack = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "h1")
check("WS hello ack 携带 welcome 数据", ack is not None and ack.get("mock") is True and "config" in ack,
str(ack)[:140])
check("welcome 里带设备与隧道快照", ack is not None and "tunnel" in ack and ack.get("version") == protocol.APP_VERSION,
str(ack)[:140])
client.send({"type": "ping", "id": "p1"})
pong = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "p1")
check("WS ping→ack(ts)", pong is not None and "ts" in pong, str(pong)[:120])
client.send({"type": "device.connect", "id": "c1"})
connected = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "c1", timeout=20)
check("WS device.connect", connected is not None and connected.get("ok", True), str(connected)[:160])
client.send({"type": "location.set", "id": "l1", "lat": 39.9088, "lng": 116.3975})
ack = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "l1", timeout=15)
check("WS location.set", ack is not None and ack.get("ok", True), str(ack)[:160])
location = client.wait_for(lambda m: m.get("type") == "location" and m.get("source") == "manual", timeout=10)
check("收到 location 事件（单点）", location is not None and abs(location["lng"] - 116.3975) < 1e-6, str(location)[:160])
# 多设备/非法命令
client.send({"type": "play.start", "id": "e1", "route": {"path": [], "waypoints": []}})
bad = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "e1", timeout=10)
check("空轨迹被拒并回中文错误", bad is not None and bad.get("ok") is False and "轨迹" in bad.get("error", ""), str(bad)[:160])
# 播放流程
route_payload = route_result["route"]
client.send({"type": "play.start", "id": "s1", "route": route_payload, "speed_scale": 8.0, "loop": False})
started = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "s1", timeout=20)
check("WS play.start", started is not None and started.get("ok", True), str(started)[:160])
first = client.wait_for(lambda m: m.get("type") == "location" and m.get("source") == "play", timeout=10)
check("播放下发首个定位", first is not None and first.get("progress", 1) < 0.2, str(first)[:160])
progressed = client.wait_for(lambda m: m.get("type") == "location" and m.get("progress", 0) > 0.05, timeout=20)
check("播放推进（进度增长）", progressed is not None, str(progressed)[:160])
status = client.wait_for(lambda m: m.get("type") == "status" and m.get("playing") is True, timeout=10)
check("status 事件含剩余时间", status is not None and "remaining" in status, str(status)[:160])
client.send({"type": "play.pause", "id": "p2"})
paused = client.wait_for(lambda m: m.get("type") == "status" and m.get("paused") is True, timeout=10)
check("暂停生效", paused is not None, str(paused)[:120])
time.sleep(0.6)
with client.lock:
after_pause = len([m for m in client.messages if m.get("type") == "location" and m.get("source") == "play"])
time.sleep(0.8)
with client.lock:
still = len([m for m in client.messages if m.get("type") == "location" and m.get("source") == "play"])
check("暂停后不再下发", still - after_pause <= 1, f"{after_pause} → {still}")
client.send({"type": "play.resume", "id": "r1"})
resumed = client.wait_for(lambda m: m.get("type") == "status" and m.get("paused") is False and m.get("playing") is True, timeout=10)
check("继续生效", resumed is not None, str(resumed)[:120])
client.send({"type": "play.speed", "id": "sp1", "speed_scale": 20.0})
fast = client.wait_for(lambda m: m.get("type") == "status" and m.get("speed_scale") == 20.0, timeout=10)
check("倍速调整生效", fast is not None, str(fast)[:120])
client.send({"type": "play.seek", "id": "sk1", "progress": 0.5})
seeked = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "sk1", timeout=10)
check("seek 生效", seeked is not None and seeked.get("ok", True), str(seeked)[:120])
at_half = client.wait_for(lambda m: m.get("type") == "location" and m.get("progress", 0) > 0.45, timeout=10)
check("seek 后位置过半", at_half is not None, str(at_half)[:140])
# 播放到结束（循环 + 高倍速）
client.send({"type": "play.speed", "id": "sp2", "speed_scale": 50.0})
client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "sp2", timeout=10)
done = client.wait_for(
lambda m: m.get("type") == "status" and m.get("playing") is False and m.get("progress", 0) >= 0.999,
timeout=90,
)
check("轨迹播放到终点", done is not None, str(done)[:160])
# 循环模式
client.send({"type": "play.start", "id": "s2", "route": route_payload, "speed_scale": 50.0, "loop": True})
client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "s2", timeout=15)
wrapped = client.wait_for(
lambda m: m.get("type") == "status" and m.get("loop") is True and m.get("progress", 1) < 0.5 and m.get("playing"),
timeout=90,
)
check("loop 模式回到起点继续", wrapped is not None, str(wrapped)[:160])
client.send({"type": "play.stop", "id": "st1"})
stopped = client.wait_for(lambda m: m.get("type") == "status" and m.get("playing") is False, timeout=15)
check("停止播放", stopped is not None, str(stopped)[:140])
cleared = client.wait_for(lambda m: m.get("type") == "location" and m.get("source") == "cleared", timeout=10)
check("停止后通知恢复真实定位", cleared is not None, str(cleared)[:120])
client.send({"type": "log.list", "id": "lg1", "limit": 50})
logs = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "lg1", timeout=10)
check("日志回填", logs is not None, str(logs)[:120])
client.send({"type": "不存在的命令", "id": "x1"})
unknown = client.wait_for(lambda m: m.get("type") == "ack" and m.get("id") == "x1", timeout=10)
check("未知命令回中文错误", unknown is not None and unknown.get("ok") is False, str(unknown)[:140])
client.close()
time.sleep(0.3)
finally:
app.shutdown()
try:
httpd.shutdown()
httpd.server_close()
except Exception:
pass
def main() -> int:
print(f"\033[1m{protocol.APP_NAME} 自检\033[0m (Python {sys.version.split()[0]})")
started = time.time()
test_protocol()
test_geo()
test_route_engine()
test_tiles()
test_config()
try:
test_integration()
except Exception as exc:
global FAIL
FAIL += 1
FAILURES.append(f"集成测试异常：{exc}")
print(f" \033[31m✗\033[0m 集成测试异常：{exc.__class__.__name__}: {exc}")
import traceback
traceback.print_exc()
print(f"\n{'=' * 56}")
print(f"通过 \033[32m{PASS}\033[0m 项，失败 \033[31m{FAIL}\033[0m 项，用时 {time.time() - started:.1f}s")
if FAILURES:
print("\n失败明细：")
for item in FAILURES:
print(f" - {item}")
return 1
print("\033[32m全部通过\033[0m")
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/ui_acceptance.py
+419
View File
@@ -0,0 +1,419 @@
#!/usr/bin/env python3
"""图形界面验收：用真实 Chrome 驱动界面完成一整套操作并截图。
做的是**图形界面级**验收：在页面上真实点击「添加途经点」→ 在地图上点三个点 →
点「加载 OSRM 路线」→ 点「播放」→ 截图，并断言界面状态与后端状态一致。
CDP 走 WebSocket，本机没有 websocket-client，因此这里复用 core/ws.py 的帧实现。
用法：
./.venv/bin/python tests/ui_acceptance.py # 默认连 127.0.0.1:8787
./.venv/bin/python tests/ui_acceptance.py --url http://127.0.0.1:8787/ --out /tmp/ui
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import socket
import struct
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from urllib.parse import urlparse # noqa: E402
CHROME_CANDIDATES = [
"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
"/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
"/Applications/Brave Browser.app/Contents/MacOS/Brave Browser",
"/Applications/Chromium.app/Contents/MacOS/Chromium",
]
PASS = 0
FAIL = 0
FAILURES: list[str] = []
def check(name: str, ok: bool, detail: str = "") -> None:
global PASS, FAIL
if ok:
PASS += 1
print(f" \033[32m✓\033[0m {name}")
else:
FAIL += 1
FAILURES.append(f"{name} {detail}")
print(f" \033[31m✗\033[0m {name} {detail}")
class CDP:
"""最小可用的 Chrome DevTools Protocol 客户端。"""
def __init__(self, ws_url: str, timeout: float = 30.0) -> None:
parsed = urlparse(ws_url)
self.sock = socket.create_connection((parsed.hostname or "127.0.0.1", parsed.port or 9222), timeout=timeout)
self.sock.settimeout(timeout)
key = base64.b64encode(os.urandom(16)).decode()
path = parsed.path + (("?" + parsed.query) if parsed.query else "")
request = (
f"GET {path} HTTP/1.1\r\nHost: {parsed.hostname}:{parsed.port}\r\n"
f"Upgrade: websocket\r\nConnection: Upgrade\r\n"
f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n"
)
self.sock.sendall(request.encode())
self.buf = b""
header = b""
while b"\r\n\r\n" not in header:
chunk = self.sock.recv(4096)
if not chunk:
raise RuntimeError("CDP 握手失败")
header += chunk
head, _, rest = header.partition(b"\r\n\r\n")
if b"101" not in head.split(b"\r\n")[0]:
raise RuntimeError(f"CDP 未升级：{head[:200]!r}")
self.buf = rest
self.counter = 0
self.timeout = timeout
def _recv(self, n: int) -> bytes:
while len(self.buf) < n:
chunk = self.sock.recv(65536)
if not chunk:
raise RuntimeError("CDP 连接关闭")
self.buf += chunk
out, self.buf = self.buf[:n], self.buf[n:]
return out
def _read_frame(self) -> str:
head = self._recv(2)
opcode = head[0] & 0x0F
length = head[1] & 0x7F
if length == 126:
length = struct.unpack("!H", self._recv(2))[0]
elif length == 127:
length = struct.unpack("!Q", self._recv(8))[0]
payload = self._recv(length) if length else b""
if opcode == 0x8:
raise RuntimeError("CDP 服务端关闭连接")
if opcode == 0x9: # ping → pong（带掩码）
mask = os.urandom(4)
masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
self.sock.sendall(bytes([0x8A, 0x80 | len(payload)]) + mask + masked)
return self._read_frame()
return payload.decode("utf-8", "replace")
def _send(self, obj: dict) -> None:
payload = json.dumps(obj).encode()
mask = os.urandom(4)
header = bytearray([0x81])
length = len(payload)
if length < 126:
header.append(0x80 | length)
elif length < 65536:
header.append(0x80 | 126)
header.extend(struct.pack("!H", length))
else:
header.append(0x80 | 127)
header.extend(struct.pack("!Q", length))
header.extend(mask)
self.sock.sendall(bytes(header) + bytes(b ^ mask[i % 4] for i, b in enumerate(payload)))
def call(self, method: str, params: dict | None = None) -> dict:
self.counter += 1
message_id = self.counter
self._send({"id": message_id, "method": method, "params": params or {}})
while True:
data = json.loads(self._read_frame())
if data.get("id") == message_id:
if "error" in data:
raise RuntimeError(f"CDP {method} 失败：{data['error']}")
return data.get("result") or {}
# 其它事件（Runtime.consoleAPICalled 等）先记录到 self.events
self.events.append(data)
events: list = []
def eval(self, expression: str, await_promise: bool = False):
result = self.call("Runtime.evaluate", {
"expression": expression,
"returnByValue": True,
"awaitPromise": await_promise,
})
if result.get("exceptionDetails"):
detail = result["exceptionDetails"]
raise RuntimeError(f"页面 JS 异常：{detail.get('text')} "
f"{(detail.get('exception') or {}).get('description', '')[:300]}")
return (result.get("result") or {}).get("value")
def click(self, x: float, y: float, button: str = "left", clicks: int = 1) -> None:
for event_type in ("mousePressed", "mouseReleased"):
self.call("Input.dispatchMouseEvent", {
"type": event_type, "x": x, "y": y, "button": button,
"clickCount": clicks, "buttons": 1 if event_type == "mousePressed" else 0,
})
def move(self, x: float, y: float) -> None:
self.call("Input.dispatchMouseEvent", {"type": "mouseMoved", "x": x, "y": y, "buttons": 0})
def shot(self, path) -> None:
data = self.call("Page.captureScreenshot", {"format": "png"})
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(base64.b64decode(data["data"]))
def close(self) -> None:
try:
self.sock.close()
except OSError:
pass
def launch_chrome(width: int, height: int, port: int) -> subprocess.Popen:
chrome = next((p for p in CHROME_CANDIDATES if os.path.exists(p)), None)
if not chrome:
raise SystemExit("未找到 Chrome / Edge / Chromium")
profile = Path("/tmp/iosloc-ui-acceptance-profile")
subprocess.run(["rm", "-rf", str(profile)], check=False)
profile.mkdir(parents=True, exist_ok=True)
process = subprocess.Popen(
[
chrome, "--headless=new", "--disable-gpu", "--no-first-run",
"--no-default-browser-check", "--hide-scrollbars", "--disable-extensions",
f"--remote-debugging-port={port}", f"--user-data-dir={profile}",
f"--window-size={width},{height}", "about:blank",
],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True,
)
deadline = time.time() + 25
while time.time() < deadline:
try:
with urllib.request.urlopen(f"http://127.0.0.1:{port}/json/list", timeout=2) as response:
pages = json.loads(response.read())
for page in pages:
if page.get("type") == "page":
return process, page["webSocketDebuggerUrl"]
except Exception:
time.sleep(0.4)
process.terminate()
raise SystemExit("无法连接 Chrome 调试端口")
def main() -> int:
global PASS, FAIL
parser = argparse.ArgumentParser()
parser.add_argument("--url", default="http://127.0.0.1:8787/")
parser.add_argument("--out", default="/tmp/ui-acceptance")
parser.add_argument("--port", type=int, default=9444)
parser.add_argument("--width", type=int, default=1440)
parser.add_argument("--height", type=int, default=900)
parser.add_argument("--keep-open", action="store_true", help="结束后不关闭 Chrome")
args = parser.parse_args()
out_dir = Path(args.out)
out_dir.mkdir(parents=True, exist_ok=True)
print("\033[1m图形界面验收（真实 Chrome 驱动）\033[0m")
process, ws_url = launch_chrome(args.width, args.height, args.port)
cdp = None
try:
cdp = CDP(ws_url)
cdp.call("Page.enable")
cdp.call("Runtime.enable")
cdp.call("Emulation.setDeviceMetricsOverride",
{"width": args.width, "height": args.height, "deviceScaleFactor": 2, "mobile": False})
cdp.call("Page.navigate", {"url": args.url})
time.sleep(3.5)
# ---- 0. 页面与服务端连通 ----
title = cdp.eval("document.title")
check("页面已加载", bool(title), str(title))
ws_state = cdp.eval("(function(){var d=document.querySelector('#connPill');return d?d.textContent.trim():''})()")
check("界面显示已连接服务端", "已连接" in str(ws_state), f"connPill={ws_state!r}")
device_text = cdp.eval("(function(){var d=document.querySelector('#deviceCard');return d?d.innerText.replace(/\\n/g,' | '):''})()")
check("设备卡片显示设备信息", "iPhone" in str(device_text) or "模拟" in str(device_text), str(device_text)[:120])
# ---- 1. 地图渲染 ----
map_info = cdp.eval(
"(function(){var c=document.querySelector('#map canvas');"
"var t=document.querySelectorAll('#map img').length;"
"return JSON.stringify({canvas:!!c,imgs:t,w:c?c.width:0,h:c?c.height:0})})()"
)
info = json.loads(map_info) if map_info else {}
check("地图画布已渲染", bool(info.get("w")) and bool(info.get("h")), str(info))
# ---- 2. 添加途经点：点工具条 + 在地图上点三个位置 ----
clicked_tool = cdp.eval(
"(function(){var b=document.querySelector('#toolAdd');if(!b)return 'no-btn';b.click();return b.className})()"
)
check("「添加途经点」按钮可点击", clicked_tool != "no-btn", str(clicked_tool))
time.sleep(0.4)
box = cdp.eval(
"(function(){var m=document.querySelector('#map');var r=m.getBoundingClientRect();"
"return JSON.stringify({x:r.x,y:r.y,w:r.width,h:r.height})})()"
)
rect = json.loads(box)
points = [
(rect["x"] + rect["w"] * 0.30, rect["y"] + rect["h"] * 0.35),
(rect["x"] + rect["w"] * 0.52, rect["y"] + rect["h"] * 0.55),
(rect["x"] + rect["w"] * 0.72, rect["y"] + rect["h"] * 0.42),
]
for (px, py) in points:
cdp.click(px, py)
time.sleep(0.45)
time.sleep(0.6)
wp_count = cdp.eval(
"(function(){var l=document.querySelectorAll('#wpList .wp-item, #wpList li, #wpList [data-index]');"
"return l.length})()"
)
check("地图点击添加了 3 个途经点", int(wp_count or 0) >= 3, f"列表条目={wp_count}")
cdp.shot(out_dir / "01-waypoints.png")
# ---- 3. 加载 OSRM 路线 ----
cdp.eval("(function(){var b=document.querySelector('#btnBuildRoute');if(b)b.click();return !!b})()")
route_ok = False
for _ in range(40):
time.sleep(0.5)
meta = cdp.eval("(function(){var d=document.querySelector('#statDistance');return d?d.textContent.trim():''})()")
if meta and meta not in ("—", "-", "0", ""):
route_ok = True
break
distance_text = cdp.eval("(function(){var d=document.querySelector('#statDistance');return d?d.textContent.trim():''})()")
check("「加载 OSRM 路线」生成了路线", route_ok, f"总距离={distance_text!r}")
engine_text = cdp.eval("(function(){var d=document.querySelector('#statEngine');return d?d.textContent.trim():''})()")
check("路线引擎已显示", bool(engine_text) and engine_text != "—", f"引擎={engine_text!r}")
cdp.shot(out_dir / "02-route.png")
# ---- 4. 播放（按钮是异步生效的：点击后等状态真正变为"播放中"，必要时重试一次） ----
def click_play() -> None:
cdp.eval("(function(){var b=document.querySelector('#btnPlay, [data-action=play]');"
"if(b && !b.disabled)b.click();return !!b})()")
def wait_playing(timeout: float = 12.0) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
state = str(cdp.eval("(function(){var d=document.querySelector('#chipPlayState');return d?d.textContent.trim():''})()"))
if "播放中" in state or "已暂停" in state:
return True
time.sleep(0.3)
return False
click_play()
playing = wait_playing(12.0)
if not playing:
time.sleep(1.0)
click_play()
playing = wait_playing(12.0)
check("点击播放后进入播放态", playing,
str(cdp.eval("document.querySelector('#chipPlayState').textContent")))
progressed = False
progress_text = ""
coord_text = ""
for _ in range(40):
time.sleep(0.5)
progress_text = str(cdp.eval("(function(){var d=document.querySelector('#chipProgress');return d?d.textContent.trim():''})()"))
coord_text = str(cdp.eval("(function(){var d=document.querySelector('#chipCoord');return d?d.textContent.trim():''})()"))
if any(ch.isdigit() for ch in progress_text) and "0%" not in progress_text and "0 %" not in progress_text:
progressed = True
break
check("播放推进（地图 chips 显示进度）", progressed, f"chipProgress={progress_text!r}")
check("实时坐标已显示", any(ch.isdigit() for ch in coord_text), f"chipCoord={coord_text!r}")
play_state = cdp.eval("(function(){var d=document.querySelector('#chipPlayState');return d?d.textContent.trim():''})()")
check("播放状态文案正常", bool(play_state) and "未播放" not in str(play_state), f"chipPlayState={play_state!r}")
cdp.shot(out_dir / "03-playing.png")
# ---- 5. 暂停 / 继续 / 停止（暂停是异步推送，采样窗口内看是否出现过"已暂停"） ----
cdp.eval("(function(){var b=document.querySelector('#btnPause, [data-action=pause]');if(b)b.click();return !!b})()")
paused_seen = False
paused_sample = ""
for _ in range(20):
time.sleep(0.15)
paused_sample = str(cdp.eval("(function(){var d=document.querySelector('#chipPlayState');return d?d.textContent.trim():''})()"))
if "暂停" in paused_sample:
paused_seen = True
break
check("暂停后状态文案变化", paused_seen, f"{paused_sample!r}")
cdp.eval("(function(){var b=document.querySelector('#btnResume, [data-action=resume]');if(b)b.click();return !!b})()")
time.sleep(1.0)
resumed_text = cdp.eval("(function(){var d=document.querySelector('#chipPlayState');return d?d.textContent.trim():''})()")
check("继续后恢复播放", "暂停" not in str(resumed_text) or "播放" in str(resumed_text), f"{resumed_text!r}")
cdp.eval("(function(){var b=document.querySelector('#btnStop, [data-action=stop]');if(b)b.click();return !!b})()")
time.sleep(1.5)
stop_text = cdp.eval("(function(){var d=document.querySelector('#chipPlayState');return d?d.textContent.trim():''})()")
check("停止后状态回到未播放", "未播放" in str(stop_text) or "已停止" in str(stop_text), f"{stop_text!r}")
cdp.shot(out_dir / "04-stopped.png")
# ---- 6. 底图切换 ----
options = cdp.eval(
"(function(){var s=document.querySelector('#tileSourceSelect');"
"return JSON.stringify(Array.from(s.options).map(function(o){return o.value}))})()"
)
opts = json.loads(options or "[]")
check("底图下拉有多个瓦片源", len(opts) >= 3, str(opts))
if len(opts) >= 2:
cdp.eval(f"(function(){{var s=document.querySelector('#tileSourceSelect');"
f"s.value={json.dumps(opts[1])};s.dispatchEvent(new Event('change',{{bubbles:true}}));return s.value}})()")
time.sleep(3.0)
current = cdp.eval("(function(){var s=document.querySelector('#tileSourceSelect');return s.value})()")
check("切换底图生效", current == opts[1], f"{current!r} vs {opts[1]!r}")
cdp.shot(out_dir / "05-tilesource.png")
# ---- 7. 日志面板（先展开再看内容） ----
cdp.eval("(function(){var b=document.querySelector('#btnToggleLog');if(b)b.click();return !!b})()")
time.sleep(1.2)
log_chars = cdp.eval("(function(){var p=document.querySelector('#logBody');return p?(p.innerText||'').length:0})()")
check("日志面板能展开且有内容", int(log_chars or 0) > 30, f"字符数={log_chars}")
log_text = cdp.eval("(function(){var p=document.querySelector('#logBody');return p?(p.innerText||'').slice(0,200):''})()")
check("日志内容为中文可读", bool(log_text) and any("\u4e00" <= ch <= "\u9fff" for ch in str(log_text)),
str(log_text)[:100])
cdp.shot(out_dir / "06-logs.png")
# ---- 8. 收集页面控制台报错 ----
errors = []
for event in cdp.events:
method = event.get("method")
if method == "Runtime.exceptionThrown":
detail = event["params"]["exceptionDetails"]
errors.append(f"{detail.get('text')} {(detail.get('exception') or {}).get('description', '')[:200]}")
elif method == "Runtime.consoleAPICalled" and event["params"].get("type") == "error":
parts = [str(a.get("value", a.get("description", "")))[:160] for a in event["params"].get("args", [])]
errors.append("console.error: " + " ".join(parts))
check("页面无 JS 报错", not errors, " | ".join(errors[:4]))
if errors:
print("\n 页面报错明细：")
for item in errors[:10]:
print(" -", item[:240])
except Exception as exc:
FAIL += 1
FAILURES.append(f"验收脚本异常 {exc}")
print(f" \033[31m✗\033[0m 验收脚本异常：{exc.__class__.__name__}: {exc}")
import traceback
traceback.print_exc()
finally:
if cdp:
cdp.close()
if not args.keep_open:
try:
os.killpg(os.getpgid(process.pid), 15)
except Exception:
process.terminate()
print(f"\n{'=' * 56}")
print(f"图形界面验收：通过 \033[32m{PASS}\033[0m 项，失败 \033[31m{FAIL}\033[0m 项")
print(f"截图目录：{out_dir}")
if FAILURES:
print("失败明细：")
for item in FAILURES:
print(" -", item[:220])
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())