494 lines
19 KiB
Python
494 lines
19 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from uuid import UUID, uuid4
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from tests.test_mvp_backend import boot, local_today
|
|
|
|
|
|
def create_habit(client, **overrides):
|
|
payload = {"name": "阅读", "kind": "numeric", "target": 2, "max_value": 4, "schedule_type": "daily"}
|
|
payload.update(overrides)
|
|
return client.post("/api/v1/habits", json=payload)
|
|
|
|
|
|
def test_habit_patch_updates_only_submitted_fields_and_trims_name(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
|
|
response = client.patch(f"/api/v1/habits/{habit['id']}", json={"name": " 深度阅读 "})
|
|
|
|
assert response.status_code == 200
|
|
updated = response.json()
|
|
assert updated["name"] == "深度阅读"
|
|
assert updated["kind"] == "numeric"
|
|
assert updated["target"] == 2
|
|
assert updated["max_value"] == 4
|
|
|
|
|
|
def test_habit_name_rejects_blank_on_create_and_update(client):
|
|
boot(client)
|
|
assert create_habit(client, name=" ").status_code == 422
|
|
habit = create_habit(client).json()
|
|
assert client.patch(f"/api/v1/habits/{habit['id']}", json={"name": " \t "}).status_code == 422
|
|
|
|
|
|
def test_habit_schedule_validation_and_irrelevant_fields_are_cleared(client):
|
|
boot(client)
|
|
invalid_payloads = [
|
|
{"schedule_type": "weekly"},
|
|
{"schedule_type": "weekly", "weekdays": [1, 1]},
|
|
{"schedule_type": "weekly", "weekdays": [-1]},
|
|
{"schedule_type": "weekly", "weekdays": [7]},
|
|
{"schedule_type": "monthly"},
|
|
{"schedule_type": "monthly", "month_days": [1, 1]},
|
|
{"schedule_type": "monthly", "month_days": [0]},
|
|
{"schedule_type": "monthly", "month_days": [32]},
|
|
{"schedule_type": "interval"},
|
|
{"schedule_type": "interval", "interval_days": 0},
|
|
]
|
|
for payload in invalid_payloads:
|
|
assert create_habit(client, **payload).status_code == 422, payload
|
|
|
|
response = create_habit(
|
|
client,
|
|
schedule_type="weekly",
|
|
weekdays=[1, 3],
|
|
month_days=[8],
|
|
interval_days=2,
|
|
)
|
|
assert response.status_code == 201
|
|
assert response.json()["weekdays"] == [1, 3]
|
|
assert response.json()["month_days"] is None
|
|
assert response.json()["interval_days"] is None
|
|
|
|
|
|
def test_habit_patch_validates_merged_schedule_and_clears_old_schedule_fields(client):
|
|
boot(client)
|
|
habit = create_habit(client, schedule_type="weekly", weekdays=[1, 3]).json()
|
|
|
|
invalid = client.patch(f"/api/v1/habits/{habit['id']}", json={"weekdays": []})
|
|
assert invalid.status_code == 422
|
|
|
|
changed = client.patch(
|
|
f"/api/v1/habits/{habit['id']}",
|
|
json={"schedule_type": "monthly", "month_days": [10]},
|
|
)
|
|
assert changed.status_code == 200
|
|
assert changed.json()["weekdays"] is None
|
|
assert changed.json()["month_days"] == [10]
|
|
assert changed.json()["interval_days"] is None
|
|
|
|
|
|
def test_habit_create_rejects_non_finite_numeric_target_and_max_value(client):
|
|
boot(client)
|
|
for field in ("target", "max_value"):
|
|
for value in ("NaN", "Infinity", "-Infinity"):
|
|
response = create_habit(client, **{field: value})
|
|
assert response.status_code == 422, (field, value, response.text)
|
|
|
|
|
|
def test_habit_update_rejects_non_finite_numeric_target_and_max_value(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
for field in ("target", "max_value"):
|
|
for value in ("NaN", "Infinity", "-Infinity"):
|
|
response = client.patch(f"/api/v1/habits/{habit['id']}", json={field: value})
|
|
assert response.status_code == 422, (field, value, response.text)
|
|
|
|
|
|
def test_habit_log_post_rejects_non_finite_value(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
day = local_today().isoformat()
|
|
|
|
for value in ("NaN", "Infinity", "-Infinity"):
|
|
response = client.post(
|
|
f"/api/v1/habits/{habit['id']}/logs",
|
|
json={"day": day, "value": value},
|
|
)
|
|
assert response.status_code == 422, (value, response.text)
|
|
|
|
|
|
def test_habit_log_put_rejects_non_finite_value(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
day = local_today().isoformat()
|
|
|
|
for value in ("NaN", "Infinity", "-Infinity"):
|
|
response = client.put(
|
|
f"/api/v1/habits/{habit['id']}/logs/{day}",
|
|
json={"value": value},
|
|
)
|
|
assert response.status_code == 422, (value, response.text)
|
|
|
|
|
|
def test_habit_numeric_targets_and_boolean_normalization(client):
|
|
boot(client)
|
|
for payload in (
|
|
{"kind": "numeric", "target": 0},
|
|
{"kind": "numeric", "max_value": 0},
|
|
{"kind": "numeric", "target": 5, "max_value": 4},
|
|
):
|
|
assert create_habit(client, **payload).status_code == 422, payload
|
|
|
|
boolean = create_habit(client, kind="boolean", target=9, max_value=12).json()
|
|
assert boolean["target"] == 1
|
|
assert boolean["max_value"] == 1
|
|
updated = client.patch(
|
|
f"/api/v1/habits/{boolean['id']}", json={"target": 7, "max_value": 8}
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["target"] == 1
|
|
assert updated.json()["max_value"] == 1
|
|
|
|
|
|
def test_archived_habit_rejects_edit_logs_pause_and_active_permanent_delete(client):
|
|
boot(client)
|
|
active = create_habit(client).json()
|
|
assert client.delete(f"/api/v1/habits/{active['id']}/permanent").status_code == 409
|
|
|
|
habit = create_habit(client).json()
|
|
hid = habit["id"]
|
|
day = local_today().isoformat()
|
|
assert client.put(f"/api/v1/habits/{hid}/logs/{day}", json={"value": 1}).status_code == 200
|
|
assert client.delete(f"/api/v1/habits/{hid}").status_code == 204
|
|
|
|
assert client.patch(f"/api/v1/habits/{hid}", json={"name": "不可编辑"}).status_code == 409
|
|
assert client.post(f"/api/v1/habits/{hid}/logs", json={"day": day, "value": 1}).status_code == 409
|
|
assert client.put(f"/api/v1/habits/{hid}/logs/{day}", json={"value": 0}).status_code == 409
|
|
assert client.delete(f"/api/v1/habits/{hid}/logs/{day}").status_code == 409
|
|
assert client.post(
|
|
f"/api/v1/habits/{hid}/pauses", json={"start_date": day, "end_date": day}
|
|
).status_code == 409
|
|
assert client.delete(f"/api/v1/habits/{hid}/permanent").status_code == 204
|
|
|
|
|
|
def test_restore_archived_habit_appends_without_reordering_and_preserves_history(client):
|
|
boot(client)
|
|
first = create_habit(client, name="第一项").json()
|
|
restored_habit = create_habit(client, name="待恢复").json()
|
|
last = create_habit(client, name="最后一项").json()
|
|
day = local_today()
|
|
assert client.put(
|
|
f"/api/v1/habits/{restored_habit['id']}/logs/{day.isoformat()}", json={"value": 2}
|
|
).status_code == 200
|
|
assert client.post(
|
|
f"/api/v1/habits/{restored_habit['id']}/pauses",
|
|
json={"start_date": day.isoformat(), "end_date": day.isoformat()},
|
|
).status_code == 201
|
|
assert client.delete(f"/api/v1/habits/{restored_habit['id']}").status_code == 204
|
|
|
|
response = client.post(f"/api/v1/habits/{restored_habit['id']}/restore")
|
|
|
|
assert response.status_code == 200
|
|
restored = response.json()
|
|
assert restored["id"] == restored_habit["id"]
|
|
assert restored["archived_at"] is None
|
|
active = client.get("/api/v1/habits").json()
|
|
assert [row["id"] for row in active] == [first["id"], last["id"], restored_habit["id"]]
|
|
assert [row["position"] for row in active] == [first["position"], last["position"], last["position"] + 1]
|
|
assert client.get(f"/api/v1/habits/{restored_habit['id']}/logs").json() == [
|
|
{"day": day.isoformat(), "value": 2.0}
|
|
]
|
|
|
|
async def pause_count():
|
|
from backend.db import get_db
|
|
from backend.models import HabitPause
|
|
|
|
db_gen = get_db()
|
|
db = await anext(db_gen)
|
|
try:
|
|
return await db.scalar(
|
|
select(func.count()).select_from(HabitPause).where(
|
|
HabitPause.habit_id == UUID(restored_habit["id"])
|
|
)
|
|
)
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
assert client.portal.call(pause_count) == 1
|
|
audit_logs = client.get("/api/v1/audit-logs").json()
|
|
assert any(
|
|
row["action"] == "restore"
|
|
and row["entity_type"] == "habit"
|
|
and row["entity_id"] == restored_habit["id"]
|
|
for row in audit_logs
|
|
)
|
|
|
|
|
|
def test_restore_habit_rejects_active_and_repeated_restore(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
url = f"/api/v1/habits/{habit['id']}/restore"
|
|
|
|
assert client.post(url).status_code == 409
|
|
assert client.delete(f"/api/v1/habits/{habit['id']}").status_code == 204
|
|
assert client.post(url).status_code == 200
|
|
assert client.post(url).status_code == 409
|
|
restores = [
|
|
row
|
|
for row in client.get("/api/v1/audit-logs").json()
|
|
if row["action"] == "restore" and row["entity_type"] == "habit"
|
|
]
|
|
assert len(restores) == 1
|
|
|
|
|
|
def test_concurrent_restore_allows_at_most_one_success(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
assert client.delete(f"/api/v1/habits/{habit['id']}").status_code == 204
|
|
url = f"/api/v1/habits/{habit['id']}/restore"
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
statuses = sorted(pool.map(lambda _: client.post(url).status_code, range(2)))
|
|
|
|
assert statuses == [200, 409]
|
|
restores = [
|
|
row
|
|
for row in client.get("/api/v1/audit-logs").json()
|
|
if row["action"] == "restore" and row["entity_type"] == "habit"
|
|
]
|
|
assert len(restores) == 1
|
|
|
|
|
|
def test_concurrent_restore_of_different_habits_appends_unique_positions(
|
|
client, monkeypatch
|
|
):
|
|
import asyncio
|
|
import threading
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
boot(client)
|
|
active = [create_habit(client, name=name).json() for name in ("活跃一", "活跃二")]
|
|
archived = [create_habit(client, name=name).json() for name in ("归档一", "归档二")]
|
|
for habit in archived:
|
|
assert client.delete(f"/api/v1/habits/{habit['id']}").status_code == 204
|
|
|
|
original_scalar = AsyncSession.scalar
|
|
max_reads = threading.Barrier(2)
|
|
|
|
async def synchronized_scalar(self, statement, *args, **kwargs):
|
|
value = await original_scalar(self, statement, *args, **kwargs)
|
|
sql = str(statement)
|
|
if "max(habits.position)" in sql and "habits.archived_at IS NULL" in sql:
|
|
try:
|
|
await asyncio.to_thread(max_reads.wait, 1)
|
|
except threading.BrokenBarrierError:
|
|
pass
|
|
return value
|
|
|
|
monkeypatch.setattr(AsyncSession, "scalar", synchronized_scalar)
|
|
urls = [f"/api/v1/habits/{habit['id']}/restore" for habit in archived]
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
responses = list(pool.map(client.post, urls))
|
|
|
|
assert [response.status_code for response in responses] == [200, 200]
|
|
rows = client.get("/api/v1/habits").json()
|
|
assert [row["id"] for row in rows[:2]] == [habit["id"] for habit in active]
|
|
assert {row["id"] for row in rows[2:]} == {habit["id"] for habit in archived}
|
|
assert [row["position"] for row in rows] == [0, 1, 2, 3]
|
|
|
|
|
|
def test_concurrent_create_and_restore_keep_append_positions_unique(client, monkeypatch):
|
|
import asyncio
|
|
import threading
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
boot(client)
|
|
archived = create_habit(client, name="待恢复").json()
|
|
active = [create_habit(client, name=name).json() for name in ("活跃一", "活跃二")]
|
|
assert client.delete(f"/api/v1/habits/{archived['id']}").status_code == 204
|
|
|
|
original_scalar = AsyncSession.scalar
|
|
max_reads = threading.Barrier(2)
|
|
|
|
async def synchronized_scalar(self, statement, *args, **kwargs):
|
|
value = await original_scalar(self, statement, *args, **kwargs)
|
|
if "max(habits.position)" in str(statement):
|
|
try:
|
|
await asyncio.to_thread(max_reads.wait, 1)
|
|
except threading.BrokenBarrierError:
|
|
pass
|
|
return value
|
|
|
|
monkeypatch.setattr(AsyncSession, "scalar", synchronized_scalar)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
create_future = pool.submit(create_habit, client, name="并发新建")
|
|
restore_future = pool.submit(
|
|
client.post, f"/api/v1/habits/{archived['id']}/restore"
|
|
)
|
|
responses = [create_future.result(), restore_future.result()]
|
|
|
|
assert [response.status_code for response in responses] == [201, 200]
|
|
rows = client.get("/api/v1/habits").json()
|
|
assert [row["id"] for row in rows[:2]] == [habit["id"] for habit in active]
|
|
assert [row["position"] for row in rows] == [1, 2, 3, 4]
|
|
assert len({row["position"] for row in rows}) == len(rows)
|
|
|
|
|
|
def test_restore_habit_uses_owner_scoped_404(client):
|
|
boot(client)
|
|
|
|
async def add_foreign_habit():
|
|
from backend.db import get_db
|
|
from backend.models import Habit, User
|
|
|
|
db_gen = get_db()
|
|
db = await anext(db_gen)
|
|
try:
|
|
other = User(username="other", password_hash="not-used")
|
|
db.add(other)
|
|
await db.flush()
|
|
habit = Habit(
|
|
user_id=other.id,
|
|
name="别人的归档习惯",
|
|
kind="boolean",
|
|
target=1,
|
|
max_value=1,
|
|
schedule_type="daily",
|
|
start_date=local_today(),
|
|
archived_at=datetime.now(UTC),
|
|
position=0,
|
|
)
|
|
db.add(habit)
|
|
await db.commit()
|
|
return str(habit.id)
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
foreign_id = client.portal.call(add_foreign_habit)
|
|
assert client.post(f"/api/v1/habits/{foreign_id}/restore").status_code == 404
|
|
assert client.post(f"/api/v1/habits/{uuid4()}/restore").status_code == 404
|
|
|
|
|
|
def test_restore_habit_validates_uuid_authentication_and_csrf(client):
|
|
boot(client)
|
|
habit = create_habit(client).json()
|
|
assert client.delete(f"/api/v1/habits/{habit['id']}").status_code == 204
|
|
|
|
assert client.post("/api/v1/habits/not-a-uuid/restore").status_code == 422
|
|
csrf_rejected = client.post(
|
|
f"/api/v1/habits/{habit['id']}/restore",
|
|
headers={"origin": "https://dodo.example", "x-csrf-token": "wrong"},
|
|
)
|
|
assert csrf_rejected.status_code == 403
|
|
|
|
anonymous = client.__class__(client.app)
|
|
with anonymous:
|
|
assert anonymous.post(f"/api/v1/habits/{habit['id']}/restore").status_code == 401
|
|
|
|
|
|
def test_archived_habits_sort_by_archive_time_then_creation_and_id(client):
|
|
boot(client)
|
|
habits = [
|
|
create_habit(client, name=name).json()
|
|
for name in ("旧归档", "较早创建", "同刻B", "同刻A")
|
|
]
|
|
for habit in habits:
|
|
assert client.delete(f"/api/v1/habits/{habit['id']}").status_code == 204
|
|
|
|
async def set_sort_timestamps():
|
|
from backend.db import get_db
|
|
from backend.models import Habit
|
|
|
|
db_gen = get_db()
|
|
db = await anext(db_gen)
|
|
try:
|
|
rows = list(
|
|
(await db.scalars(select(Habit).where(Habit.id.in_([UUID(row["id"]) for row in habits])))).all()
|
|
)
|
|
by_id = {str(row.id): row for row in rows}
|
|
older = datetime(2026, 9, 8, tzinfo=UTC)
|
|
newer = datetime(2026, 9, 9, tzinfo=UTC)
|
|
by_id[habits[0]["id"]].archived_at = older
|
|
for index, habit in enumerate(habits[1:]):
|
|
row = by_id[habit["id"]]
|
|
row.archived_at = newer
|
|
row.created_at = newer - timedelta(days=1) if index == 0 else newer
|
|
await db.commit()
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
client.portal.call(set_sort_timestamps)
|
|
tied_ids = sorted([habits[2]["id"], habits[3]["id"]])
|
|
expected = tied_ids + [habits[1]["id"], habits[0]["id"]]
|
|
assert [row["id"] for row in client.get("/api/v1/habits", params={"archived": True}).json()] == expected
|
|
|
|
|
|
def test_paused_or_unscheduled_day_rejects_positive_progress_but_allows_correction(client):
|
|
boot(client)
|
|
monday = date(2026, 9, 7)
|
|
tuesday = monday + timedelta(days=1)
|
|
habit = create_habit(
|
|
client,
|
|
schedule_type="weekly",
|
|
weekdays=[monday.weekday()],
|
|
start_date=monday.isoformat(),
|
|
).json()
|
|
hid = habit["id"]
|
|
|
|
assert client.post(
|
|
f"/api/v1/habits/{hid}/logs", json={"day": tuesday.isoformat(), "value": 1}
|
|
).status_code == 409
|
|
assert client.get(f"/api/v1/habits/{hid}/stats").json()["total"] == 0
|
|
|
|
assert client.put(
|
|
f"/api/v1/habits/{hid}/logs/{monday.isoformat()}", json={"value": 2}
|
|
).status_code == 200
|
|
assert client.post(
|
|
f"/api/v1/habits/{hid}/pauses",
|
|
json={"start_date": monday.isoformat(), "end_date": monday.isoformat()},
|
|
).status_code == 201
|
|
assert client.post(
|
|
f"/api/v1/habits/{hid}/logs", json={"day": monday.isoformat(), "value": 1}
|
|
).status_code == 409
|
|
assert client.put(
|
|
f"/api/v1/habits/{hid}/logs/{monday.isoformat()}", json={"value": 3}
|
|
).status_code == 409
|
|
corrected = client.put(
|
|
f"/api/v1/habits/{hid}/logs/{monday.isoformat()}", json={"value": 0}
|
|
)
|
|
assert corrected.status_code == 200
|
|
assert corrected.json()["value"] == 0
|
|
assert client.delete(f"/api/v1/habits/{hid}/logs/{monday.isoformat()}").status_code == 204
|
|
|
|
|
|
def test_task_folder_and_list_names_are_trimmed_and_blank_rejected(client):
|
|
inbox = boot(client)
|
|
|
|
assert client.post("/api/v1/folders", json={"name": " "}).status_code == 422
|
|
folder_response = client.post("/api/v1/folders", json={"name": " 工作 "})
|
|
assert folder_response.status_code == 201
|
|
folder = folder_response.json()
|
|
assert folder["name"] == "工作"
|
|
assert client.patch(f"/api/v1/folders/{folder['id']}", json={"name": " \t "}).status_code == 422
|
|
assert client.patch(f"/api/v1/folders/{folder['id']}", json={"name": " 生活 "}).json()["name"] == "生活"
|
|
|
|
assert client.post("/api/v1/lists", json={"name": " "}).status_code == 422
|
|
list_response = client.post("/api/v1/lists", json={"name": " 清单 ", "folder_id": folder["id"]})
|
|
assert list_response.status_code == 201
|
|
task_list = list_response.json()
|
|
assert task_list["name"] == "清单"
|
|
assert client.patch(f"/api/v1/lists/{task_list['id']}", json={"name": " "}).status_code == 422
|
|
assert client.patch(f"/api/v1/lists/{task_list['id']}", json={"name": " 新清单 "}).json()["name"] == "新清单"
|
|
|
|
assert client.post("/api/v1/tasks", json={"title": " ", "list_id": inbox["id"]}).status_code == 422
|
|
task_response = client.post("/api/v1/tasks", json={"title": " 待办 ", "list_id": inbox["id"]})
|
|
assert task_response.status_code == 201
|
|
task = task_response.json()
|
|
assert task["title"] == "待办"
|
|
assert client.patch(
|
|
f"/api/v1/tasks/{task['id']}", json={"title": " ", "version": task["version"]}
|
|
).status_code == 422
|
|
renamed = client.patch(
|
|
f"/api/v1/tasks/{task['id']}", json={"title": " 已更新 ", "version": task["version"]}
|
|
)
|
|
assert renamed.status_code == 200
|
|
assert renamed.json()["title"] == "已更新"
|