feat: recurring tasks, habits, attachments, import/export, audit and security
This commit is contained in:
@@ -0,0 +1,171 @@
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
|
||||
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_recurring_calendar_exceptions_and_scopes(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=WEEKLY;BYDAY=TU,TH;COUNT=5"}
|
||||
)
|
||||
assert recurrence.status_code == 201
|
||||
recurrence_id = recurrence.json()["id"]
|
||||
calendar = client.get(
|
||||
"/api/v1/calendar", params={"start": "2026-09-01", "end": "2026-09-30"}
|
||||
).json()
|
||||
occurrences = [row for row in calendar if row["recurrence_id"] == recurrence_id]
|
||||
assert len(occurrences) == 5
|
||||
assert occurrences[0]["title"] == "站会"
|
||||
|
||||
at = occurrences[1]["occurrence_at"]
|
||||
edited = client.patch(
|
||||
f"/api/v1/recurrences/{recurrence_id}",
|
||||
params={"scope": "this", "occurrence_at": at},
|
||||
json={"title": "特殊站会"},
|
||||
)
|
||||
assert edited.status_code == 200
|
||||
client.post(f"/api/v1/recurrences/{recurrence_id}/complete", json={"occurrence_at": at})
|
||||
changed = client.get(
|
||||
"/api/v1/calendar", params={"start": "2026-09-01", "end": "2026-09-30"}
|
||||
).json()
|
||||
exception = next(row for row in changed if row.get("occurrence_at") == at)
|
||||
assert exception["title"] == "特殊站会" and exception["completed"] is True
|
||||
assert client.delete(
|
||||
f"/api/v1/recurrences/{recurrence_id}",
|
||||
params={"scope": "this", "occurrence_at": occurrences[2]["occurrence_at"]},
|
||||
).status_code == 204
|
||||
assert len(client.get(
|
||||
"/api/v1/calendar", params={"start": "2026-09-01", "end": "2026-09-30"}
|
||||
).json()) == 4
|
||||
|
||||
|
||||
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"]
|
||||
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_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_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"]
|
||||
Reference in New Issue
Block a user