701 lines
29 KiB
Python
701 lines
29 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
from uuid import UUID
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
os.environ.setdefault("DODO_DATABASE_URL", "sqlite+aiosqlite:////tmp/dodo-health-test.db")
|
|
os.environ.setdefault("DODO_AUTO_CREATE_SCHEMA", "true")
|
|
|
|
from backend.main import app
|
|
|
|
|
|
def test_health_live():
|
|
with TestClient(app) as client:
|
|
response = client.get("/health/live")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
|
|
|
|
def test_setup_status_starts_uninitialized(client):
|
|
response = client.get("/api/v1/setup/status")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"initialized": False}
|
|
|
|
|
|
def test_initialize_login_and_create_task(client):
|
|
initialized = client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
assert initialized.status_code == 201
|
|
assert initialized.json()["username"] == "owner"
|
|
assert "dodo_session" in initialized.cookies
|
|
|
|
folder = client.post("/api/v1/folders", json={"name": "工作"})
|
|
assert folder.status_code == 201
|
|
|
|
task_list = client.post(
|
|
"/api/v1/lists", json={"name": "本周", "folder_id": folder.json()["id"]}
|
|
)
|
|
assert task_list.status_code == 201
|
|
|
|
task = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "完成 dodo 第一阶段", "list_id": task_list.json()["id"], "priority": 3},
|
|
)
|
|
assert task.status_code == 201
|
|
assert task.json()["title"] == "完成 dodo 第一阶段"
|
|
|
|
tasks = client.get("/api/v1/tasks")
|
|
assert tasks.status_code == 200
|
|
assert len(tasks.json()["items"]) == 1
|
|
|
|
|
|
def test_initialize_is_closed_after_first_user(client):
|
|
payload = {"username": "owner", "password": "correct horse battery staple"}
|
|
assert client.post("/api/v1/setup/initialize", json=payload).status_code == 201
|
|
response = client.post("/api/v1/setup/initialize", json=payload)
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_unauthenticated_task_access_is_rejected(client):
|
|
response = client.get("/api/v1/tasks")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_task_can_be_updated_completed_and_soft_deleted(client):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
task = client.post(
|
|
"/api/v1/tasks", json={"title": "旧标题", "list_id": inbox["id"]}
|
|
).json()
|
|
|
|
updated = client.patch(
|
|
f"/api/v1/tasks/{task['id']}",
|
|
json={"title": "新标题", "completed": True, "version": task["version"]},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["title"] == "新标题"
|
|
assert updated.json()["completed"] is True
|
|
assert updated.json()["version"] == 2
|
|
|
|
conflict = client.patch(
|
|
f"/api/v1/tasks/{task['id']}",
|
|
json={"title": "冲突标题", "version": task["version"]},
|
|
)
|
|
assert conflict.status_code == 409
|
|
|
|
deleted = client.delete(f"/api/v1/tasks/{task['id']}")
|
|
assert deleted.status_code == 204
|
|
assert client.get("/api/v1/tasks").json()["items"] == []
|
|
|
|
|
|
def test_reorder_tasks_persists_top_level_and_subtask_order(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
first = client.post("/api/v1/tasks", json={"title": "第一个", "list_id": inbox["id"]}).json()
|
|
second = client.post("/api/v1/tasks", json={"title": "第二个", "list_id": inbox["id"]}).json()
|
|
child_a = client.post(
|
|
"/api/v1/tasks", json={"title": "子任务 A", "list_id": inbox["id"], "parent_id": first["id"]}
|
|
).json()
|
|
child_b = client.post(
|
|
"/api/v1/tasks", json={"title": "子任务 B", "list_id": inbox["id"], "parent_id": first["id"]}
|
|
).json()
|
|
|
|
top_level = client.put("/api/v1/tasks/reorder", json={"task_ids": [second["id"], first["id"]]})
|
|
assert top_level.status_code == 204
|
|
children = client.put("/api/v1/tasks/reorder", json={"task_ids": [child_b["id"], child_a["id"]]})
|
|
assert children.status_code == 204
|
|
|
|
client.post("/api/v1/tasks", json={"title": "第三个", "list_id": inbox["id"]})
|
|
listed = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 1}).json()["items"]
|
|
assert [task["title"] for task in listed] == ["第二个", "第一个", "第三个"]
|
|
assert [task["title"] for task in listed[1]["subtasks"]] == ["子任务 B", "子任务 A"]
|
|
|
|
|
|
def test_reorder_tasks_rejects_mixed_parent_scopes_and_lists(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
parent = client.post("/api/v1/tasks", json={"title": "父任务", "list_id": inbox["id"]}).json()
|
|
child = client.post(
|
|
"/api/v1/tasks", json={"title": "子任务", "list_id": inbox["id"], "parent_id": parent["id"]}
|
|
).json()
|
|
|
|
response = client.put("/api/v1/tasks/reorder", json={"task_ids": [parent["id"], child["id"]]})
|
|
assert response.status_code == 400
|
|
|
|
other = client.post("/api/v1/lists", json={"name": "其他清单"}).json()
|
|
other_task = client.post("/api/v1/tasks", json={"title": "其他任务", "list_id": other["id"]}).json()
|
|
response = client.put("/api/v1/tasks/reorder", json={"task_ids": [parent["id"], other_task["id"]]})
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_task_update_rejects_null_title(client):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
task = client.post("/api/v1/tasks", json={"title": "任务", "list_id": inbox["id"]}).json()
|
|
response = client.patch(
|
|
f"/api/v1/tasks/{task['id']}", json={"title": None, "version": task["version"]}
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_subtask_parent_must_belong_to_same_user_and_list(client):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
other = client.post("/api/v1/lists", json={"name": "其他"}).json()
|
|
parent = client.post("/api/v1/tasks", json={"title": "父任务", "list_id": inbox["id"]}).json()
|
|
response = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "子任务", "list_id": other["id"], "parent_id": parent["id"]},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_logout_revokes_current_session(client):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
assert client.get("/api/v1/me").status_code == 200
|
|
assert client.post("/api/v1/auth/logout").status_code == 204
|
|
assert client.get("/api/v1/me").status_code == 401
|
|
|
|
|
|
def test_revoke_other_sessions_keeps_current_session(client):
|
|
password = "correct horse battery staple"
|
|
client.post("/api/v1/setup/initialize", json={"username": "owner", "password": password})
|
|
other = type(client)(client.app)
|
|
third = type(client)(client.app)
|
|
try:
|
|
assert other.post("/api/v1/auth/login", json={"username": "owner", "password": password}).status_code == 200
|
|
assert third.post("/api/v1/auth/login", json={"username": "owner", "password": password}).status_code == 200
|
|
assert len(client.get("/api/v1/sessions").json()) == 3
|
|
|
|
assert client.delete("/api/v1/sessions/others").status_code == 204
|
|
|
|
remaining = client.get("/api/v1/sessions")
|
|
assert remaining.status_code == 200
|
|
assert len(remaining.json()) == 1
|
|
assert remaining.json()[0]["current"] is True
|
|
assert client.get("/api/v1/me").status_code == 200
|
|
assert other.get("/api/v1/me").status_code == 401
|
|
assert third.get("/api/v1/me").status_code == 401
|
|
finally:
|
|
other.close()
|
|
third.close()
|
|
|
|
|
|
def test_change_password_checks_current_password_and_revokes_other_sessions(client):
|
|
old_password = "correct horse battery staple"
|
|
new_password = "new correct horse battery staple"
|
|
client.post("/api/v1/setup/initialize", json={"username": "owner", "password": old_password})
|
|
other = type(client)(client.app)
|
|
try:
|
|
assert other.post("/api/v1/auth/login", json={"username": "owner", "password": old_password}).status_code == 200
|
|
wrong = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "wrong password", "new_password": new_password},
|
|
)
|
|
assert wrong.status_code == 400
|
|
assert wrong.json()["detail"] == "当前密码不正确"
|
|
|
|
changed = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": old_password, "new_password": new_password},
|
|
)
|
|
assert changed.status_code == 204
|
|
assert client.get("/api/v1/me").status_code == 200
|
|
assert other.get("/api/v1/me").status_code == 401
|
|
assert other.post("/api/v1/auth/login", json={"username": "owner", "password": old_password}).status_code == 401
|
|
assert other.post("/api/v1/auth/login", json={"username": "owner", "password": new_password}).status_code == 200
|
|
finally:
|
|
other.close()
|
|
|
|
|
|
def test_change_password_validates_new_password(client):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
too_short = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "correct horse battery staple", "new_password": "short"},
|
|
)
|
|
assert too_short.status_code == 422
|
|
same = client.post(
|
|
"/api/v1/auth/change-password",
|
|
json={"current_password": "correct horse battery staple", "new_password": "correct horse battery staple"},
|
|
)
|
|
assert same.status_code == 422
|
|
|
|
|
|
def initialized_client(client):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
return client
|
|
|
|
|
|
def test_folder_and_list_lifecycle(client):
|
|
client = initialized_client(client)
|
|
folder = client.post("/api/v1/folders", json={"name": "工作"}).json()
|
|
renamed = client.patch(f"/api/v1/folders/{folder['id']}", json={"name": "生活"})
|
|
assert renamed.status_code == 200
|
|
assert renamed.json()["name"] == "生活"
|
|
|
|
task_list = client.post(
|
|
"/api/v1/lists", json={"name": "采购", "folder_id": folder["id"]}
|
|
).json()
|
|
renamed_list = client.patch(f"/api/v1/lists/{task_list['id']}", json={"name": "购物"})
|
|
assert renamed_list.status_code == 200
|
|
assert renamed_list.json()["name"] == "购物"
|
|
assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
|
|
assert client.delete(f"/api/v1/folders/{folder['id']}").status_code == 204
|
|
|
|
|
|
def test_task_search_subtasks_and_recycle_bin(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
parent = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "购买咖啡豆", "description": "手冲用", "list_id": inbox["id"]},
|
|
).json()
|
|
child = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "比较价格", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
)
|
|
assert child.status_code == 201
|
|
detail = client.get(f"/api/v1/tasks/{parent['id']}")
|
|
assert detail.status_code == 200
|
|
assert "tags" not in detail.json()
|
|
assert len(detail.json()["subtasks"]) == 1
|
|
listed = client.get("/api/v1/tasks", params={"q": "咖啡"}).json()["items"]
|
|
assert len(listed) == 1
|
|
assert "tags" not in listed[0]
|
|
assert listed[0]["subtasks"][0]["title"] == "比较价格"
|
|
|
|
assert client.delete(f"/api/v1/tasks/{parent['id']}").status_code == 204
|
|
trash = client.get("/api/v1/trash").json()["items"]
|
|
assert len(trash) == 1
|
|
assert client.post(f"/api/v1/tasks/{parent['id']}/restore").status_code == 200
|
|
|
|
|
|
def test_batch_complete_and_move(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
other = client.post("/api/v1/lists", json={"name": "稍后"}).json()
|
|
ids = [
|
|
client.post("/api/v1/tasks", json={"title": f"任务{i}", "list_id": inbox["id"]}).json()["id"]
|
|
for i in range(2)
|
|
]
|
|
response = client.post(
|
|
"/api/v1/tasks/batch", json={
|
|
"task_ids": ids,
|
|
"completed": True,
|
|
"list_id": other["id"],
|
|
"versions": {task_id: 1 for task_id in ids},
|
|
}
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["updated"] == 2
|
|
rows = client.get("/api/v1/tasks").json()["items"]
|
|
assert all(row["completed"] and row["list_id"] == other["id"] for row in rows)
|
|
|
|
|
|
def test_batch_move_rejects_standalone_subtask_and_delete_cascades(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
other = client.post("/api/v1/lists", json={"name": "稍后"}).json()
|
|
parent = client.post("/api/v1/tasks", json={"title": "父", "list_id": inbox["id"]}).json()
|
|
child = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "子", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
).json()
|
|
|
|
moved_child = client.post(
|
|
"/api/v1/tasks/batch", json={"task_ids": [child["id"]], "list_id": other["id"]}
|
|
)
|
|
assert moved_child.status_code == 400
|
|
|
|
deleted_parent = client.post(
|
|
"/api/v1/tasks/batch", json={"task_ids": [parent["id"]], "soft_delete": True}
|
|
)
|
|
assert deleted_parent.status_code == 200
|
|
assert client.get(f"/api/v1/tasks/{parent['id']}").status_code == 404
|
|
assert client.get(f"/api/v1/tasks/{child['id']}").status_code == 404
|
|
|
|
|
|
def test_inbox_is_protected_and_deleted_collections_are_hidden(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
assert client.patch(f"/api/v1/lists/{inbox['id']}", json={"name": "x"}).status_code == 409
|
|
assert client.delete(f"/api/v1/lists/{inbox['id']}").status_code == 409
|
|
|
|
folder = client.post("/api/v1/folders", json={"name": "项目"}).json()
|
|
task_list = client.post(
|
|
"/api/v1/lists", json={"name": "待办", "folder_id": folder["id"]}
|
|
).json()
|
|
task = client.post(
|
|
"/api/v1/tasks", json={"title": "保留任务", "list_id": task_list["id"]}
|
|
).json()
|
|
completed_task = client.post(
|
|
"/api/v1/tasks", json={"title": "已完成任务", "list_id": task_list["id"]}
|
|
).json()
|
|
completed_task = client.patch(
|
|
f"/api/v1/tasks/{completed_task['id']}",
|
|
json={"completed": True, "version": completed_task["version"]},
|
|
).json()
|
|
assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
|
|
assert all(row["id"] != task_list["id"] for row in client.get("/api/v1/lists").json())
|
|
assert client.get("/api/v1/tasks", params={"q": "保留任务"}).json()["items"] == []
|
|
assert client.get(f"/api/v1/tasks/{task['id']}").status_code == 404
|
|
archived = client.get("/api/v1/lists", params={"archived": True})
|
|
assert archived.status_code == 200
|
|
assert [row["id"] for row in archived.json()] == [task_list["id"]]
|
|
assert client.post(f"/api/v1/lists/{task_list['id']}/restore").status_code == 200
|
|
assert any(row["id"] == task_list["id"] for row in client.get("/api/v1/lists").json())
|
|
restored = client.get("/api/v1/tasks", params={"list_id": task_list["id"]}).json()["items"]
|
|
assert [row["id"] for row in restored] == [task["id"], completed_task["id"]]
|
|
assert restored[1]["completed"] is True
|
|
assert client.delete(f"/api/v1/folders/{folder['id']}").status_code == 204
|
|
assert client.get("/api/v1/folders").json() == []
|
|
|
|
|
|
def test_nested_subtasks_are_rejected(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
parent = client.post("/api/v1/tasks", json={"title": "父", "list_id": inbox["id"]}).json()
|
|
child = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "子", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
).json()
|
|
nested = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "孙", "list_id": inbox["id"], "parent_id": child["id"]},
|
|
)
|
|
assert nested.status_code == 400
|
|
|
|
|
|
|
|
def test_restore_replace_recovers_habits_and_task_links_without_tags(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "备份任务", "list_id": inbox["id"]},
|
|
).json()
|
|
client.post(
|
|
"/api/v1/habits",
|
|
json={"name": "俯卧撑", "kind": "boolean", "schedule_type": "daily"},
|
|
).json()
|
|
exported = client.get("/api/v1/export")
|
|
assert exported.status_code == 200
|
|
assert "tags" not in exported.json()
|
|
assert "task_tags" not in exported.json()
|
|
|
|
exported_csv = client.get("/api/v1/export.csv")
|
|
assert exported_csv.status_code == 200
|
|
assert exported_csv.headers["content-type"].startswith("text/csv")
|
|
assert "dodo-export.csv" in exported_csv.headers["content-disposition"]
|
|
assert exported_csv.content.startswith(b"\xef\xbb\xbfentity,data")
|
|
|
|
restored_csv = client.post(
|
|
"/api/v1/restore.csv?mode=merge",
|
|
files={"file": ("dodo-export.csv", exported_csv.content, "text/csv")},
|
|
)
|
|
assert restored_csv.status_code == 200
|
|
|
|
client.post(
|
|
"/api/v1/habits",
|
|
json={"name": "深蹲", "kind": "boolean", "schedule_type": "daily"},
|
|
)
|
|
restored = client.post("/api/v1/restore?mode=replace", json=exported.json())
|
|
assert restored.status_code == 200
|
|
|
|
assert client.get("/api/v1/tags").status_code == 404
|
|
habits = client.get("/api/v1/habits").json()
|
|
assert [row["name"] for row in habits] == ["俯卧撑"]
|
|
listed = client.get("/api/v1/tasks", params={"q": "备份任务"}).json()["items"]
|
|
assert "tags" not in listed[0]
|
|
|
|
|
|
def test_trash_cursor_paginates_more_than_fifty_items(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
task_ids = [
|
|
client.post("/api/v1/tasks", json={"title": f"回收任务 {index}", "list_id": inbox["id"]}).json()["id"]
|
|
for index in range(52)
|
|
]
|
|
for task_id in task_ids:
|
|
assert client.delete(f"/api/v1/tasks/{task_id}").status_code == 204
|
|
|
|
first = client.get("/api/v1/trash", params={"limit": 50})
|
|
assert first.status_code == 200
|
|
first_page = first.json()
|
|
assert len(first_page["items"]) == 50
|
|
assert first_page["next_cursor"]
|
|
|
|
second = client.get("/api/v1/trash", params={"limit": 50, "cursor": first_page["next_cursor"]})
|
|
assert second.status_code == 200
|
|
assert len(second.json()["items"]) == 2
|
|
assert second.json()["next_cursor"] is None
|
|
|
|
|
|
def test_recycle_bin_restore_and_permanent_delete_include_subtasks(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
parent = client.post("/api/v1/tasks", json={"title": "父", "list_id": inbox["id"]}).json()
|
|
client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "子", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
)
|
|
client.delete(f"/api/v1/tasks/{parent['id']}")
|
|
assert len(client.get("/api/v1/trash").json()["items"]) == 1
|
|
restored = client.post(f"/api/v1/tasks/{parent['id']}/restore")
|
|
assert restored.status_code == 200
|
|
assert len(client.get(f"/api/v1/tasks/{parent['id']}").json()["subtasks"]) == 1
|
|
client.delete(f"/api/v1/tasks/{parent['id']}")
|
|
assert client.delete(f"/api/v1/trash/{parent['id']}").status_code == 204
|
|
assert client.post(f"/api/v1/tasks/{parent['id']}/restore").status_code == 404
|
|
|
|
|
|
def test_batch_supports_due_date_and_soft_delete_atomically(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
ids = [
|
|
client.post("/api/v1/tasks", json={"title": str(i), "list_id": inbox["id"]}).json()["id"]
|
|
for i in range(2)
|
|
]
|
|
due_at = "2026-09-10T08:00:00+08:00"
|
|
response = client.post(
|
|
"/api/v1/tasks/batch",
|
|
json={"task_ids": ids, "due_at": due_at},
|
|
)
|
|
assert response.status_code == 200 and response.json()["updated"] == 2
|
|
assert all(client.get(f"/api/v1/tasks/{task_id}").json()["due_at"] is not None for task_id in ids)
|
|
|
|
parent = client.post("/api/v1/tasks", json={"title": "父", "list_id": inbox["id"]}).json()
|
|
client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "子", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
)
|
|
other = client.post("/api/v1/lists", json={"name": "批量目标"}).json()
|
|
assert client.post(
|
|
"/api/v1/tasks/batch", json={"task_ids": [parent["id"]], "list_id": other["id"]}
|
|
).status_code == 200
|
|
moved = client.get(f"/api/v1/tasks/{parent['id']}").json()
|
|
assert moved["subtasks"][0]["list_id"] == other["id"]
|
|
|
|
failed = client.post(
|
|
"/api/v1/tasks/batch",
|
|
json={
|
|
"task_ids": [ids[0], "00000000-0000-0000-0000-000000000001"],
|
|
"completed": True,
|
|
"versions": {ids[0]: 1, "00000000-0000-0000-0000-000000000001": 1},
|
|
},
|
|
)
|
|
assert failed.status_code == 404
|
|
assert client.get(f"/api/v1/tasks/{ids[0]}").json()["completed"] is False
|
|
assert client.post("/api/v1/tasks/batch", json={"task_ids": ids, "soft_delete": True}).status_code == 200
|
|
assert len(client.get("/api/v1/trash").json()["items"]) == 2
|
|
|
|
|
|
def test_cursor_pagination_is_stable_and_rejects_bad_cursor(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
fixtures = [
|
|
("未完成较晚", "2026-09-12T08:00:00Z", False),
|
|
("未完成较早 A", "2026-09-10T08:00:00Z", False),
|
|
("未完成较早 B", "2026-09-10T08:00:00Z", False),
|
|
("未完成无日期", None, False),
|
|
("已完成较早", "2026-09-09T08:00:00Z", True),
|
|
("已完成无日期", None, True),
|
|
]
|
|
for title, due_at, completed in fixtures:
|
|
task = client.post(
|
|
"/api/v1/tasks", json={"title": title, "list_id": inbox["id"], "due_at": due_at}
|
|
).json()
|
|
if completed:
|
|
response = client.patch(
|
|
f"/api/v1/tasks/{task['id']}", json={"completed": True, "version": task["version"]}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
pages = []
|
|
cursor = None
|
|
while True:
|
|
params = {"limit": 2}
|
|
if cursor:
|
|
params["cursor"] = cursor
|
|
page = client.get("/api/v1/tasks", params=params).json()
|
|
pages.append(page)
|
|
cursor = page["next_cursor"]
|
|
if cursor is None:
|
|
break
|
|
|
|
rows = [row for page in pages for row in page["items"]]
|
|
assert [row["title"] for row in rows] == [
|
|
"未完成较早 A", "未完成较早 B", "未完成较晚", "未完成无日期", "已完成较早", "已完成无日期"
|
|
]
|
|
assert len({row["id"] for row in rows}) == len(fixtures)
|
|
assert all(page["total"] == len(fixtures) for page in pages)
|
|
assert client.get("/api/v1/tasks", params={"cursor": "broken"}).status_code == 422
|
|
|
|
|
|
def _encoded_cursor_payload(keys):
|
|
return base64.urlsafe_b64encode(json.dumps({"v": 2, "keys": keys}).encode()).decode().rstrip("=")
|
|
|
|
|
|
def test_active_cursor_rejects_malicious_key_matrix_with_stable_422(client):
|
|
client = initialized_client(client)
|
|
task_id = "00000000-0000-0000-0000-000000000001"
|
|
created_at = "2026-09-10T08:00:00Z"
|
|
valid = [0, 1, None, 0, created_at, task_id]
|
|
invalid_keys = [
|
|
[True, 1, None, 0, created_at, task_id],
|
|
[0, False, "2026-09-10T08:00:00Z", 0, created_at, task_id],
|
|
["0", 1, None, 0, created_at, task_id],
|
|
[0, 1.0, None, 0, created_at, task_id],
|
|
[9, 1, None, 0, created_at, task_id],
|
|
[0, 9, None, 0, created_at, task_id],
|
|
[0, 0, None, 0, created_at, task_id],
|
|
[0, 1, "2026-09-10T08:00:00Z", 0, created_at, task_id],
|
|
[0, 0, "2026-09-10", 0, created_at, task_id],
|
|
[0, 0, "2026-09-10T08:00:00", 0, created_at, task_id],
|
|
[0, 0, "2026-09-10T08Z", 0, created_at, task_id],
|
|
[0, 0, "9999-12-31T23:59:59-14:00", 0, created_at, task_id],
|
|
[0, 1, None, True, created_at, task_id],
|
|
[0, 1, None, "0", created_at, task_id],
|
|
[0, 1, None, -1, created_at, task_id],
|
|
[0, 1, None, 2**63, created_at, task_id],
|
|
[0, 1, None, 0, "2026-09-10", task_id],
|
|
[0, 1, None, 0, "2026-09-10T08:00:00", task_id],
|
|
[0, 1, None, 0, "9999-12-31T23:59:59-14:00", task_id],
|
|
[0, 1, None, 0, created_at, True],
|
|
[0, 1, None, 0, created_at, "not-a-uuid"],
|
|
valid[:-1],
|
|
[*valid, "extra"],
|
|
]
|
|
for keys in invalid_keys:
|
|
response = client.get("/api/v1/tasks", params={"cursor": _encoded_cursor_payload(keys)})
|
|
assert response.status_code == 422, keys
|
|
assert response.json()["detail"] == "无效的游标"
|
|
|
|
|
|
def test_task_details_do_not_include_foreign_or_wrong_list_subtasks(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
other_list = client.post("/api/v1/lists", json={"name": "其他清单"}).json()
|
|
parent = client.post("/api/v1/tasks", json={"title": "父", "list_id": inbox["id"]}).json()
|
|
|
|
async def inject_malformed_children():
|
|
from sqlalchemy import select
|
|
|
|
from backend.db import get_engine
|
|
from backend.models import Task, User
|
|
|
|
session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
|
|
async with session_factory() as db:
|
|
owner = await db.scalar(select(User).where(User.username == "owner"))
|
|
foreign = User(username="foreign", password_hash="unused")
|
|
db.add(foreign)
|
|
await db.flush()
|
|
db.add_all([
|
|
Task(
|
|
user_id=foreign.id,
|
|
list_id=UUID(inbox["id"]),
|
|
parent_id=UUID(parent["id"]),
|
|
title="FOREIGN SECRET",
|
|
),
|
|
Task(
|
|
user_id=owner.id,
|
|
list_id=UUID(other_list["id"]),
|
|
parent_id=UUID(parent["id"]),
|
|
title="WRONG LIST SECRET",
|
|
),
|
|
])
|
|
await db.commit()
|
|
|
|
import asyncio
|
|
asyncio.run(inject_malformed_children())
|
|
|
|
detail = client.get(f"/api/v1/tasks/{parent['id']}")
|
|
assert detail.status_code == 200
|
|
assert detail.json()["subtasks"] == []
|
|
|
|
|
|
def test_task_details_use_same_due_and_completion_ordering(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
parent = client.post("/api/v1/tasks", json={"title": "父", "list_id": inbox["id"]}).json()
|
|
fixtures = [
|
|
("子无日期", None, False),
|
|
("子较晚", "2026-09-12T08:00:00Z", False),
|
|
("子较早 A", "2026-09-10T08:00:00Z", False),
|
|
("子较早 B", "2026-09-10T08:00:00Z", False),
|
|
("子已完成", "2026-09-09T08:00:00Z", True),
|
|
]
|
|
for title, due_at, completed in fixtures:
|
|
child = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": title, "list_id": inbox["id"], "parent_id": parent["id"], "due_at": due_at},
|
|
).json()
|
|
if completed:
|
|
client.patch(f"/api/v1/tasks/{child['id']}", json={"completed": True, "version": child["version"]})
|
|
|
|
detail = client.get(f"/api/v1/tasks/{parent['id']}").json()
|
|
assert [row["title"] for row in detail["subtasks"]] == [
|
|
"子较早 A", "子较早 B", "子较晚", "子无日期", "子已完成"
|
|
]
|
|
|
|
|
|
def test_reorder_treats_offset_equivalent_due_times_as_same_tier(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
first = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "UTC", "list_id": inbox["id"], "due_at": "2026-09-10T08:00:00Z"},
|
|
).json()
|
|
second = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "OFFSET", "list_id": inbox["id"], "due_at": "2026-09-10T16:00:00+08:00"},
|
|
).json()
|
|
|
|
response = client.put("/api/v1/tasks/reorder", json={"task_ids": [second["id"], first["id"]]})
|
|
assert response.status_code == 204
|
|
listed = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 1}).json()["items"]
|
|
assert [item["title"] for item in listed] == ["OFFSET", "UTC"]
|
|
assert {item["due_at"] for item in listed} == {"2026-09-10T08:00:00Z"}
|
|
|
|
|
|
def test_reorder_tasks_rejects_different_sort_tiers(client):
|
|
client = initialized_client(client)
|
|
inbox = client.get("/api/v1/lists").json()[0]
|
|
no_due = client.post("/api/v1/tasks", json={"title": "无日期", "list_id": inbox["id"]}).json()
|
|
due_a = client.post(
|
|
"/api/v1/tasks", json={"title": "日期 A", "list_id": inbox["id"], "due_at": "2026-09-10T08:00:00Z"}
|
|
).json()
|
|
due_b = client.post(
|
|
"/api/v1/tasks", json={"title": "日期 B", "list_id": inbox["id"], "due_at": "2026-09-11T08:00:00Z"}
|
|
).json()
|
|
completed = client.patch(
|
|
f"/api/v1/tasks/{due_a['id']}", json={"completed": True, "version": due_a["version"]}
|
|
).json()
|
|
|
|
assert client.put("/api/v1/tasks/reorder", json={"task_ids": [no_due["id"], due_b["id"]]}).status_code == 400
|
|
assert client.put("/api/v1/tasks/reorder", json={"task_ids": [completed["id"], due_b["id"]]}).status_code == 400
|