653 lines
27 KiB
Python
653 lines
27 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from uuid import UUID
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sqlalchemy import event, select, text
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from backend.models import RecurrenceException, RecurrenceTemplate, Task, UTCDateTime
|
|
|
|
BUSINESS_TIME_ZONE = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def local_today():
|
|
return datetime.now(BUSINESS_TIME_ZONE).date()
|
|
|
|
|
|
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 all(row["title"] != "缺少日期" for row in client.get("/api/v1/tasks").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=merge", json=exported)
|
|
assert restored.status_code == 200
|
|
restored_items = client.get("/api/v1/tasks", params={"limit": 100}).json()["items"]
|
|
restored_task = next(item for item in restored_items if item["title"] == "每周整理")
|
|
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 datetime.fromisoformat(recurrence["starts_at"]) == datetime(2026, 9, 8, 9, tzinfo=UTC)
|
|
|
|
|
|
def test_completing_repeating_task_resets_completed_subtasks_for_next_occurrence(client):
|
|
inbox = boot(client)
|
|
parent = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"title": "每日清理",
|
|
"list_id": inbox["id"],
|
|
"due_at": "2026-09-07T09:00:00Z",
|
|
"rrule": "FREQ=DAILY",
|
|
},
|
|
).json()
|
|
first = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "清理下载", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
).json()
|
|
second = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "清理缓存", "list_id": inbox["id"], "parent_id": parent["id"]},
|
|
).json()
|
|
for child in (first, second):
|
|
response = client.patch(
|
|
f"/api/v1/tasks/{child['id']}",
|
|
json={"completed": True, "version": child["version"]},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
completed = client.patch(
|
|
f"/api/v1/tasks/{parent['id']}",
|
|
json={"completed": True, "version": parent["version"]},
|
|
)
|
|
|
|
assert completed.status_code == 200
|
|
assert completed.json()["completed"] is False
|
|
assert completed.json()["due_at"].replace("Z", "") == "2026-09-08T09:00:00"
|
|
assert [child["completed"] for child in completed.json()["subtasks"]] == [False, False]
|
|
listed = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 1}).json()["items"]
|
|
repeated = next(task for task in listed if task["id"] == parent["id"])
|
|
assert [child["completed"] for child in repeated["subtasks"]] == [False, False]
|
|
|
|
|
|
|
|
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_chain_normalizes_absolute_instants_to_utc_on_sqlite(client):
|
|
inbox = boot(client)
|
|
created = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"title": "北京时间重复任务",
|
|
"list_id": inbox["id"],
|
|
"due_at": "2026-09-07T16:00:00+08:00",
|
|
"rrule": "FREQ=DAILY;COUNT=3",
|
|
},
|
|
)
|
|
assert created.status_code == 201
|
|
task = created.json()
|
|
assert datetime.fromisoformat(task["due_at"]) == datetime(2026, 9, 7, 8, tzinfo=UTC)
|
|
|
|
recurrence = client.get(f"/api/v1/tasks/{task['id']}/recurrence").json()
|
|
assert datetime.fromisoformat(recurrence["starts_at"]) == datetime(2026, 9, 7, 8, tzinfo=UTC)
|
|
|
|
edited = client.patch(
|
|
f"/api/v1/recurrences/{recurrence['id']}",
|
|
params={"scope": "this", "occurrence_at": "2026-09-08T16:00:00+08:00"},
|
|
json={"due_at": "2026-09-08T17:30:00+08:00"},
|
|
)
|
|
assert edited.status_code == 200
|
|
completed = client.post(
|
|
f"/api/v1/recurrences/{recurrence['id']}/complete",
|
|
json={"occurrence_at": "2026-09-08T16:00:00+08:00"},
|
|
)
|
|
assert completed.status_code == 200
|
|
|
|
async def stored_values():
|
|
from backend.db import get_engine
|
|
|
|
session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
|
|
async with session_factory() as db:
|
|
template = await db.scalar(
|
|
select(RecurrenceTemplate).where(RecurrenceTemplate.id == UUID(recurrence["id"]))
|
|
)
|
|
exception = await db.scalar(
|
|
select(RecurrenceException).where(RecurrenceException.template_id == template.id)
|
|
)
|
|
return template.starts_at, exception.occurrence_at, exception.due_at, exception.completed
|
|
|
|
import asyncio
|
|
|
|
starts_at, occurrence_at, due_at, is_completed = asyncio.run(stored_values())
|
|
assert starts_at == datetime(2026, 9, 7, 8, tzinfo=UTC)
|
|
assert occurrence_at == datetime(2026, 9, 8, 8, tzinfo=UTC)
|
|
assert due_at == datetime(2026, 9, 8, 9, 30, tzinfo=UTC)
|
|
assert is_completed is True
|
|
|
|
|
|
def test_utc_datetime_reads_legacy_sqlite_offset_text_as_the_same_instant(client):
|
|
inbox = boot(client)
|
|
task = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "历史数据", "list_id": inbox["id"], "due_at": "2026-09-07T08:00:00Z"},
|
|
).json()
|
|
|
|
async def inject_and_read():
|
|
from backend.db import get_engine
|
|
|
|
engine = get_engine()
|
|
async with engine.begin() as connection:
|
|
await connection.execute(
|
|
text("UPDATE tasks SET due_at = :value WHERE id = :task_id"),
|
|
{"value": "2026-09-07 16:00:00+08:00", "task_id": task["id"]},
|
|
)
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with session_factory() as db:
|
|
return await db.scalar(select(Task.due_at).where(Task.id == UUID(task["id"])))
|
|
|
|
import asyncio
|
|
|
|
assert asyncio.run(inject_and_read()) == datetime(2026, 9, 7, 8, tzinfo=UTC)
|
|
|
|
|
|
def test_recurrence_absolute_columns_use_utc_type_without_schema_change():
|
|
for column in (
|
|
Task.__table__.c.due_at,
|
|
RecurrenceTemplate.__table__.c.starts_at,
|
|
RecurrenceTemplate.__table__.c.ends_at,
|
|
RecurrenceException.__table__.c.occurrence_at,
|
|
RecurrenceException.__table__.c.due_at,
|
|
):
|
|
assert isinstance(column.type, UTCDateTime)
|
|
assert column.type.compile(dialect=postgresql.dialect()) == "TIMESTAMP WITH TIME ZONE"
|
|
|
|
|
|
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",
|
|
"start_date": "2026-08-01",
|
|
},
|
|
).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 = local_today().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 = (local_today() - 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 = local_today().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}").status_code == 204
|
|
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": 3,
|
|
"start_date": "2026-09-01",
|
|
},
|
|
)
|
|
assert habit.status_code == 201
|
|
habit_id = habit.json()["id"]
|
|
|
|
overdue_grid = client.get("/api/v1/habits/grid", params={"week": "2026-09-07"}).json()
|
|
overdue_cells = {cell["day"]: cell for cell in overdue_grid["habits"][0]["cells"]}
|
|
assert overdue_cells["2026-09-07"]["scheduled"] is True
|
|
assert overdue_cells["2026-09-08"]["scheduled"] is True
|
|
assert overdue_cells["2026-09-09"]["scheduled"] is True
|
|
|
|
completed = client.post(
|
|
f"/api/v1/habits/{habit_id}/logs", json={"day": "2026-09-09", "value": 1}
|
|
)
|
|
assert completed.status_code == 200
|
|
|
|
next_grid = client.get("/api/v1/habits/grid", params={"week": "2026-09-07"}).json()
|
|
next_cells = {cell["day"]: cell for cell in next_grid["habits"][0]["cells"]}
|
|
assert next_cells["2026-09-09"]["scheduled"] is True
|
|
assert next_cells["2026-09-10"]["scheduled"] is False
|
|
assert next_cells["2026-09-11"]["scheduled"] is False
|
|
assert next_cells["2026-09-12"]["scheduled"] is True
|
|
assert next_cells["2026-09-13"]["scheduled"] is True
|
|
|
|
|
|
def test_interval_habit_partial_progress_does_not_restart_interval(client):
|
|
boot(client)
|
|
habit = client.post(
|
|
"/api/v1/habits",
|
|
json={
|
|
"name": "喝水",
|
|
"kind": "numeric",
|
|
"target": 3,
|
|
"schedule_type": "interval",
|
|
"interval_days": 3,
|
|
"start_date": "2026-09-01",
|
|
},
|
|
).json()
|
|
|
|
assert client.put(f"/api/v1/habits/{habit['id']}/logs/2026-09-04", json={"value": 2}).status_code == 200
|
|
grid = client.get("/api/v1/habits/grid", params={"week": "2026-09-07"}).json()
|
|
cells = {cell["day"]: cell for cell in grid["habits"][0]["cells"]}
|
|
assert cells["2026-09-07"]["scheduled"] is True
|
|
|
|
assert client.put(f"/api/v1/habits/{habit['id']}/logs/2026-09-08", json={"value": 3}).status_code == 200
|
|
grid = client.get("/api/v1/habits/grid", params={"week": "2026-09-07"}).json()
|
|
cells = {cell["day"]: cell for cell in grid["habits"][0]["cells"]}
|
|
assert cells["2026-09-09"]["scheduled"] is False
|
|
assert cells["2026-09-10"]["scheduled"] is False
|
|
assert cells["2026-09-11"]["scheduled"] is True
|
|
|
|
|
|
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-05T18:00:00Z"})
|
|
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"] == 2
|
|
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": "merge"}, json=export)
|
|
assert restored.status_code == 200
|
|
# Legacy merge is non-destructive and does not resurrect a soft-deleted
|
|
# task whose external ID already exists.
|
|
assert len(client.get("/api/v1/tasks").json()["items"]) == 0
|
|
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"]
|