from datetime import UTC, datetime, timedelta from sqlalchemy import event def boot(client): response = client.post( "/api/v1/setup/initialize", json={"username": "owner", "password": "correct horse battery staple"}, ) assert response.status_code == 201 return client.get("/api/v1/lists").json()[0] def test_bootstrap_returns_navigation_and_current_user(client): inbox = boot(client) client.post("/api/v1/folders", json={"name": "工作"}) client.post("/api/v1/tasks", json={"title": "首屏任务", "list_id": inbox["id"]}) response = client.get("/api/v1/bootstrap") assert response.status_code == 200 data = response.json() assert data["user"]["username"] == "owner" assert any(row["id"] == inbox["id"] for row in data["lists"]) assert [row["name"] for row in data["folders"]] == ["工作"] assert "tags" not in data assert data["inbox_id"] == inbox["id"] def test_task_due_date_without_time_round_trips(client): inbox = boot(client) created = client.post( "/api/v1/tasks", json={"title": "全天任务", "list_id": inbox["id"], "due_at": "2026-09-08T23:59:00Z", "due_has_time": False}, ) assert created.status_code == 201 assert created.json()["due_has_time"] is False listed = client.get("/api/v1/tasks", params={"list_id": inbox["id"]}).json()["items"] assert listed[0]["due_has_time"] is False def test_creating_task_with_recurrence_is_atomic(client): inbox = boot(client) created = client.post( "/api/v1/tasks", json={ "title": "隔周复盘", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z", "rrule": "FREQ=WEEKLY;INTERVAL=2;BYDAY=MO,FR", }, ) assert created.status_code == 201 recurrence = client.get(f"/api/v1/tasks/{created.json()['id']}/recurrence").json() assert recurrence["rrule"] == "FREQ=WEEKLY;INTERVAL=2;BYDAY=MO,FR" rejected = client.post( "/api/v1/tasks", json={"title": "缺少日期", "list_id": inbox["id"], "rrule": "FREQ=DAILY"}, ) assert rejected.status_code == 422 assert client.get("/api/v1/tasks", params={"q": "缺少日期"}).json()["items"] == [] def test_custom_recurrence_rejects_invalid_weekdays_month_days_and_until(client): inbox = boot(client) task = client.post( "/api/v1/tasks", json={"title": "自定义重复", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z"}, ).json() for rrule in ( "FREQ=WEEKLY;BYDAY=XX", "FREQ=MONTHLY;BYMONTHDAY=0,32", "FREQ=DAILY;UNTIL=not-a-date", "FREQ=DAILY;UNKNOWN=1", ): response = client.post("/api/v1/recurrences", json={"task_id": task["id"], "rrule": rrule}) assert response.status_code == 422 def test_get_recurrence_by_task_returns_rule_or_null(client): inbox = boot(client) task = client.post( "/api/v1/tasks", json={"title": "每天复盘", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z"}, ).json() empty = client.get(f"/api/v1/tasks/{task['id']}/recurrence") assert empty.status_code == 200 assert empty.json() is None created = client.post( "/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=DAILY"} ).json() found = client.get(f"/api/v1/tasks/{task['id']}/recurrence") assert found.status_code == 200 assert found.json()["id"] == created["id"] assert found.json()["rrule"] == "FREQ=DAILY" def test_export_and_restore_preserve_task_recurrence(client): inbox = boot(client) task = client.post( "/api/v1/tasks", json={"title": "每周整理", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z"}, ).json() client.post("/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=WEEKLY"}) exported = client.get("/api/v1/export").json() assert exported["recurrences"][0]["task_id"] == task["id"] restored = client.post("/api/v1/restore?mode=replace", json=exported) assert restored.status_code == 200 restored_task = client.get("/api/v1/tasks", params={"q": "每周整理"}).json()["items"][0] recurrence = client.get(f"/api/v1/tasks/{restored_task['id']}/recurrence").json() assert recurrence["rrule"] == "FREQ=WEEKLY" def test_completing_repeating_task_advances_due_date_instead_of_closing_it(client): inbox = boot(client) task = client.post( "/api/v1/tasks", json={"title": "每日复盘", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z"}, ).json() client.post("/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=DAILY"}) completed = client.patch( f"/api/v1/tasks/{task['id']}", json={"completed": True, "version": task["version"]} ) assert completed.status_code == 200 assert completed.json()["completed"] is False assert completed.json()["due_at"].replace("Z", "") == "2026-09-08T09:00:00" recurrence = client.get(f"/api/v1/tasks/{task['id']}/recurrence").json() assert recurrence["starts_at"].replace("Z", "") == "2026-09-08T09:00:00" def test_recurrence_mutations_keep_exact_timestamp_validation(client): inbox = boot(client) task = client.post( "/api/v1/tasks", json={"title": "九点站会", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z"}, ).json() recurrence = client.post( "/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=WEEKLY;BYDAY=MO;COUNT=3"} ).json() wrong_hour = client.patch( f"/api/v1/recurrences/{recurrence['id']}", params={"scope": "this", "occurrence_at": "2026-09-07T10:00:00Z"}, json={"title": "幽灵"}, ) assert wrong_hour.status_code == 422 same_instant = client.patch( f"/api/v1/recurrences/{recurrence['id']}", params={"scope": "this", "occurrence_at": "2026-09-07T17:00:00+08:00"}, json={"title": "等价时刻"}, ) assert same_instant.status_code == 200 def test_recurrence_rejects_occurrence_after_cutoff(client): inbox = boot(client) task = client.post( "/api/v1/tasks", json={"title": "每天重复", "list_id": inbox["id"], "due_at": "2026-09-01T09:00:00Z"}, ).json() recurrence = client.post( "/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=DAILY;COUNT=10"} ).json() assert client.patch( f"/api/v1/recurrences/{recurrence['id']}", params={"scope": "this-and-future", "occurrence_at": "2026-09-03T09:00:00Z"}, json={}, ).status_code == 200 assert client.patch( f"/api/v1/recurrences/{recurrence['id']}", params={"scope": "this", "occurrence_at": "2026-09-05T09:00:00Z"}, json={"title": "晚于截止"}, ).status_code == 422 def test_habit_reorder_persists_in_lists_and_grid(client): boot(client) first = client.post("/api/v1/habits", json={"name": "第一个", "kind": "boolean", "schedule_type": "daily"}).json() second = client.post("/api/v1/habits", json={"name": "第二个", "kind": "boolean", "schedule_type": "daily"}).json() response = client.put("/api/v1/habits/reorder", json={"habit_ids": [second["id"], first["id"]]}) assert response.status_code == 204 client.post("/api/v1/habits", json={"name": "第三个", "kind": "boolean", "schedule_type": "daily"}) assert [habit["name"] for habit in client.get("/api/v1/habits").json()] == ["第二个", "第一个", "第三个"] grid = client.get("/api/v1/habits/grid", params={"week": "2026-09-07"}).json() assert [habit["name"] for habit in grid["habits"]] == ["第二个", "第一个", "第三个"] def test_habit_reorder_rejects_foreign_or_missing_ids(client): boot(client) habit = client.post("/api/v1/habits", json={"name": "自己的习惯", "kind": "boolean", "schedule_type": "daily"}).json() response = client.put( "/api/v1/habits/reorder", json={"habit_ids": [habit["id"], "00000000-0000-0000-0000-000000000001"]}, ) assert response.status_code == 404 def test_habit_logs_support_date_range_filter(client): boot(client) habit = client.post("/api/v1/habits", json={"name": "跑步", "kind": "boolean", "schedule_type": "daily"}).json() hid = habit["id"] for day in ("2026-08-01", "2026-08-15", "2026-09-01"): client.post(f"/api/v1/habits/{hid}/logs", json={"day": day, "value": 1}) all_logs = client.get(f"/api/v1/habits/{hid}/logs").json() ranged = client.get(f"/api/v1/habits/{hid}/logs", params={"from": "2026-08-10", "to": "2026-08-31"}).json() assert len(all_logs) == 3 assert [row["day"] for row in ranged] == ["2026-08-15"] def test_habit_grid_uses_a_bounded_number_of_queries(client, monkeypatch): boot(client) for index in range(6): client.post("/api/v1/habits", json={"name": f"习惯 {index}", "kind": "boolean", "schedule_type": "daily"}) from backend.db import get_engine statement_count = 0 engine = get_engine().sync_engine def count_queries(*_): nonlocal statement_count statement_count += 1 event.listen(engine, "before_cursor_execute", count_queries) try: response = client.get("/api/v1/habits/grid", params={"week": "2026-09-01"}) finally: event.remove(engine, "before_cursor_execute", count_queries) assert response.status_code == 200 assert len(response.json()["habits"]) == 6 assert statement_count <= 5 def test_habits_numeric_accumulation_pause_archive_grid_and_stats(client): boot(client) habit = client.post( "/api/v1/habits", json={"name": "喝水", "kind": "numeric", "target": 8, "schedule_type": "daily", "max_value": 10}, ) assert habit.status_code == 201 habit_id = habit.json()["id"] today = datetime.now(UTC).date().isoformat() for value in (6, 7): assert client.post(f"/api/v1/habits/{habit_id}/logs", json={"day": today, "value": value}).status_code == 200 assert client.get(f"/api/v1/habits/{habit_id}/logs").json()[0]["value"] == 10 assert client.put(f"/api/v1/habits/{habit_id}/logs/{today}", json={"value": 8}).json()["value"] == 8 yesterday = (datetime.now(UTC).date() - timedelta(days=1)).isoformat() assert client.post(f"/api/v1/habits/{habit_id}/pauses", json={"start_date": yesterday, "end_date": today}).status_code == 201 grid = client.get("/api/v1/habits/grid", params={"week": yesterday}).json() assert len(grid["days"]) == 7 and grid["habits"][0]["cells"] assert grid["habits"][0]["kind"] == "numeric" assert grid["habits"][0]["target"] == 8 assert grid["habits"][0]["max_value"] == 10 assert grid["habits"][0]["stats"]["total"] == 8 assert grid["habits"][0]["stats"]["completed_days"] == 1 stats = client.get(f"/api/v1/habits/{habit_id}/stats").json() assert stats["total"] == 8 and stats["completed_days"] == 1 assert client.delete(f"/api/v1/habits/{habit_id}").status_code == 204 archived = client.get("/api/v1/habits", params={"archived": True}).json() assert archived[0]["id"] == habit_id assert client.get(f"/api/v1/habits/{habit_id}/stats").json()["total"] == 8 def test_habit_permanent_delete_removes_habit_and_history(client): boot(client) created = client.post( "/api/v1/habits", json={"name": "待删除习惯", "kind": "numeric", "target": 3, "schedule_type": "daily"}, ) habit_id = created.json()["id"] today = datetime.now(UTC).date().isoformat() assert client.put(f"/api/v1/habits/{habit_id}/logs/{today}", json={"value": 2}).status_code == 200 assert client.delete(f"/api/v1/habits/{habit_id}/permanent").status_code == 204 assert all(row["id"] != habit_id for row in client.get("/api/v1/habits").json()) assert all(row["id"] != habit_id for row in client.get("/api/v1/habits", params={"archived": True}).json()) assert client.get(f"/api/v1/habits/{habit_id}/stats").status_code == 404 def test_boolean_interval_habit_schedule(client): boot(client) habit = client.post( "/api/v1/habits", json={"name": "拉伸", "kind": "boolean", "schedule_type": "interval", "interval_days": 2}, ) assert habit.status_code == 201 assert client.post( f"/api/v1/habits/{habit.json()['id']}/logs", json={"day": datetime.now(UTC).date().isoformat(), "value": 1} ).json()["value"] == 1 def test_attachment_security_ownership_and_size(client, tmp_path, monkeypatch): monkeypatch.setenv("DODO_ATTACHMENT_DIR", str(tmp_path / "uploads")) inbox = boot(client) task = client.post("/api/v1/tasks", json={"title": "文件", "list_id": inbox["id"]}).json() uploaded = client.post( f"/api/v1/tasks/{task['id']}/attachments", files={"file": ("notes.txt", b"safe text", "text/plain")}, ) assert uploaded.status_code == 201 attachment = uploaded.json() assert attachment["filename"] == "notes.txt" assert client.get(f"/api/v1/attachments/{attachment['id']}").content == b"safe text" bad = client.post( f"/api/v1/tasks/{task['id']}/attachments", files={"file": ("../evil.exe", b"x", "application/x-msdownload")}, ) assert bad.status_code == 400 assert client.delete(f"/api/v1/attachments/{attachment['id']}").status_code == 204 def test_tasks_can_be_loaded_by_list_and_completion_filter(client): inbox = boot(client) other = client.post("/api/v1/lists", json={"name": "其他清单"}).json() client.post("/api/v1/tasks", json={"title": "收集箱未完成", "list_id": inbox["id"]}) done = client.post("/api/v1/tasks", json={"title": "收集箱已完成", "list_id": inbox["id"]}).json() client.patch(f"/api/v1/tasks/{done['id']}", json={"completed": True, "version": done["version"]}) client.post("/api/v1/tasks", json={"title": "其他清单未完成", "list_id": other["id"]}) response = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "completed": False}) assert response.status_code == 200 assert [item["title"] for item in response.json()["items"]] == ["收集箱未完成"] def test_tasks_support_numbered_pagination_with_total(client): inbox = boot(client) for index in range(5): client.post("/api/v1/tasks", json={"title": f"任务 {index}", "list_id": inbox["id"]}) first = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 1, "page_size": 2}) second = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 2, "page_size": 2}) assert first.status_code == 200 assert first.json()["total"] == 5 assert first.json()["page"] == 1 assert first.json()["page_size"] == 2 assert len(first.json()["items"]) == 2 assert len(second.json()["items"]) == 2 assert {item["id"] for item in first.json()["items"]}.isdisjoint( {item["id"] for item in second.json()["items"]} ) def test_tasks_support_due_range_pagination(client): inbox = boot(client) client.post("/api/v1/tasks", json={"title": "今天", "list_id": inbox["id"], "due_at": "2026-09-05T08:00:00Z"}) client.post("/api/v1/tasks", json={"title": "以后", "list_id": inbox["id"], "due_at": "2026-09-08T08:00:00Z"}) client.post("/api/v1/tasks", json={"title": "无日期", "list_id": inbox["id"]}) response = client.get( "/api/v1/tasks", params={"due_from": "2026-09-05T00:00:00Z", "due_to": "2026-09-06T00:00:00Z", "page": 1}, ) assert response.status_code == 200 assert response.json()["total"] == 1 assert [item["title"] for item in response.json()["items"]] == ["今天"] def test_trash_supports_numbered_pagination_with_total(client): inbox = boot(client) for index in range(3): task = client.post("/api/v1/tasks", json={"title": f"删除 {index}", "list_id": inbox["id"]}).json() client.delete(f"/api/v1/tasks/{task['id']}") response = client.get("/api/v1/trash", params={"page": 2, "page_size": 2}) assert response.status_code == 200 assert response.json()["total"] == 3 assert response.json()["page"] == 2 assert len(response.json()["items"]) == 1 def test_ticktick_preview_import_dedupe_and_json_restore(client): boot(client) csv_data = "Title,List Name,Due Date,Status,ID\nImported,Inbox,2026-10-01,0,ext-1\n" preview = client.post("/api/v1/import/ticktick/preview", files={"file": ("tasks.csv", csv_data, "text/csv")}) assert preview.status_code == 200 and preview.json()["valid"] == 1 for _ in range(2): response = client.post("/api/v1/import/ticktick", files={"file": ("tasks.csv", csv_data, "text/csv")}) assert response.status_code == 200 assert response.json()["skipped"] == 1 assert len(client.get("/api/v1/tasks").json()["items"]) == 1 export = client.get("/api/v1/export").json() assert export["version"] == 1 and export["tasks"][0]["external_id"] == "ext-1" client.delete(f"/api/v1/tasks/{export['tasks'][0]['id']}") restored = client.post("/api/v1/restore", params={"mode": "replace"}, json=export) assert restored.status_code == 200 assert len(client.get("/api/v1/tasks").json()["items"]) == 1 assert client.post("/api/v1/restore", json={"version": 999}).status_code == 422 def test_audit_logs_cover_task_and_collection_operations(client): boot(client) folder = client.post("/api/v1/folders", json={"name": "F"}).json() task_list = client.post("/api/v1/lists", json={"name": "L", "folder_id": folder["id"]}).json() task = client.post("/api/v1/tasks", json={"title": "T", "list_id": task_list["id"]}).json() client.patch(f"/api/v1/tasks/{task['id']}", json={"completed": True, "version": task["version"]}) client.delete(f"/api/v1/tasks/{task['id']}") client.post(f"/api/v1/tasks/{task['id']}/restore") client.patch(f"/api/v1/lists/{task_list['id']}", json={"name": "L2"}) client.delete(f"/api/v1/folders/{folder['id']}") logs = client.get("/api/v1/audit-logs").json() pairs = {(row["entity_type"], row["action"]) for row in logs} assert {("task", "create"), ("task", "complete"), ("task", "delete"), ("task", "restore"), ("list", "update"), ("folder", "delete")} <= pairs def test_sessions_csrf_headers_revocation_and_docs(client): boot(client) assert client.get("/api/docs").status_code == 200 sessions = client.get("/api/v1/sessions").json() assert len(sessions) == 1 and sessions[0]["current"] is True assert client.delete(f"/api/v1/sessions/{sessions[0]['id']}").status_code == 204 assert client.get("/api/v1/me").status_code == 401 anonymous = client.__class__(client.app) with anonymous: assert anonymous.get("/api/docs").status_code == 401 def test_login_rate_limit_is_progressive(client): boot(client) client.post("/api/v1/auth/logout") statuses = [ client.post("/api/v1/auth/login", json={"username": "owner", "password": "wrong password"}).status_code for _ in range(8) ] assert 429 in statuses def test_security_headers(client): response = client.get("/health/live") assert response.headers["x-content-type-options"] == "nosniff" assert response.headers["content-security-policy"]