john
1aaef71f52
Document-driven MVP with FastAPI backend, Vue H5, WeChat mini shell, product demo, and Docker dev stack. Co-authored-by: Cursor <cursoragent@cursor.com>
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
def test_local_direct_upload(client, auth_headers):
|
|
child_id = client.post(
|
|
"/api/children",
|
|
headers=auth_headers,
|
|
json={"name": "小直传", "birthday": "2016-03-10", "gender": "female"},
|
|
).json()["data"]["id"]
|
|
|
|
token_resp = client.post(
|
|
"/api/videos/upload-token",
|
|
headers=auth_headers,
|
|
json={
|
|
"childId": child_id,
|
|
"fileName": "direct.mp4",
|
|
"contentType": "video/mp4",
|
|
"size": 128,
|
|
},
|
|
)
|
|
assert token_resp.status_code == 200
|
|
payload = token_resp.json()["data"]
|
|
assert payload["storage"] == "local"
|
|
assert payload["method"] == "PUT"
|
|
assert payload["uploadToken"]
|
|
|
|
put_resp = client.put(
|
|
"/api/videos/direct-upload",
|
|
headers={
|
|
"X-Upload-Token": payload["uploadToken"],
|
|
"X-Object-Key": payload["objectKey"],
|
|
"Content-Type": "video/mp4",
|
|
},
|
|
content=b"\x00\x00\x00\x18ftypmp42",
|
|
)
|
|
assert put_resp.status_code == 204
|
|
|
|
bad_put = client.put(
|
|
"/api/videos/direct-upload",
|
|
headers={
|
|
"X-Upload-Token": "invalid-token",
|
|
"X-Object-Key": payload["objectKey"],
|
|
},
|
|
content=b"bad",
|
|
)
|
|
assert bad_put.status_code == 403
|
|
|
|
video_resp = client.post(
|
|
"/api/videos",
|
|
headers=auth_headers,
|
|
json={
|
|
"childId": child_id,
|
|
"objectKey": payload["objectKey"],
|
|
"scene": "front_posture",
|
|
},
|
|
)
|
|
assert video_resp.status_code == 201
|
|
video_id = video_resp.json()["data"]["id"]
|
|
|
|
task_resp = client.post(
|
|
"/api/analysis/tasks",
|
|
headers={**auth_headers, "Idempotency-Key": "sprint4-direct-upload"},
|
|
json={
|
|
"childId": child_id,
|
|
"videoId": video_id,
|
|
"taskType": "posture_screening",
|
|
},
|
|
)
|
|
assert task_resp.status_code == 201
|
|
task = task_resp.json()["data"]
|
|
assert task["status"] == "SUCCEEDED"
|
|
assert task["reportId"]
|
|
|
|
report_resp = client.get(f"/api/reports/{task['reportId']}", headers=auth_headers)
|
|
assert report_resp.status_code == 200
|
|
|
|
|
|
def test_pose_analyzer_mock():
|
|
from pathlib import Path
|
|
|
|
from app.ai.pose_analyzer import analyze_task
|
|
|
|
out = analyze_task("posture_screening", None)
|
|
assert out["engine"] == "mock"
|
|
assert out["report"]
|
|
|
|
missing = analyze_task("posture_screening", Path("/tmp/does-not-exist.mp4"))
|
|
assert missing["engine"] == "mock"
|