282 lines
11 KiB
Python
282 lines
11 KiB
Python
from sqlalchemy import select
|
|
|
|
from backend.auth import hash_password
|
|
from backend.db import get_db
|
|
from backend.models import Memo, TaskList, User
|
|
|
|
|
|
def boot(client):
|
|
response = client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
|
|
async def add_user(username: str):
|
|
db_gen = get_db()
|
|
db = await anext(db_gen)
|
|
try:
|
|
user = User(username=username, password_hash=hash_password("correct horse battery staple"))
|
|
db.add(user)
|
|
await db.flush()
|
|
db.add(TaskList(user_id=user.id, name="收集箱", is_inbox=True))
|
|
await db.commit()
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
|
|
def test_create_and_get_memo_trims_title_and_defaults_content(client):
|
|
boot(client)
|
|
|
|
created = client.post("/api/v1/memos", json={"title": " 第一篇 "})
|
|
|
|
assert created.status_code == 201
|
|
item = created.json()
|
|
assert item["title"] == "第一篇"
|
|
assert item["content"] == ""
|
|
assert item["version"] == 1
|
|
assert item["deleted_at"] is None
|
|
assert item["created_at"]
|
|
assert item["updated_at"]
|
|
assert client.get(f"/api/v1/memos/{item['id']}").json() == item
|
|
|
|
|
|
def test_memo_title_validation(client):
|
|
boot(client)
|
|
assert client.post("/api/v1/memos", json={"title": " "}).status_code == 422
|
|
assert client.post("/api/v1/memos", json={"title": "x" * 201}).status_code == 422
|
|
|
|
|
|
def test_create_memo_validates_title_length_after_trimming(client):
|
|
boot(client)
|
|
|
|
one = client.post("/api/v1/memos", json={"title": " x "})
|
|
two_hundred = client.post("/api/v1/memos", json={"title": f" {'x' * 200} "})
|
|
|
|
assert one.status_code == 201
|
|
assert one.json()["title"] == "x"
|
|
assert two_hundred.status_code == 201
|
|
assert two_hundred.json()["title"] == "x" * 200
|
|
assert client.post("/api/v1/memos", json={"title": " "}).status_code == 422
|
|
assert client.post("/api/v1/memos", json={"title": f" {'x' * 201} "}).status_code == 422
|
|
|
|
|
|
def test_update_memo_validates_title_length_after_trimming(client):
|
|
boot(client)
|
|
memo = client.post("/api/v1/memos", json={"title": "Before"}).json()
|
|
|
|
one = client.patch(
|
|
f"/api/v1/memos/{memo['id']}", json={"title": " x ", "version": memo["version"]}
|
|
)
|
|
two_hundred = client.patch(
|
|
f"/api/v1/memos/{memo['id']}",
|
|
json={"title": f" {'x' * 200} ", "version": one.json()["version"]},
|
|
)
|
|
|
|
assert one.status_code == 200
|
|
assert one.json()["title"] == "x"
|
|
assert two_hundred.status_code == 200
|
|
assert two_hundred.json()["title"] == "x" * 200
|
|
assert client.patch(
|
|
f"/api/v1/memos/{memo['id']}", json={"title": " ", "version": two_hundred.json()["version"]}
|
|
).status_code == 422
|
|
assert client.patch(
|
|
f"/api/v1/memos/{memo['id']}",
|
|
json={"title": f" {'x' * 201} ", "version": two_hundred.json()["version"]},
|
|
).status_code == 422
|
|
|
|
|
|
def test_list_memos_scopes_searches_orders_paginates_and_builds_excerpt(client):
|
|
boot(client)
|
|
first = client.post(
|
|
"/api/v1/memos", json={"title": "Alpha note", "content": " many\n words\tinside "}
|
|
).json()
|
|
second = client.post(
|
|
"/api/v1/memos", json={"title": "Second", "content": "alpha in body " + "x" * 200}
|
|
).json()
|
|
third = client.post("/api/v1/memos", json={"title": "Third"}).json()
|
|
assert client.delete(f"/api/v1/memos/{third['id']}").status_code == 204
|
|
|
|
page = client.get("/api/v1/memos", params={"scope": "active", "q": "ALPHA", "page_size": 1})
|
|
assert page.status_code == 200
|
|
data = page.json()
|
|
assert data["total"] == 2
|
|
assert data["page"] == 1
|
|
assert data["page_size"] == 1
|
|
assert data["items"][0]["id"] == second["id"]
|
|
assert data["items"][0]["excerpt"] == ("alpha in body " + "x" * 105 + "…")
|
|
assert "content" not in data["items"][0]
|
|
|
|
second_page = client.get(
|
|
"/api/v1/memos", params={"scope": "active", "q": "alpha", "page": 2, "page_size": 1}
|
|
).json()
|
|
assert second_page["items"][0]["id"] == first["id"]
|
|
assert second_page["items"][0]["excerpt"] == "many words inside"
|
|
assert [item["id"] for item in client.get("/api/v1/memos", params={"scope": "trash"}).json()["items"]] == [third["id"]]
|
|
assert client.get("/api/v1/memos", params={"page_size": 101}).status_code == 422
|
|
|
|
|
|
def test_patch_uses_atomic_version_lock_and_deleted_memos_reject_edit(client):
|
|
boot(client)
|
|
memo = client.post("/api/v1/memos", json={"title": "Before", "content": "secret"}).json()
|
|
|
|
updated = client.patch(
|
|
f"/api/v1/memos/{memo['id']}",
|
|
json={"title": " After ", "content": "new secret", "version": 1},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["title"] == "After"
|
|
assert updated.json()["content"] == "new secret"
|
|
assert updated.json()["version"] == 2
|
|
assert client.patch(
|
|
f"/api/v1/memos/{memo['id']}", json={"title": "stale", "version": 1}
|
|
).status_code == 409
|
|
assert client.get(f"/api/v1/memos/{memo['id']}").json()["title"] == "After"
|
|
|
|
assert client.delete(f"/api/v1/memos/{memo['id']}").status_code == 204
|
|
assert client.patch(
|
|
f"/api/v1/memos/{memo['id']}", json={"content": "forbidden", "version": 2}
|
|
).status_code == 409
|
|
|
|
|
|
def test_patch_requires_version(client):
|
|
boot(client)
|
|
memo = client.post("/api/v1/memos", json={"title": "Versioned"}).json()
|
|
assert client.patch(f"/api/v1/memos/{memo['id']}", json={"title": "No version"}).status_code == 422
|
|
|
|
|
|
def test_restore_and_purge_lifecycle(client):
|
|
boot(client)
|
|
active = client.post("/api/v1/memos", json={"title": "Active"}).json()
|
|
assert client.delete(f"/api/v1/memos/{active['id']}/purge").status_code == 409
|
|
|
|
assert client.delete(f"/api/v1/memos/{active['id']}").status_code == 204
|
|
deleted = client.get(f"/api/v1/memos/{active['id']}").json()
|
|
restored = client.post(f"/api/v1/memos/{active['id']}/restore")
|
|
assert restored.status_code == 200
|
|
assert restored.json()["deleted_at"] is None
|
|
assert restored.json()["updated_at"] > deleted["updated_at"]
|
|
|
|
assert client.delete(f"/api/v1/memos/{active['id']}").status_code == 204
|
|
assert client.delete(f"/api/v1/memos/{active['id']}/purge").status_code == 204
|
|
assert client.get(f"/api/v1/memos/{active['id']}").status_code == 404
|
|
|
|
|
|
def test_cross_user_memo_access_is_uniform_404(client):
|
|
boot(client)
|
|
memo = client.post("/api/v1/memos", json={"title": "Owner only"}).json()
|
|
client.post("/api/v1/auth/logout")
|
|
client.portal.call(add_user, "other")
|
|
assert client.post(
|
|
"/api/v1/auth/login", json={"username": "other", "password": "correct horse battery staple"}
|
|
).status_code == 200
|
|
|
|
assert client.get(f"/api/v1/memos/{memo['id']}").status_code == 404
|
|
assert client.patch(f"/api/v1/memos/{memo['id']}", json={"title": "x", "version": 1}).status_code == 404
|
|
assert client.delete(f"/api/v1/memos/{memo['id']}").status_code == 404
|
|
assert client.post(f"/api/v1/memos/{memo['id']}/restore").status_code == 404
|
|
assert client.delete(f"/api/v1/memos/{memo['id']}/purge").status_code == 404
|
|
|
|
|
|
def test_memo_audit_records_actions_without_content(client):
|
|
boot(client)
|
|
memo = client.post("/api/v1/memos", json={"title": "Audit", "content": "TOP SECRET"}).json()
|
|
client.patch(
|
|
f"/api/v1/memos/{memo['id']}", json={"content": "NEW SECRET", "version": memo["version"]}
|
|
)
|
|
client.delete(f"/api/v1/memos/{memo['id']}")
|
|
client.post(f"/api/v1/memos/{memo['id']}/restore")
|
|
client.delete(f"/api/v1/memos/{memo['id']}")
|
|
client.delete(f"/api/v1/memos/{memo['id']}/purge")
|
|
|
|
logs = [row for row in client.get("/api/v1/audit-logs").json() if row["entity_type"] == "memo"]
|
|
assert [row["action"] for row in reversed(logs)] == ["create", "update", "delete", "restore", "delete", "purge"]
|
|
serialized = str(logs)
|
|
assert "TOP SECRET" not in serialized
|
|
assert "NEW SECRET" not in serialized
|
|
|
|
|
|
def test_json_and_csv_backup_round_trip_memos_with_all_fields(client):
|
|
boot(client)
|
|
active = client.post("/api/v1/memos", json={"title": "Active", "content": "alpha"}).json()
|
|
active = client.patch(
|
|
f"/api/v1/memos/{active['id']}", json={"title": "Active 2", "version": 1}
|
|
).json()
|
|
deleted = client.post("/api/v1/memos", json={"title": "Deleted", "content": "trash"}).json()
|
|
client.delete(f"/api/v1/memos/{deleted['id']}")
|
|
|
|
exported = client.get("/api/v1/export").json()
|
|
assert len(exported["memos"]) == 2
|
|
exported_active = next(row for row in exported["memos"] if row["title"] == "Active 2")
|
|
assert exported_active["content"] == "alpha"
|
|
assert exported_active["version"] == 2
|
|
assert exported_active["created_at"].replace("+00:00", "Z") == active["created_at"]
|
|
assert exported_active["updated_at"].replace("+00:00", "Z") == active["updated_at"]
|
|
assert exported_active["deleted_at"] is None
|
|
|
|
csv_backup = client.get("/api/v1/export.csv")
|
|
assert b"memos" in csv_backup.content
|
|
restored_csv = client.post(
|
|
"/api/v1/restore.csv?mode=merge",
|
|
files={"file": ("dodo-export.csv", csv_backup.content, "text/csv")},
|
|
)
|
|
assert restored_csv.status_code == 200
|
|
assert client.get("/api/v1/memos").json()["total"] == 1
|
|
assert client.get("/api/v1/memos", params={"scope": "trash"}).json()["total"] == 1
|
|
restored_active = client.get("/api/v1/memos").json()["items"][0]
|
|
detail = client.get(f"/api/v1/memos/{restored_active['id']}").json()
|
|
assert detail["title"] == "Active 2"
|
|
assert detail["content"] == "alpha"
|
|
assert detail["version"] == 2
|
|
assert detail["created_at"] == active["created_at"]
|
|
assert detail["updated_at"] == active["updated_at"]
|
|
|
|
client.post("/api/v1/memos", json={"title": "Extra"})
|
|
merged = client.post("/api/v1/restore?mode=merge", json=exported)
|
|
assert merged.status_code == 200
|
|
assert client.get("/api/v1/memos").json()["total"] == 2
|
|
|
|
|
|
def test_restore_reassigns_foreign_memo_to_current_user(client):
|
|
boot(client)
|
|
exported = client.get("/api/v1/export").json()
|
|
client.portal.call(add_user, "other")
|
|
|
|
async def add_foreign_memo():
|
|
db_gen = get_db()
|
|
db = await anext(db_gen)
|
|
try:
|
|
other = await db.scalar(select(User).where(User.username == "other"))
|
|
memo = Memo(user_id=other.id, title="Foreign")
|
|
db.add(memo)
|
|
await db.commit()
|
|
return str(memo.id)
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
foreign_id = client.portal.call(add_foreign_memo)
|
|
exported["memos"] = [{
|
|
"id": foreign_id,
|
|
"title": "Imported",
|
|
"content": "owned here",
|
|
"version": 3,
|
|
"created_at": "2026-01-01T00:00:00+00:00",
|
|
"updated_at": "2026-01-02T00:00:00+00:00",
|
|
"deleted_at": None,
|
|
}]
|
|
response = client.post("/api/v1/restore?mode=merge", json=exported)
|
|
assert response.status_code == 200
|
|
imported = client.get("/api/v1/memos", params={"q": "Imported"}).json()["items"][0]
|
|
assert imported["id"] != foreign_id
|
|
assert client.get(f"/api/v1/memos/{imported['id']}").json()["content"] == "owned here"
|
|
assert client.post("/api/v1/restore?mode=merge", json=exported).status_code == 200
|
|
assert client.get("/api/v1/memos", params={"q": "Imported"}).json()["total"] == 1
|
|
|
|
|
|
def test_legacy_backup_without_memos_still_restores(client):
|
|
boot(client)
|
|
backup = client.get("/api/v1/export").json()
|
|
backup.pop("memos", None)
|
|
assert client.post("/api/v1/restore?mode=merge", json=backup).status_code == 200
|