162 lines
6.6 KiB
GDScript
162 lines
6.6 KiB
GDScript
extends TestCase
|
||
|
||
## T-WIN-002 集成测试：直接调用**真实运行中**的 Web 服务端。
|
||
##
|
||
## 若服务端不可达则整组跳过（不伪装成通过）。
|
||
## 启动服务端：cd opendance-web/web-server && npm run start:prod
|
||
##
|
||
## 覆盖：
|
||
## - GET /health、/api/catalog/version、/api/catalog/export 的真实响应
|
||
## - NDJSON 关键点序列解析
|
||
## - **跨实现交叉校验**：本端 PoseFeatures 计算的 9 个关节角
|
||
## 与服务端 TypeScript 实现算出的 angles 必须一致
|
||
|
||
const ApiClient = preload("res://src/net/api_client.gd")
|
||
const CatalogTypes = preload("res://src/net/catalog_types.gd")
|
||
const KeypointSequence = preload("res://src/net/keypoint_sequence.gd")
|
||
const PoseFeatures = preload("res://src/score/pose_features.gd")
|
||
|
||
var _client: ApiClient = null
|
||
var _available := false
|
||
var _unavailable_reason := ""
|
||
|
||
|
||
func _ensure_client() -> bool:
|
||
if _client != null:
|
||
# 每次都要显式 skip：否则后续测试会静默 return，
|
||
# 被运行器判为「未执行任何断言」
|
||
if not _available:
|
||
skip("API 集成", "Web 服务端不可达（%s）" % _unavailable_reason)
|
||
return _available
|
||
|
||
var tree := Engine.get_main_loop() as SceneTree
|
||
_client = ApiClient.new("http://127.0.0.1:3000")
|
||
_client.timeout_seconds = 8.0
|
||
tree.root.add_child(_client)
|
||
|
||
var result: Dictionary = await _client.health()
|
||
_available = result["ok"] and int(result["status"]) == 200
|
||
if _available:
|
||
var data: Dictionary = result["data"]
|
||
ok(bool(data.get("database", {}).get("connected", false)), "服务端报告数据库已连接")
|
||
else:
|
||
_unavailable_reason = str(result.get("error", ""))
|
||
skip("API 集成", "Web 服务端不可达（%s）" % _unavailable_reason)
|
||
return _available
|
||
|
||
|
||
func test_health_reports_contract_version() -> void:
|
||
if not await _ensure_client():
|
||
return
|
||
var result: Dictionary = await _client.health()
|
||
is_true(result["ok"], "GET /health 成功")
|
||
var version: Dictionary = result["data"].get("contractVersion", {})
|
||
eq(str(version.get("rest-api", "")), "0.1.0", "契约 rest-api 版本为 0.1.0")
|
||
eq(str(version.get("websocket-keypoints", "")), "0.1.0", "契约 websocket-keypoints 版本为 0.1.0")
|
||
eq(str(version.get("skeleton-topology", "")), "0.1.0", "契约 skeleton-topology 版本为 0.1.0")
|
||
|
||
|
||
func test_catalog_version_returns_revision() -> void:
|
||
if not await _ensure_client():
|
||
return
|
||
var result: Dictionary = await _client.catalog_version()
|
||
is_true(result["ok"], "GET /api/catalog/version 成功")
|
||
var revision := str(result["data"].get("revision", ""))
|
||
eq(revision.length(), 64, "revision 为 64 位十六进制指纹")
|
||
|
||
|
||
func test_catalog_export_parses_into_typed_objects() -> void:
|
||
if not await _ensure_client():
|
||
return
|
||
var result: Dictionary = await _client.catalog_export()
|
||
is_true(result["ok"], "GET /api/catalog/export 成功")
|
||
|
||
var bundle: CatalogTypes.CatalogBundle = CatalogTypes.CatalogBundle.from_dict(result["data"])
|
||
not_null(bundle, "目录解析成功")
|
||
eq(bundle.songs.size(), result["data"].get("songs", []).size(), "曲目数量一致")
|
||
eq(bundle.motions.size(), result["data"].get("motions", []).size(), "动作数量一致")
|
||
eq(bundle.models.size(), result["data"].get("models", []).size(), "模型数量一致")
|
||
|
||
# 契约 §4.1：defaultModelId 由服务端解析，客户端直接用
|
||
var resolved := bundle.default_model()
|
||
if bundle.models.size() > 0:
|
||
not_null(resolved, "存在已发布模型时可解析出默认模型")
|
||
if resolved != null:
|
||
eq(resolved.status, "published", "默认模型状态为 published")
|
||
else:
|
||
is_null(resolved, "无已发布模型时默认模型为 null")
|
||
|
||
# 契约要求来源五元组齐备
|
||
for song in bundle.songs:
|
||
is_true(song.source.is_complete(), "曲目「%s」来源五元组齐备" % song.title)
|
||
for model in bundle.models:
|
||
is_true(model.source.is_complete(), "模型「%s」来源五元组齐备" % model.name)
|
||
|
||
|
||
func test_keypoint_sequence_matches_contract_and_local_math() -> void:
|
||
if not await _ensure_client():
|
||
return
|
||
|
||
var export_result: Dictionary = await _client.catalog_export()
|
||
var bundle: CatalogTypes.CatalogBundle = CatalogTypes.CatalogBundle.from_dict(export_result["data"])
|
||
|
||
var target: CatalogTypes.Motion = null
|
||
for motion in bundle.motions:
|
||
if motion.is_scorable():
|
||
target = motion
|
||
break
|
||
|
||
if target == null:
|
||
skip("关键点序列", "服务端暂无已完成预处理的动作（先上传并预处理一个 BVH）")
|
||
return
|
||
|
||
eq(target.keypoint_format, KeypointSequence.EXPECTED_FORMAT, "keypointFormat 与契约一致")
|
||
|
||
var sequence_result: Dictionary = await _client.motion_keypoints(target.id)
|
||
is_true(sequence_result["ok"], "GET /api/motions/%d/keypoints 成功" % target.id)
|
||
|
||
var sequence: KeypointSequence.ParsedSequence = KeypointSequence.parse_ndjson(str(sequence_result["data"]))
|
||
eq(sequence.skipped_lines, 0, "NDJSON 无跳过行")
|
||
eq(sequence.frame_count(), target.keypoint_frames, "帧数与 motion.keypointFrames 一致")
|
||
ok(sequence.frame_count() > 0, "序列非空")
|
||
|
||
# 每帧 33 点（一次性校验，避免刷屏式断言）
|
||
var shape_ok := true
|
||
for frame in sequence.frames:
|
||
if frame.points.size() != 33 or frame.confidence.size() != 33:
|
||
shape_ok = false
|
||
break
|
||
is_true(shape_ok, "每帧均为 33 个关键点 + 33 个置信度")
|
||
|
||
# 坐标落在 MediaPipe 归一化范围
|
||
var first: KeypointFrame.PoseFrame = sequence.frames[0]
|
||
var in_range := true
|
||
for p in first.points:
|
||
if p.x < -0.001 or p.x > 1.001 or p.y < -0.001 or p.y > 1.001:
|
||
in_range = false
|
||
break
|
||
is_true(in_range, "关键点坐标落在归一化范围")
|
||
|
||
# 帧率推导合理
|
||
var fps := sequence.estimated_fps()
|
||
ok(fps > 1.0 and fps < 240.0, "由时间戳推导的帧率合理", "实际 %.2f" % fps)
|
||
|
||
# ★ 跨实现交叉校验：服务端 TypeScript 与本端 GDScript 的关节角必须一致。
|
||
# GDScript 的 % 运算符不支持 %e，故用 str() 输出浮点。
|
||
var comparison := KeypointSequence.compare_angles(sequence, 1e-3)
|
||
ok(int(comparison["checked"]) > 0, "存在可比对的服务端关节角", "比对帧数 %d" % int(comparison["checked"]))
|
||
ok(
|
||
int(comparison["mismatched_frames"]) == 0,
|
||
"所有帧的 9 个关节角与服务端实现一致",
|
||
"不一致帧数 %d，最大偏差 %s" % [int(comparison["mismatched_frames"]), str(comparison["max_delta"])]
|
||
)
|
||
ok(float(comparison["max_delta"]) < 1e-3, "关节角最大偏差小于 1e-3", "实际 %s" % str(comparison["max_delta"]))
|
||
|
||
func test_unknown_route_returns_structured_error() -> void:
|
||
if not await _ensure_client():
|
||
return
|
||
var result: Dictionary = await _client.get_json("/api/songs/99999999")
|
||
is_false(result["ok"], "不存在的资源返回失败")
|
||
eq(int(result["status"]), 404, "状态码为 404")
|
||
ok(str(result["error"]).length() > 0, "错误信息非空", str(result["error"]))
|