diff --git a/backend/notes.py b/backend/notes.py index a5a36ae..3714bd9 100644 --- a/backend/notes.py +++ b/backend/notes.py @@ -1,4 +1,77 @@ -"""Notes slice: a per-book reading journal.""" -from flask import Blueprint +"""Notes slice: a per-book reading journal. Ownership is always derived through books.user_id.""" +from flask import Blueprint, g, jsonify + +from auth import login_required +from db import query, query_one +from validation import ApiError, json_body, text bp = Blueprint("notes", __name__, url_prefix="/api") + +MAX_BODY = 10_000 +NOTE_COLUMNS = "n.id, n.book_id, n.body, n.created_at, n.updated_at" + + +def to_json(row: dict) -> dict: + return { + "id": row["id"], + "book_id": row["book_id"], + "body": row["body"], + "created_at": row["created_at"].isoformat(), + "updated_at": row["updated_at"].isoformat(), + } + + +def _require_book(book_id: int) -> None: + if query_one("SELECT 1 FROM books WHERE id = %s AND user_id = %s", (book_id, g.user_id)) is None: + raise ApiError(404, "Book not found") + + +@bp.get("/books//notes") +@login_required +def list_notes(book_id: int): + _require_book(book_id) + rows = query( + f"SELECT {NOTE_COLUMNS} FROM notes n WHERE n.book_id = %s ORDER BY n.created_at DESC, n.id DESC", + (book_id,), + ) + return jsonify([to_json(row) for row in rows]) + + +@bp.post("/books//notes") +@login_required +def add_note(book_id: int): + _require_book(book_id) + body = text(json_body({"body"}), "body", MAX_BODY) + row = query_one(f"INSERT INTO notes AS n (book_id, body) VALUES (%s, %s) RETURNING {NOTE_COLUMNS}", (book_id, body)) + return jsonify(to_json(row)), 201 + + +@bp.patch("/notes/") +@login_required +def update_note(note_id: int): + body = text(json_body({"body"}), "body", MAX_BODY) + row = query_one( + f""" + UPDATE notes n + SET body = %s, updated_at = now() + FROM books b + WHERE n.id = %s AND b.id = n.book_id AND b.user_id = %s + RETURNING {NOTE_COLUMNS} + """, + (body, note_id, g.user_id), + ) + if row is None: + raise ApiError(404, "Note not found") + return jsonify(to_json(row)) + + +@bp.delete("/notes/") +@login_required +def delete_note(note_id: int): + row = query_one( + "DELETE FROM notes n USING books b WHERE n.id = %s AND b.id = n.book_id AND b.user_id = %s RETURNING n.id", + (note_id, g.user_id), + ) + if row is None: + raise ApiError(404, "Note not found") + return "", 204 diff --git a/backend/tests/test_notes.py b/backend/tests/test_notes.py new file mode 100644 index 0000000..d6c3b47 --- /dev/null +++ b/backend/tests/test_notes.py @@ -0,0 +1,68 @@ +from tests.support import ApiTestCase, db_execute + + +class NoteTests(ApiTestCase): + def setUp(self): + super().setUp() + self.register() + self.book = self.create_book() + self.notes_path = f"/api/books/{self.book['id']}/notes" + + def add(self, body: str, client=None): + return self.call("POST", self.notes_path, {"body": body}, client) + + def test_add_and_list_newest_first(self): + first = self.add(" Started reading ") + self.assertEqual(first.status_code, 201) + self.assertEqual(first.get_json()["body"], "Started reading") + self.add("Chapter 3 was great") + bodies = [n["body"] for n in self.call("GET", self.notes_path).get_json()] + self.assertEqual(bodies, ["Chapter 3 was great", "Started reading"]) + + def test_edit_updates_body_and_timestamp(self): + note = self.add("Draft").get_json() + response = self.call("PATCH", f"/api/notes/{note['id']}", {"body": "Final"}) + self.assertEqual(response.status_code, 200) + edited = response.get_json() + self.assertEqual(edited["body"], "Final") + self.assertGreater(edited["updated_at"], edited["created_at"]) + + def test_delete(self): + note = self.add("Temporary").get_json() + self.assertEqual(self.call("DELETE", f"/api/notes/{note['id']}").status_code, 204) + self.assertEqual(self.call("GET", self.notes_path).get_json(), []) + self.assertEqual(self.call("DELETE", f"/api/notes/{note['id']}").status_code, 404) + + def test_validation(self): + for body, status in ((" ", 400), ("x" * 10_001, 400), ("a\x00b", 400)): + with self.subTest(length=len(body)): + self.assertEqual(self.add(body).status_code, status) + self.assertEqual(self.call("POST", self.notes_path, {"body": "ok", "book_id": 2}).status_code, 400) + + def test_missing_book_and_out_of_range_ids_are_404(self): + self.assertEqual(self.call("GET", "/api/books/999/notes").status_code, 404) + self.assertEqual(self.call("PATCH", "/api/notes/99999999999", {"body": "x"}).status_code, 404) + + def test_deleting_book_deletes_its_notes(self): + self.add("Will be gone") + self.call("DELETE", f"/api/books/{self.book['id']}") + self.assertEqual(db_execute("SELECT count(*) FROM notes")[0][0], 0) + + +class NoteIsolationTests(ApiTestCase): + def test_other_user_cannot_read_add_edit_or_delete_notes(self): + self.register("alice") + book = self.create_book() + note = self.call("POST", f"/api/books/{book['id']}/notes", {"body": "Private"}).get_json() + bob = self.other_user("bob") + attempts = ( + ("GET", f"/api/books/{book['id']}/notes", None), + ("POST", f"/api/books/{book['id']}/notes", {"body": "Intrusion"}), + ("PATCH", f"/api/notes/{note['id']}", {"body": "Hacked"}), + ("DELETE", f"/api/notes/{note['id']}", None), + ) + for method, path, payload in attempts: + with self.subTest(method=method, path=path): + self.assertEqual(self.call(method, path, payload, client=bob).status_code, 404) + bodies = [n["body"] for n in self.call("GET", f"/api/books/{book['id']}/notes").get_json()] + self.assertEqual(bodies, ["Private"])