142 lines
5.2 KiB
Python
142 lines
5.2 KiB
Python
"""
|
|
芽说英语 (SpeakSprout) - 同步服务自动化端到端测试
|
|
"""
|
|
import os
|
|
import tempfile
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
# Use temporary database for tests
|
|
temp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
os.environ["DATABASE_PATH"] = temp_db.name
|
|
os.environ["JWT_SECRET"] = "test-secret-key-1234567890-secure-32b"
|
|
|
|
from main import app, init_db
|
|
|
|
async def test_full_sync_flow():
|
|
await init_db()
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
# 1. Health check
|
|
res = await ac.get("/api/v1/health")
|
|
assert res.status_code == 200
|
|
assert res.json()["status"] == "ok"
|
|
|
|
# 2. Register
|
|
reg_res = await ac.post("/api/v1/auth/register", json={
|
|
"username": "test_user_01",
|
|
"password": "Password123!",
|
|
"device_name": "Test iPhone"
|
|
})
|
|
assert reg_res.status_code == 200
|
|
reg_data = reg_res.json()
|
|
assert reg_data["code"] == 0
|
|
token = reg_data["data"]["token"]
|
|
user_id = reg_data["data"]["user_id"]
|
|
assert token and user_id
|
|
|
|
# 3. Duplicate register should fail
|
|
dup_res = await ac.post("/api/v1/auth/register", json={
|
|
"username": "test_user_01",
|
|
"password": "Password123!"
|
|
})
|
|
assert dup_res.status_code == 400
|
|
|
|
# 4. Login
|
|
login_res = await ac.post("/api/v1/auth/login", json={
|
|
"username": "test_user_01",
|
|
"password": "Password123!",
|
|
"device_name": "Test Mac"
|
|
})
|
|
assert login_res.status_code == 200
|
|
assert login_res.json()["data"]["token"]
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# 5. Push progress from Device 1 (iPhone)
|
|
push_1 = await ac.post("/api/v1/sync/push", headers=headers, json={
|
|
"client_time": "2026-09-16T10:00:00Z",
|
|
"device_name": "Test iPhone",
|
|
"progress": {
|
|
"active_lesson_id": "a0-02",
|
|
"completed_lesson_ids": ["a0-01"],
|
|
"completed_segment_ids": ["a0-01-a", "a0-01-b"],
|
|
"active_step": "speaking",
|
|
"streak_days": 1,
|
|
"updated_at": "2026-09-16T10:00:00Z"
|
|
},
|
|
"mastery_updates": [
|
|
{
|
|
"item_id": "A0-W01",
|
|
"checkpoint": 1,
|
|
"status": "learning",
|
|
"due_at": "2026-09-17T10:00:00Z",
|
|
"successful_reviews": 1,
|
|
"attempts": 1,
|
|
"updated_at": "2026-09-16T10:00:00Z"
|
|
}
|
|
],
|
|
"profile": {
|
|
"daily_minutes": 20,
|
|
"show_chinese_hints": True,
|
|
"updated_at": "2026-09-16T10:00:00Z"
|
|
}
|
|
})
|
|
assert push_1.status_code == 200
|
|
assert push_1.json()["data"]["synced_mastery_count"] == 1
|
|
|
|
# 6. Push progress from Device 2 (Mac) with non-overlapping lesson and higher checkpoint
|
|
push_2 = await ac.post("/api/v1/sync/push", headers=headers, json={
|
|
"client_time": "2026-09-16T12:00:00Z",
|
|
"device_name": "Test Mac",
|
|
"progress": {
|
|
"active_lesson_id": "a0-03",
|
|
"completed_lesson_ids": ["a0-02"],
|
|
"completed_segment_ids": ["a0-02-a"],
|
|
"active_step": "dialogue",
|
|
"streak_days": 2,
|
|
"updated_at": "2026-09-16T12:00:00Z"
|
|
},
|
|
"mastery_updates": [
|
|
{
|
|
"item_id": "A0-W01",
|
|
"checkpoint": 2, # Upgraded checkpoint
|
|
"status": "reviewing",
|
|
"due_at": "2026-09-19T12:00:00Z",
|
|
"successful_reviews": 2,
|
|
"attempts": 2,
|
|
"updated_at": "2026-09-16T12:00:00Z"
|
|
},
|
|
{
|
|
"item_id": "A0-W02",
|
|
"checkpoint": 1,
|
|
"status": "learning",
|
|
"due_at": "2026-09-17T12:00:00Z",
|
|
"successful_reviews": 1,
|
|
"attempts": 1,
|
|
"updated_at": "2026-09-16T12:00:00Z"
|
|
}
|
|
]
|
|
})
|
|
assert push_2.status_code == 200
|
|
|
|
# 7. Pull merged progress
|
|
pull_res = await ac.get("/api/v1/sync/pull", headers=headers)
|
|
assert pull_res.status_code == 200
|
|
p_data = pull_res.json()["data"]
|
|
|
|
# Verify Union of completed lessons: ["a0-01", "a0-02"]
|
|
assert "a0-01" in p_data["progress"]["completed_lesson_ids"]
|
|
assert "a0-02" in p_data["progress"]["completed_lesson_ids"]
|
|
assert p_data["progress"]["active_lesson_id"] == "a0-03"
|
|
|
|
# Verify Mastery items have merged highest checkpoint
|
|
mastery_map = {item["item_id"]: item for item in p_data["mastery_updates"]}
|
|
assert mastery_map["A0-W01"]["checkpoint"] == 2
|
|
assert mastery_map["A0-W02"]["checkpoint"] == 1
|
|
|
|
print("\n✅ All Backend Sync API tests passed with flying colors!")
|
|
|
|
if __name__ == "__main__":
|
|
import asyncio
|
|
asyncio.run(test_full_sync_flow())
|