diff --git a/jottit/blueprints/secret.py b/jottit/blueprints/secret.py index b2e2554..2daea30 100644 --- a/jottit/blueprints/secret.py +++ b/jottit/blueprints/secret.py @@ -62,9 +62,14 @@ ) secret_bp.add_url_rule("/site/changes", endpoint="site_changes", view_func=site_views.changes) secret_bp.add_url_rule( - "/site/changes.atom", - endpoint="site_changes_atom", - view_func=site_views.changes_atom, + "/site/changes.rss", + endpoint="site_changes_rss", + view_func=site_views.changes_rss, +) +secret_bp.add_url_rule( + "/site/changes.json", + endpoint="site_changes_json", + view_func=site_views.changes_json, ) secret_bp.add_url_rule( "/site/hide-primer", diff --git a/jottit/blueprints/site.py b/jottit/blueprints/site.py index 74c6f57..527b7cd 100644 --- a/jottit/blueprints/site.py +++ b/jottit/blueprints/site.py @@ -12,5 +12,6 @@ site_bp.add_url_rule("/forgot-password", view_func=views.forgot_password, methods=["GET", "POST"]) site_bp.add_url_rule("/change-password", view_func=views.change_password, methods=["GET", "POST"]) site_bp.add_url_rule("/changes", view_func=views.changes) -site_bp.add_url_rule("/changes.atom", view_func=views.changes_atom) +site_bp.add_url_rule("/changes.rss", view_func=views.changes_rss) +site_bp.add_url_rule("/changes.json", view_func=views.changes_json) site_bp.add_url_rule("/hide-primer", view_func=views.hide_primer, methods=["POST"]) diff --git a/jottit/db.py b/jottit/db.py index 74e5ed6..eccfe0e 100644 --- a/jottit/db.py +++ b/jottit/db.py @@ -255,6 +255,74 @@ def get_revision( return conn.execute(stmt.limit(1)).first() +def get_revisions( + conn: Connection, + *, + page_id: int, + before: int | None = None, + limit: int = 20, +) -> list[Row]: + """List revisions for a page in newest-first order. + + `before` filters to revisions strictly less than the given revision + number — used to page back through history (the original called this + `start`). `limit` is per-page; the default of 20 matches the 2007 UI. + """ + stmt = ( + select(revisions) + .where(revisions.c.page_id == page_id, revisions.c.revision > 0) + .order_by(revisions.c.revision.desc()) + .limit(limit) + ) + if before is not None: + stmt = stmt.where(revisions.c.revision < before) + return list(conn.execute(stmt).all()) + + +def get_revisions_count(conn: Connection, *, page_id: int) -> int: + """Count of non-sentinel revisions for a page; used for history pagination.""" + return conn.execute( + select(func.count()) + .select_from(revisions) + .where(revisions.c.page_id == page_id, revisions.c.revision > 0) + ).scalar_one() + + +def get_changes( + conn: Connection, + *, + site_id: int, + before: int | None = None, + limit: int = 20, +) -> list[Row]: + """Site-wide activity feed: every revision across every page, newest first. + + Each row carries `page_name`, `page_deleted`, the revision fields, and + `id` of the revision (for the `before=` cursor). The 2007 implementation + of this was stubbed out with `return []` and a comment marking the query + as impossible at the time; modern Postgres handles it without issue. + """ + stmt = ( + select( + revisions.c.id, + revisions.c.revision, + revisions.c.content, + revisions.c.changes, + revisions.c.ip, + revisions.c.created, + pages.c.name.label("page_name"), + pages.c.deleted.label("page_deleted"), + ) + .select_from(revisions.join(pages, revisions.c.page_id == pages.c.id)) + .where(pages.c.site_id == site_id, revisions.c.revision > 0) + .order_by(revisions.c.created.desc(), revisions.c.id.desc()) + .limit(limit) + ) + if before is not None: + stmt = stmt.where(revisions.c.id < before) + return list(conn.execute(stmt).all()) + + def new_page( conn: Connection, *, diff --git a/jottit/templates/changes.html b/jottit/templates/changes.html new file mode 100644 index 0000000..b01f799 --- /dev/null +++ b/jottit/templates/changes.html @@ -0,0 +1,30 @@ + + + + + Recent changes + + +

Recent changes

+ {% if changes %} + + {% if older_before %} +

Older →

+ {% endif %} + {% else %} +

No changes yet.

+ {% endif %} +

Back to site

+ + diff --git a/jottit/templates/diff.html b/jottit/templates/diff.html new file mode 100644 index 0000000..82187d2 --- /dev/null +++ b/jottit/templates/diff.html @@ -0,0 +1,18 @@ + + + + + Diff: {{ page_name or "Home" }} + + +

Diff: {{ page_name or "Home" }}

+

+ Comparing + revision {{ a.revision }} + with + revision {{ b.revision }}. +

+
{{ diff_html|safe }}
+

Back to history

+ + diff --git a/jottit/templates/feeds/changes.rss.xml b/jottit/templates/feeds/changes.rss.xml new file mode 100644 index 0000000..25568a7 --- /dev/null +++ b/jottit/templates/feeds/changes.rss.xml @@ -0,0 +1,18 @@ + + + + {{ site_title|e }} — recent changes + {{ site_url|e }} + + Recent changes across all pages. + {% for c in changes %} + + {{ (c.page_name or "Home")|e }} — revision {{ c.revision }} + {{ absolute_url(page_slug(c.page_name) + "?r=" + c.revision|string)|e }} + {{ absolute_url(page_slug(c.page_name) + "?r=" + c.revision|string)|e }} + {{ c.created.strftime("%a, %d %b %Y %H:%M:%S GMT") }} + {% if c.changes %}{% endif %} + + {% endfor %} + + diff --git a/jottit/templates/feeds/history.rss.xml b/jottit/templates/feeds/history.rss.xml new file mode 100644 index 0000000..622f525 --- /dev/null +++ b/jottit/templates/feeds/history.rss.xml @@ -0,0 +1,18 @@ + + + + {{ site_title|e }} — {{ page_label|e }} history + {{ page_url|e }} + + Revision history for {{ page_label|e }}. + {% for r in revisions %} + + Revision {{ r.revision }} + {{ page_url|e }}?r={{ r.revision }} + {{ page_url|e }}?r={{ r.revision }} + {{ r.created.strftime("%a, %d %b %Y %H:%M:%S GMT") }} + {% if r.changes %}{% endif %} + + {% endfor %} + + diff --git a/jottit/templates/history.html b/jottit/templates/history.html new file mode 100644 index 0000000..40e99e3 --- /dev/null +++ b/jottit/templates/history.html @@ -0,0 +1,24 @@ + + + + + History: {{ page_name or "Home" }} + + +

History: {{ page_name or "Home" }}

+

{{ total }} revision{{ "s" if total != 1 }}.

+ + {% if older_before %} +

Older →

+ {% endif %} +

Back to page

+ + diff --git a/jottit/views/page.py b/jottit/views/page.py index 3c23946..9f10703 100644 --- a/jottit/views/page.py +++ b/jottit/views/page.py @@ -1,8 +1,8 @@ from __future__ import annotations -from flask import abort, g, jsonify, redirect, render_template, request +from flask import abort, g, jsonify, make_response, redirect, render_template, request from flask.typing import ResponseReturnValue -from sqlalchemy import Connection +from sqlalchemy import Connection, Row from jottit import auth from jottit.db import ( @@ -11,10 +11,14 @@ get_page, get_request_conn, get_revision, + get_revisions, + get_revisions_count, new_page, + new_revision, undelete_page, update_page, ) +from jottit.diff import better_diff from jottit.render import format_content from jottit.urls import page_slug, site_root @@ -39,11 +43,27 @@ def view(site_slug: str, page_name: str) -> ResponseReturnValue: if request.method == "POST": if mode == "current_revision": return _current_revision(conn, page_name) + if mode == "revert": + return _revert(conn, page_name) + if mode == "undelete": + return _undelete(conn, page_name) return _save_edit(conn, page_name) if mode == "edit": return _render_edit_form(conn, page_name) + if mode == "history": + return _render_history(conn, page_name) + + if mode == "history_rss": + return _render_history_rss(conn, page_name) + + if mode == "history_json": + return _render_history_json(conn, page_name) + + if mode == "diff": + return _render_diff(conn, page_name) + return _render_view(conn, page_name) @@ -56,10 +76,13 @@ def _action_for(mode: str) -> str: return "edit" if mode == "edit": return "edit" - # GET ?r=N exposes an old revision, which is gated more tightly than - # the latest one on `public` sites — see is_action_allowed. - if request.args.get("r") is not None: + # Old revisions, history listings, and diffs all expose pre-current + # content; on public sites that's gated more tightly than the latest + # revision. See is_action_allowed. + if mode in ("history", "diff") or request.args.get("r") is not None: return "view_revision" + # RSS / JSON Feed readers don't typically auth, so feeds use the same + # loose "view" check public sites use for the latest revision. return "view" @@ -85,6 +108,136 @@ def _render_view(conn: Connection, page_name: str) -> ResponseReturnValue: ) +def _render_history_rss(conn: Connection, page_name: str) -> ResponseReturnValue: + page = get_page(conn, site_id=g.site.id, page_name=page_name) + if page is None: + abort(404) + revisions_page = get_revisions(conn, page_id=page.id, limit=20) + body = render_template( + "feeds/history.rss.xml", + page_name=page_name, + revisions=revisions_page, + page_url=_page_absolute_url(page_name), + feed_url=_page_absolute_url(page_name, "m=history_rss"), + site_title=(g.site.title or g.site.public_url or g.site.secret_url), + page_label=page_name or "Home", + ) + response = make_response(body) + response.headers["Content-Type"] = "application/rss+xml; charset=utf-8" + return response + + +def _render_history_json(conn: Connection, page_name: str) -> ResponseReturnValue: + page = get_page(conn, site_id=g.site.id, page_name=page_name) + if page is None: + abort(404) + revisions_page = get_revisions(conn, page_id=page.id, limit=20) + payload = { + "version": "https://jsonfeed.org/version/1.1", + "title": (g.site.title or g.site.public_url or g.site.secret_url) + + f" — {page_name or 'Home'} history", + "home_page_url": _page_absolute_url(page_name), + "feed_url": _page_absolute_url(page_name, "m=history_json"), + "items": [ + { + "id": _page_absolute_url(page_name, f"r={r.revision}"), + "url": _page_absolute_url(page_name, f"r={r.revision}"), + "title": f"Revision {r.revision}", + "content_html": r.changes or "", + "date_published": r.created.isoformat() + "Z", + } + for r in revisions_page + ], + } + response = jsonify(payload) + response.headers["Content-Type"] = "application/feed+json" + return response + + +def _page_absolute_url(page_name: str, query: str = "") -> str: + base = f"{request.scheme}://{request.host}{site_root()}{page_slug(page_name)}" + return f"{base}?{query}" if query else base + + +def _render_diff(conn: Connection, page_name: str) -> ResponseReturnValue: + """Render the diff between two revisions. + + URL shapes (matches the original): + - `?m=diff` → latest vs previous + - `?m=diff&r=N` → revision N vs N-1 + - `?m=diff&r=A&r=B` → between A and B (ordering normalized) + """ + page = get_page(conn, site_id=g.site.id, page_name=page_name) + if page is None: + return render_template("notfound.html", page_name=page_name), 404 + + a_rev, b_rev = _resolve_diff_revisions(conn, page_id=page.id) + if a_rev is None or b_rev is None: + abort(400) + + return render_template( + "diff.html", + page_name=page_name, + a=a_rev, + b=b_rev, + diff_html=better_diff(a_rev.content, b_rev.content), + site_root_path=site_root(), + ) + + +def _resolve_diff_revisions(conn: Connection, *, page_id: int) -> tuple[Row | None, Row | None]: + """Pick the (older, newer) revisions to diff based on the `r` query params.""" + rs = request.args.getlist("r") + if len(rs) > 2: + return None, None + try: + ints = sorted({int(r) for r in rs}) + except ValueError: + return None, None + + if not ints: + # No params: latest vs previous. + latest = get_revision(conn, page_id=page_id) + if latest is None: + return None, None + prev = get_revision(conn, page_id=page_id, revision=max(latest.revision - 1, 1)) + return prev, latest + + if len(ints) == 1: + n = ints[0] + b = get_revision(conn, page_id=page_id, revision=n) + a = get_revision(conn, page_id=page_id, revision=max(n - 1, 1)) + return a, b + + a = get_revision(conn, page_id=page_id, revision=ints[0]) + b = get_revision(conn, page_id=page_id, revision=ints[1]) + return a, b + + +def _render_history(conn: Connection, page_name: str) -> ResponseReturnValue: + page = get_page(conn, site_id=g.site.id, page_name=page_name) + if page is None: + return render_template("notfound.html", page_name=page_name), 404 + + total = get_revisions_count(conn, page_id=page.id) + if total == 0: + # Nothing to show — drop the user back at the page itself. + return redirect(site_root() + page_slug(page_name), code=303) + + before = request.args.get("before", type=int) + revisions_page = get_revisions(conn, page_id=page.id, before=before, limit=20) + older_before = revisions_page[-1].revision if len(revisions_page) == 20 else None + + return render_template( + "history.html", + page_name=page_name, + revisions=revisions_page, + total=total, + older_before=older_before, + site_root_path=site_root(), + ) + + def _render_edit_form(conn: Connection, page_name: str) -> ResponseReturnValue: page = get_page(conn, site_id=g.site.id, page_name=page_name) if page is None: @@ -108,6 +261,62 @@ def _render_edit_form(conn: Connection, page_name: str) -> ResponseReturnValue: ) +def _revert(conn: Connection, page_name: str) -> ResponseReturnValue: + """Restore a page's content to an earlier revision (recorded as a new revision).""" + page = get_page(conn, site_id=g.site.id, page_name=page_name) + if page is None: + abort(400) + + target_revision = request.form.get("r", type=int) + if target_revision is None: + abort(400) + target = get_revision(conn, page_id=page.id, revision=target_revision) + if target is None: + abort(400) + + # Undelete first so update_page sees a live page; update_page then + # records the revert as a normal revision with its own diff summary. + if page.deleted: + undelete_page(conn, page_id=page.id, name=page_name) + latest = get_revision(conn, page_id=page.id) + if latest is not None and latest.content != target.content: + update_page(conn, page_id=page.id, content=target.content, ip=request.remote_addr) + + return _redirect_to(page_name) + + +def _undelete(conn: Connection, page_name: str) -> ResponseReturnValue: + """Restore a deleted page and append an "undeleted" revision. + + The previous revision (before the delete sentinel) had the real + content; we re-record it so the page's latest revision shows the + live body again. + """ + page = get_page(conn, site_id=g.site.id, page_name=page_name) + if page is None: + abort(400) + + undelete_page(conn, page_id=page.id, name=page_name) + latest = get_revision(conn, page_id=page.id) + if latest is None: + return _redirect_to(page_name) + + # The "deleted" sentinel is the latest revision; the one before holds + # the last real content. Restore that. + pre_delete = get_revision(conn, page_id=page.id, revision=max(latest.revision - 1, 1)) + if pre_delete is not None: + new_revision( + conn, + page_id=page.id, + revision=latest.revision + 1, + content=pre_delete.content, + changes="Delete undone.", + ip=request.remote_addr, + ) + + return _redirect_to(page_name) + + def _current_revision(conn: Connection, page_name: str) -> ResponseReturnValue: """JSON endpoint used by the editor to poll for concurrent edits.""" page = get_page(conn, site_id=g.site.id, page_name=page_name) diff --git a/jottit/views/site.py b/jottit/views/site.py index 993df3b..dd4b67a 100644 --- a/jottit/views/site.py +++ b/jottit/views/site.py @@ -6,8 +6,14 @@ from flask.typing import ResponseReturnValue from jottit import auth, mail -from jottit.db import claim_site, get_request_conn, recover_password, set_change_pwd_token -from jottit.urls import site_root +from jottit.db import ( + claim_site, + get_changes, + get_request_conn, + recover_password, + set_change_pwd_token, +) +from jottit.urls import page_slug, site_root _ALLOWED_SECURITY_LEVELS = {"private", "public", "open"} @@ -204,12 +210,90 @@ def _send_recovery_email(*, to: str, token: str) -> None: ) -def changes(site_slug: str) -> str: - return f"site:{site_slug} site/changes GET (TODO)" +def changes(site_slug: str) -> ResponseReturnValue: + """Site-wide activity feed: recent revisions across all pages.""" + if g.site is None: + abort(404) + if (response := auth.gate("view_revision")) is not None: + return response + + conn = get_request_conn() + if conn is None: + abort(500) + + before = request.args.get("before", type=int) + rows = get_changes(conn, site_id=g.site.id, before=before, limit=20) + older_before = rows[-1].id if len(rows) == 20 else None + + return render_template( + "changes.html", + changes=rows, + older_before=older_before, + site_root_path=site_root(), + page_slug=page_slug, + ) + + +def changes_rss(site_slug: str) -> ResponseReturnValue: + """RSS 2.0 site-wide changes feed.""" + if g.site is None: + abort(404) + if (response := auth.gate("view")) is not None: + return response + rows = _recent_changes() + body = render_template( + "feeds/changes.rss.xml", + changes=rows, + site_title=g.site.title or g.site.public_url or g.site.secret_url, + feed_url=_absolute_url("site/changes.rss"), + site_url=_absolute_url(""), + site_root_path=site_root(), + page_slug=page_slug, + absolute_url=_absolute_url, + ) + return body, 200, {"Content-Type": "application/rss+xml; charset=utf-8"} + + +def changes_json(site_slug: str) -> ResponseReturnValue: + """JSON Feed (jsonfeed.org) site-wide changes feed.""" + if g.site is None: + abort(404) + if (response := auth.gate("view")) is not None: + return response + rows = _recent_changes() + payload = { + "version": "https://jsonfeed.org/version/1.1", + "title": (g.site.title or g.site.public_url or g.site.secret_url) + " — changes", + "home_page_url": _absolute_url(""), + "feed_url": _absolute_url("site/changes.json"), + "items": [ + { + "id": _absolute_url(f"{page_slug(c.page_name)}?r={c.revision}"), + "url": _absolute_url(f"{page_slug(c.page_name)}?r={c.revision}"), + "title": (c.page_name or "Home") + f" — revision {c.revision}", + "content_html": c.changes or "", + "date_published": c.created.isoformat() + "Z", + } + for c in rows + ], + } + from flask import jsonify + + response = jsonify(payload) + response.headers["Content-Type"] = "application/feed+json" + return response + + +def _recent_changes(): + conn = get_request_conn() + if conn is None: + abort(500) + return get_changes(conn, site_id=g.site.id, limit=20) -def changes_atom(site_slug: str) -> str: - return f"site:{site_slug} site/changes.atom GET (TODO)" +def _absolute_url(path: str) -> str: + """Build an absolute URL inside this site for a feed entry.""" + return f"{request.scheme}://{request.host}{site_root()}{path}" def hide_primer(site_slug: str) -> str: diff --git a/tests/test_db_queries.py b/tests/test_db_queries.py index 8a197fc..9947c12 100644 --- a/tests/test_db_queries.py +++ b/tests/test_db_queries.py @@ -13,10 +13,13 @@ delete_site, designs, drafts, + get_changes, get_design, get_page, get_pages_for_export, get_revision, + get_revisions, + get_revisions_count, get_site, is_public_url_available, new_page, @@ -655,3 +658,110 @@ def test_get_pages_for_export_scoped_to_site(db_conn: Connection) -> None: names = [r.page_name for r in rows] assert "b-page" not in names + + +# ---- get_revisions / get_revisions_count ---- + + +def test_get_revisions_returns_newest_first(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="v1", secret_url="gr1") + page = get_page(db_conn, site_id=site_id, page_name="") + assert page is not None + update_page(db_conn, page_id=page.id, content="v2") + update_page(db_conn, page_id=page.id, content="v3") + + rows = get_revisions(db_conn, page_id=page.id) + + assert [r.revision for r in rows] == [3, 2, 1] + assert rows[0].content == "v3" + + +def test_get_revisions_limit_caps_results(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="r1", secret_url="gr2") + page = get_page(db_conn, site_id=site_id, page_name="") + assert page is not None + for n in range(2, 6): + update_page(db_conn, page_id=page.id, content=f"r{n}") + + rows = get_revisions(db_conn, page_id=page.id, limit=2) + + assert len(rows) == 2 + assert [r.revision for r in rows] == [5, 4] + + +def test_get_revisions_before_pages_through_history(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="r1", secret_url="gr3") + page = get_page(db_conn, site_id=site_id, page_name="") + assert page is not None + for n in range(2, 6): + update_page(db_conn, page_id=page.id, content=f"r{n}") + + page_one = get_revisions(db_conn, page_id=page.id, limit=2) + page_two = get_revisions(db_conn, page_id=page.id, before=page_one[-1].revision, limit=2) + + assert [r.revision for r in page_one] == [5, 4] + assert [r.revision for r in page_two] == [3, 2] + + +def test_get_revisions_count_counts_non_sentinel_revisions(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="r1", secret_url="gr4") + page = get_page(db_conn, site_id=site_id, page_name="") + assert page is not None + update_page(db_conn, page_id=page.id, content="r2") + update_page(db_conn, page_id=page.id, content="r3") + + assert get_revisions_count(db_conn, page_id=page.id) == 3 + + +# ---- get_changes ---- + + +def test_get_changes_returns_revisions_across_all_pages(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="home v1", secret_url="ch1") + new_page(db_conn, site_id=site_id, name="notes", content="notes v1") + + rows = get_changes(db_conn, site_id=site_id) + + by_page = {(r.page_name, r.revision): r for r in rows} + assert ("", 1) in by_page + assert ("notes", 1) in by_page + + +def test_get_changes_orders_by_created_desc(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="home v1", secret_url="ch2") + home = get_page(db_conn, site_id=site_id, page_name="") + assert home is not None + update_page(db_conn, page_id=home.id, content="home v2") + new_page(db_conn, site_id=site_id, name="late", content="late v1") + + rows = get_changes(db_conn, site_id=site_id) + + # Newest created first; the just-inserted "late" page leads. + assert rows[0].page_name == "late" + + +def test_get_changes_limit_and_before_paginate(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="home v1", secret_url="ch3") + home = get_page(db_conn, site_id=site_id, page_name="") + assert home is not None + for n in range(2, 6): + update_page(db_conn, page_id=home.id, content=f"home v{n}") + + page_one = get_changes(db_conn, site_id=site_id, limit=2) + page_two = get_changes(db_conn, site_id=site_id, before=page_one[-1].id, limit=2) + + assert len(page_one) == 2 + assert len(page_two) == 2 + page_one_ids = {r.id for r in page_one} + page_two_ids = {r.id for r in page_two} + assert not (page_one_ids & page_two_ids) + + +def test_get_changes_scoped_to_site(db_conn: Connection) -> None: + site_a = new_site(db_conn, content="a", secret_url="ch4a") + site_b = new_site(db_conn, content="b", secret_url="ch4b") + new_page(db_conn, site_id=site_b, name="b-page", content="b-only") + + rows = get_changes(db_conn, site_id=site_a) + + assert all(r.page_name != "b-page" for r in rows) diff --git a/tests/test_feeds.py b/tests/test_feeds.py new file mode 100644 index 0000000..55804d1 --- /dev/null +++ b/tests/test_feeds.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import json +from collections.abc import Iterator + +import pytest +from flask.testing import FlaskClient +from sqlalchemy import Engine + +from jottit.auth import hash_password +from jottit.db import claim_site, get_page, metadata, new_page, new_site, update_page + +APEX = "http://jottit.test/" + + +@pytest.fixture(autouse=True) +def _truncate(db_engine: Engine) -> Iterator[None]: + yield + with db_engine.begin() as conn: + for table in reversed(metadata.sorted_tables): + conn.execute(table.delete()) + + +def _seed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + security: str | None = None, + password: str | None = None, +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content="hi", secret_url=secret_url, public_url=public_url) + if password is not None: + claim_site( + conn, + site_id=site_id, + password_hash=hash_password(password), + email="o@example.com", + security=security or "private", + ) + return site_id + + +# ---- /?m=history_rss ---- + + +def test_history_rss_returns_rss(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="hf1", public_url="alpha") + with db_engine.begin() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + update_page(conn, page_id=page.id, content="v2") + + response = client.get("/?m=history_rss", base_url="http://alpha.jottit.test/") + + assert response.status_code == 200 + assert response.mimetype == "application/rss+xml" + body = response.data.decode() + assert "Revision 2" in body + assert "Revision 1" in body + # Self-link uses an absolute URL pointing at this feed. + assert "?m=history_rss" in body + + +# ---- /?m=history_json ---- + + +def test_history_json_returns_jsonfeed(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="hf2", public_url="beta") + with db_engine.begin() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + update_page(conn, page_id=page.id, content="v2") + + response = client.get("/?m=history_json", base_url="http://beta.jottit.test/") + + assert response.status_code == 200 + assert response.mimetype == "application/feed+json" + payload = json.loads(response.data) + assert payload["version"].startswith("https://jsonfeed.org/version/") + assert len(payload["items"]) == 2 + assert payload["items"][0]["title"] == "Revision 2" + assert payload["items"][1]["title"] == "Revision 1" + assert payload["items"][0]["url"].endswith("?r=2") + + +# ---- /site/changes.rss ---- + + +def test_site_changes_rss(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="cf1", public_url="gamma") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="notes body") + + response = client.get("/site/changes.rss", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + assert response.mimetype == "application/rss+xml" + body = response.data.decode() + assert " None: + site_id = _seed_site(db_engine, secret_url="cf2", public_url="delta") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="notes body") + + response = client.get("/site/changes.json", base_url="http://delta.jottit.test/") + + assert response.status_code == 200 + assert response.mimetype == "application/feed+json" + payload = json.loads(response.data) + assert payload["version"].startswith("https://jsonfeed.org/version/") + item_urls = [item["url"] for item in payload["items"]] + assert any("notes" in url for url in item_urls) + + +# ---- Auth: private gates feeds; public/open allow them ---- + + +def test_feeds_on_private_site_redirect_to_signin(client: FlaskClient, db_engine: Engine) -> None: + _seed_site( + db_engine, secret_url="cf3", public_url="epsilon", security="private", password="hunter2" + ) + + response = client.get("/site/changes.rss", base_url="http://epsilon.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_feeds_on_public_site_allowed_anonymously(client: FlaskClient, db_engine: Engine) -> None: + _seed_site( + db_engine, secret_url="cf4", public_url="zeta", security="public", password="hunter2" + ) + + rss = client.get("/site/changes.rss", base_url="http://zeta.jottit.test/") + js = client.get("/site/changes.json", base_url="http://zeta.jottit.test/") + + assert rss.status_code == 200 + assert js.status_code == 200 + + +def test_history_feeds_on_public_site_allowed_anonymously( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site(db_engine, secret_url="cf5", public_url="eta", security="public", password="hunter2") + + rss = client.get("/?m=history_rss", base_url="http://eta.jottit.test/") + js = client.get("/?m=history_json", base_url="http://eta.jottit.test/") + + assert rss.status_code == 200 + assert js.status_code == 200 + + +# ---- Secret-URL routing ---- + + +def test_changes_rss_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="abc12") + + response = client.get("/abc12/site/changes.rss", base_url=APEX) + + assert response.status_code == 200 + body = response.data.decode() + # Self-link points at the secret URL. + assert "/abc12/site/changes.rss" in body + + +def test_history_json_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="abc34") + + response = client.get("/abc34/?m=history_json", base_url=APEX) + + assert response.status_code == 200 + payload = json.loads(response.data) + assert payload["home_page_url"].endswith("/abc34/") diff --git a/tests/test_page_diff.py b/tests/test_page_diff.py new file mode 100644 index 0000000..7fd62d9 --- /dev/null +++ b/tests/test_page_diff.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from flask.testing import FlaskClient +from sqlalchemy import Engine + +from jottit.auth import hash_password +from jottit.db import claim_site, get_page, metadata, new_site, update_page + +APEX = "http://jottit.test/" + + +@pytest.fixture(autouse=True) +def _truncate(db_engine: Engine) -> Iterator[None]: + yield + with db_engine.begin() as conn: + for table in reversed(metadata.sorted_tables): + conn.execute(table.delete()) + + +def _seed_site_with_revisions( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + revisions_content: list[str] | None = None, + security: str | None = None, + password: str | None = None, +) -> int: + revisions_content = revisions_content or ["v1", "v2", "v3"] + with db_engine.begin() as conn: + site_id = new_site( + conn, content=revisions_content[0], secret_url=secret_url, public_url=public_url + ) + if password is not None: + claim_site( + conn, + site_id=site_id, + password_hash=hash_password(password), + email="o@example.com", + security=security or "private", + ) + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + for body in revisions_content[1:]: + update_page(conn, page_id=page.id, content=body) + return site_id + + +def _sign_in(client: FlaskClient, *, base_url: str, site_id: int) -> None: + with client.session_transaction(base_url=base_url) as sess: + sess["signed_in_sites"] = [*sess.get("signed_in_sites", []), site_id] + + +# ---- Happy path ---- + + +def test_diff_with_no_r_compares_latest_vs_previous(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions( + db_engine, + secret_url="d1", + public_url="alpha", + revisions_content=["hello world", "hello there"], + ) + + response = client.get("/?m=diff", base_url="http://alpha.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + assert "Comparing" in body + assert "revision 1" in body + assert "revision 2" in body + assert "world" in body + assert "there" in body + + +def test_diff_with_single_r_compares_that_revision_to_previous( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site_with_revisions( + db_engine, + secret_url="d2", + public_url="beta", + revisions_content=["a", "b", "c"], + ) + + response = client.get("/?m=diff&r=2", base_url="http://beta.jottit.test/") + + body = response.data.decode() + assert "revision 1" in body + assert "revision 2" in body + assert "revision 3" not in body + + +def test_diff_with_two_r_compares_those_revisions(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions( + db_engine, + secret_url="d3", + public_url="gamma", + revisions_content=["one", "two", "three", "four"], + ) + + response = client.get("/?m=diff&r=1&r=4", base_url="http://gamma.jottit.test/") + + body = response.data.decode() + assert "revision 1" in body + assert "revision 4" in body + assert "one" in body + assert "four" in body + + +def test_diff_normalizes_revision_order(client: FlaskClient, db_engine: Engine) -> None: + """?r=3&r=1 should produce the same diff as ?r=1&r=3.""" + _seed_site_with_revisions( + db_engine, + secret_url="d4", + public_url="delta", + revisions_content=["one", "two", "three"], + ) + + swapped = client.get("/?m=diff&r=3&r=1", base_url="http://delta.jottit.test/") + correct = client.get("/?m=diff&r=1&r=3", base_url="http://delta.jottit.test/") + + assert swapped.data == correct.data + + +def test_diff_for_first_revision_only_shows_self(client: FlaskClient, db_engine: Engine) -> None: + """`?r=1` clamps the implicit prev-revision to 1 too — both sides become rev 1.""" + _seed_site_with_revisions( + db_engine, secret_url="d5", public_url="epsilon", revisions_content=["only"] + ) + + response = client.get("/?m=diff&r=1", base_url="http://epsilon.jottit.test/") + + assert response.status_code == 200 + + +# ---- Bad inputs ---- + + +def test_diff_too_many_revisions_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions(db_engine, secret_url="d6", public_url="zeta") + + response = client.get("/?m=diff&r=1&r=2&r=3", base_url="http://zeta.jottit.test/") + + assert response.status_code == 400 + + +def test_diff_non_integer_revision_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions(db_engine, secret_url="d7", public_url="eta") + + response = client.get("/?m=diff&r=banana", base_url="http://eta.jottit.test/") + + assert response.status_code == 400 + + +def test_diff_unknown_revision_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions(db_engine, secret_url="d8", public_url="theta") + + response = client.get("/?m=diff&r=99", base_url="http://theta.jottit.test/") + + assert response.status_code == 400 + + +def test_diff_for_unknown_page_404s(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions(db_engine, secret_url="d9", public_url="iota") + + response = client.get("/no-such-page?m=diff", base_url="http://iota.jottit.test/") + + assert response.status_code == 404 + + +# ---- Auth gating ---- + + +def test_diff_on_public_site_redirects_anonymous_to_signin( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site_with_revisions( + db_engine, + secret_url="d10", + public_url="kappa", + security="public", + password="hunter2", + ) + + response = client.get("/?m=diff", base_url="http://kappa.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_diff_on_open_site_allowed_anonymously(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions( + db_engine, secret_url="d11", public_url="lambda", security="open", password="hunter2" + ) + + response = client.get("/?m=diff", base_url="http://lambda.jottit.test/") + + assert response.status_code == 200 + + +# ---- Secret-URL routing ---- + + +def test_diff_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + _seed_site_with_revisions(db_engine, secret_url="abc12") + + response = client.get("/abc12/?m=diff", base_url=APEX) + + assert response.status_code == 200 + assert 'href="/abc12/?r=' in response.data.decode() diff --git a/tests/test_page_history.py b/tests/test_page_history.py new file mode 100644 index 0000000..a60160a --- /dev/null +++ b/tests/test_page_history.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from flask.testing import FlaskClient +from sqlalchemy import Engine + +from jottit.auth import hash_password +from jottit.db import claim_site, get_page, metadata, new_page, new_site, update_page + +APEX = "http://jottit.test/" + + +@pytest.fixture(autouse=True) +def _truncate(db_engine: Engine) -> Iterator[None]: + yield + with db_engine.begin() as conn: + for table in reversed(metadata.sorted_tables): + conn.execute(table.delete()) + + +def _seed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + security: str | None = None, + password: str | None = None, +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content="v1", secret_url=secret_url, public_url=public_url) + if password is not None: + claim_site( + conn, + site_id=site_id, + password_hash=hash_password(password), + email="o@example.com", + security=security or "private", + ) + return site_id + + +def _sign_in(client: FlaskClient, *, base_url: str, site_id: int) -> None: + with client.session_transaction(base_url=base_url) as sess: + sess["signed_in_sites"] = [*sess.get("signed_in_sites", []), site_id] + + +def _make_revisions(db_engine: Engine, site_id: int, page_name: str, count: int) -> None: + """Land `count` revisions on the named page (including the seed).""" + with db_engine.begin() as conn: + if page_name: + new_page(conn, site_id=site_id, name=page_name, content="v1") + page = get_page(conn, site_id=site_id, page_name=page_name) + assert page is not None + for n in range(2, count + 1): + update_page(conn, page_id=page.id, content=f"v{n}") + + +# ---- Happy path ---- + + +def test_history_lists_revisions_newest_first(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="h1", public_url="alpha") + _make_revisions(db_engine, site_id, "", count=3) + + response = client.get("/?m=history", base_url="http://alpha.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + # Newest revision number appears before older ones. + assert body.index("Revision 3") < body.index("Revision 2") < body.index("Revision 1") + + +def test_history_renders_total_count(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="h2", public_url="beta") + _make_revisions(db_engine, site_id, "", count=4) + + response = client.get("/?m=history", base_url="http://beta.jottit.test/") + + assert "4 revisions" in response.data.decode() + + +def test_history_singular_count_for_one_revision(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="h3", public_url="gamma") + + response = client.get("/?m=history", base_url="http://gamma.jottit.test/") + + assert "1 revision." in response.data.decode() + + +def test_history_for_named_page(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="h4", public_url="delta") + _make_revisions(db_engine, site_id, "notes", count=2) + + response = client.get("/notes?m=history", base_url="http://delta.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + assert "History: notes" in body + assert "Revision 2" in body + + +# ---- Pagination ---- + + +def test_history_first_page_capped_at_twenty(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="h5", public_url="epsilon") + _make_revisions(db_engine, site_id, "", count=25) + + response = client.get("/?m=history", base_url="http://epsilon.jottit.test/") + + body = response.data.decode() + # Latest 20 shown (revisions 25..6), older link points back further. + assert "Revision 25" in body + assert "Revision 6" in body + assert "Revision 5" not in body + assert "Older" in body + assert "before=6" in body + + +def test_history_older_link_pages_back(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="h6", public_url="zeta") + _make_revisions(db_engine, site_id, "", count=25) + + response = client.get("/?m=history&before=6", base_url="http://zeta.jottit.test/") + + body = response.data.decode() + # Revisions 5..1 shown; no more "Older" link because that's the tail. + assert "Revision 5" in body + assert "Revision 1" in body + assert "Revision 6" not in body + assert "Older" not in body + + +def test_history_no_older_link_when_results_fit_in_one_page( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_site(db_engine, secret_url="h7", public_url="eta") + _make_revisions(db_engine, site_id, "", count=10) + + response = client.get("/?m=history", base_url="http://eta.jottit.test/") + + assert "Older" not in response.data.decode() + + +# ---- Missing / empty cases ---- + + +def test_history_for_unknown_page_returns_404(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="h8", public_url="theta") + + response = client.get("/no-such-page?m=history", base_url="http://theta.jottit.test/") + + assert response.status_code == 404 + + +# ---- Auth gating ---- + + +def test_history_on_private_site_redirects_to_signin( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site( + db_engine, secret_url="h9", public_url="iota", security="private", password="hunter2" + ) + + response = client.get("/?m=history", base_url="http://iota.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_history_on_public_site_redirects_anonymous_to_signin( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site( + db_engine, secret_url="h10", public_url="kappa", security="public", password="hunter2" + ) + + response = client.get("/?m=history", base_url="http://kappa.jottit.test/") + + # Public sites gate everything except plain `view` — history is + # under view_revision. + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_history_on_open_site_allowed_anonymously(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site( + db_engine, secret_url="h11", public_url="lambda", security="open", password="hunter2" + ) + _make_revisions(db_engine, site_id, "", count=2) + + response = client.get("/?m=history", base_url="http://lambda.jottit.test/") + + assert response.status_code == 200 + + +def test_history_on_private_site_with_signin_allowed( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_site( + db_engine, + secret_url="h12", + public_url="mu", + security="private", + password="hunter2", + ) + _make_revisions(db_engine, site_id, "", count=2) + _sign_in(client, base_url="http://mu.jottit.test/", site_id=site_id) + + response = client.get("/?m=history", base_url="http://mu.jottit.test/") + + assert response.status_code == 200 + + +# ---- Secret-URL routing ---- + + +def test_history_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="abc12") + _make_revisions(db_engine, site_id, "", count=2) + + response = client.get("/abc12/?m=history", base_url=APEX) + + assert response.status_code == 200 + body = response.data.decode() + # Revision links should be rooted at the secret URL prefix. + assert 'href="/abc12/?r=1"' in body + assert 'href="/abc12/?r=2"' in body diff --git a/tests/test_page_revert_undelete.py b/tests/test_page_revert_undelete.py new file mode 100644 index 0000000..abf85de --- /dev/null +++ b/tests/test_page_revert_undelete.py @@ -0,0 +1,245 @@ +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from flask.testing import FlaskClient +from sqlalchemy import Engine + +from jottit.auth import hash_password +from jottit.db import ( + claim_site, + delete_page, + get_page, + get_revision, + metadata, + new_page, + new_site, + update_page, +) + +APEX = "http://jottit.test/" + + +@pytest.fixture(autouse=True) +def _truncate(db_engine: Engine) -> Iterator[None]: + yield + with db_engine.begin() as conn: + for table in reversed(metadata.sorted_tables): + conn.execute(table.delete()) + + +def _seed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + security: str | None = None, + password: str | None = None, +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content="v1", secret_url=secret_url, public_url=public_url) + if password is not None: + claim_site( + conn, + site_id=site_id, + password_hash=hash_password(password), + email="o@example.com", + security=security or "private", + ) + return site_id + + +def _sign_in(client: FlaskClient, *, base_url: str, site_id: int) -> None: + with client.session_transaction(base_url=base_url) as sess: + sess["signed_in_sites"] = [*sess.get("signed_in_sites", []), site_id] + + +# ---- revert ---- + + +def test_revert_writes_new_revision_with_old_content( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_site(db_engine, secret_url="r1", public_url="alpha") + with db_engine.begin() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + update_page(conn, page_id=page.id, content="v2") + update_page(conn, page_id=page.id, content="v3") + + response = client.post("/?m=revert", base_url="http://alpha.jottit.test/", data={"r": "1"}) + + assert response.status_code == 303 + with db_engine.connect() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + latest = get_revision(conn, page_id=page.id) + assert latest is not None + assert latest.revision == 4 + assert latest.content == "v1" + + +def test_revert_to_current_content_is_noop(client: FlaskClient, db_engine: Engine) -> None: + """If the target revision's content already matches latest, no new revision is created.""" + site_id = _seed_site(db_engine, secret_url="r2", public_url="beta") + with db_engine.begin() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + update_page(conn, page_id=page.id, content="same") + + # Revision 2 has "same"; reverting to it shouldn't create revision 3. + client.post("/?m=revert", base_url="http://beta.jottit.test/", data={"r": "2"}) + + with db_engine.connect() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + latest = get_revision(conn, page_id=page.id) + assert latest is not None + assert latest.revision == 2 + + +def test_revert_undeletes_a_deleted_page(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="r3", public_url="gamma") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="kept") + page = get_page(conn, site_id=site_id, page_name="notes") + assert page is not None + delete_page(conn, page_id=page.id) + + client.post("/notes?m=revert", base_url="http://gamma.jottit.test/", data={"r": "1"}) + + with db_engine.connect() as conn: + page = get_page(conn, site_id=site_id, page_name="notes") + assert page is not None + assert page.deleted is False + + +def test_revert_missing_r_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="r4", public_url="delta") + + response = client.post("/?m=revert", base_url="http://delta.jottit.test/", data={}) + + assert response.status_code == 400 + + +def test_revert_unknown_revision_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="r5", public_url="epsilon") + + response = client.post("/?m=revert", base_url="http://epsilon.jottit.test/", data={"r": "99"}) + + assert response.status_code == 400 + + +def test_revert_unknown_page_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="r6", public_url="zeta") + + response = client.post( + "/no-such-page?m=revert", base_url="http://zeta.jottit.test/", data={"r": "1"} + ) + + assert response.status_code == 400 + + +def test_revert_redirects_to_page(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="r7", public_url="eta") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="x") + page = get_page(conn, site_id=site_id, page_name="notes") + assert page is not None + update_page(conn, page_id=page.id, content="y") + + response = client.post("/notes?m=revert", base_url="http://eta.jottit.test/", data={"r": "1"}) + + assert response.status_code == 303 + assert response.headers["Location"] == "/notes" + + +# ---- undelete ---- + + +def test_undelete_restores_deleted_page_content(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="u1", public_url="theta") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="alive") + page = get_page(conn, site_id=site_id, page_name="notes") + assert page is not None + delete_page(conn, page_id=page.id) + + response = client.post("/notes?m=undelete", base_url="http://theta.jottit.test/") + + assert response.status_code == 303 + with db_engine.connect() as conn: + page = get_page(conn, site_id=site_id, page_name="notes") + assert page is not None + assert page.deleted is False + latest = get_revision(conn, page_id=page.id) + assert latest is not None + assert latest.content == "alive" + assert latest.changes == "Delete undone." + + +def test_undelete_unknown_page_returns_400(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="u2", public_url="iota") + + response = client.post("/no-such-page?m=undelete", base_url="http://iota.jottit.test/") + + assert response.status_code == 400 + + +# ---- Auth gating ---- + + +def test_revert_returns_401_anonymous_on_private(client: FlaskClient, db_engine: Engine) -> None: + _seed_site( + db_engine, secret_url="u3", public_url="kappa", security="private", password="hunter2" + ) + + response = client.post("/?m=revert", base_url="http://kappa.jottit.test/", data={"r": "1"}) + + assert response.status_code == 401 + + +def test_revert_succeeds_anonymously_on_open_site(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site( + db_engine, secret_url="u4", public_url="lambda", security="open", password="hunter2" + ) + with db_engine.begin() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + update_page(conn, page_id=page.id, content="v2") + + response = client.post("/?m=revert", base_url="http://lambda.jottit.test/", data={"r": "1"}) + + assert response.status_code == 303 + + +def test_undelete_returns_401_anonymous_on_public(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site( + db_engine, secret_url="u5", public_url="mu", security="public", password="hunter2" + ) + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="x") + page = get_page(conn, site_id=site_id, page_name="notes") + assert page is not None + delete_page(conn, page_id=page.id) + + response = client.post("/notes?m=undelete", base_url="http://mu.jottit.test/") + + assert response.status_code == 401 + + +# ---- Secret-URL routing ---- + + +def test_revert_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="abc12") + with db_engine.begin() as conn: + page = get_page(conn, site_id=site_id, page_name="") + assert page is not None + update_page(conn, page_id=page.id, content="v2") + + response = client.post("/abc12/?m=revert", base_url=APEX, data={"r": "1"}) + + assert response.status_code == 303 + assert response.headers["Location"] == "/abc12/" diff --git a/tests/test_secret_routing.py b/tests/test_secret_routing.py index c4f4bd0..e3fdc90 100644 --- a/tests/test_secret_routing.py +++ b/tests/test_secret_routing.py @@ -10,11 +10,9 @@ ("method", "path", "expected_substring"), [ # /site/claim, /site/signin, /site/signout, /site/forgot-password, - # and /site/change-password are exercised end-to-end in - # tests/test_claim.py, tests/test_signin.py, and - # tests/test_password_recovery.py. - ("GET", "/abc123/site/changes", "site/changes GET"), - ("GET", "/abc123/site/changes.atom", "site/changes.atom GET"), + # /site/change-password, /site/changes, /site/changes.rss, and + # /site/changes.json are exercised end-to-end in their respective + # test files. ("POST", "/abc123/site/hide-primer", "site/hide-primer POST"), ], ) diff --git a/tests/test_site_changes.py b/tests/test_site_changes.py new file mode 100644 index 0000000..27f4345 --- /dev/null +++ b/tests/test_site_changes.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from flask.testing import FlaskClient +from sqlalchemy import Engine + +from jottit.auth import hash_password +from jottit.db import ( + claim_site, + delete_page, + get_page, + metadata, + new_page, + new_site, + update_page, +) + +APEX = "http://jottit.test/" + + +@pytest.fixture(autouse=True) +def _truncate(db_engine: Engine) -> Iterator[None]: + yield + with db_engine.begin() as conn: + for table in reversed(metadata.sorted_tables): + conn.execute(table.delete()) + + +def _seed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + security: str | None = None, + password: str | None = None, +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content="hi", secret_url=secret_url, public_url=public_url) + if password is not None: + claim_site( + conn, + site_id=site_id, + password_hash=hash_password(password), + email="o@example.com", + security=security or "private", + ) + return site_id + + +def _sign_in(client: FlaskClient, *, base_url: str, site_id: int) -> None: + with client.session_transaction(base_url=base_url) as sess: + sess["signed_in_sites"] = [*sess.get("signed_in_sites", []), site_id] + + +# ---- Happy path ---- + + +def test_changes_lists_recent_revisions_across_pages( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_site(db_engine, secret_url="c1", public_url="alpha") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="notes body") + home = get_page(conn, site_id=site_id, page_name="") + assert home is not None + update_page(conn, page_id=home.id, content="home v2") + + response = client.get("/site/changes", base_url="http://alpha.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + # Both pages appear. + assert "notes" in body + assert "Home" in body + # Newest first: the home v2 update has the latest timestamp. + assert body.index("Home") < body.index("notes") + + +def test_changes_renders_links_to_specific_revisions( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_site(db_engine, secret_url="c2", public_url="beta") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="My Notes", content="x") + + response = client.get("/site/changes", base_url="http://beta.jottit.test/") + + body = response.data.decode() + # Page name gets slugified to my_notes in the URL. + assert "my_notes?r=1" in body + + +def test_changes_empty_site_shows_only_seed_revision( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site(db_engine, secret_url="c3", public_url="gamma") + + response = client.get("/site/changes", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + # new_site seeds revision 1 of the home page. + assert "revision 1" in response.data.decode() + + +def test_changes_marks_deleted_pages(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="c4", public_url="delta") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="gone", content="bye") + page = get_page(conn, site_id=site_id, page_name="gone") + assert page is not None + delete_page(conn, page_id=page.id) + + response = client.get("/site/changes", base_url="http://delta.jottit.test/") + + assert "page since deleted" in response.data.decode() + + +# ---- Pagination ---- + + +def test_changes_page_size_caps_at_twenty(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="c5", public_url="epsilon") + with db_engine.begin() as conn: + home = get_page(conn, site_id=site_id, page_name="") + assert home is not None + for n in range(2, 26): + update_page(conn, page_id=home.id, content=f"v{n}") + + response = client.get("/site/changes", base_url="http://epsilon.jottit.test/") + + body = response.data.decode() + # We have 25 revisions; the page should show 20 and offer an Older link. + assert "Older" in body + assert "before=" in body + + +def test_changes_older_link_pages_back(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_site(db_engine, secret_url="c6", public_url="zeta") + with db_engine.begin() as conn: + home = get_page(conn, site_id=site_id, page_name="") + assert home is not None + for n in range(2, 26): + update_page(conn, page_id=home.id, content=f"v{n}") + + first = client.get("/site/changes", base_url="http://zeta.jottit.test/") + # Extract the `before=` value from the first page's Older link. + import re + + match = re.search(r"before=(\d+)", first.data.decode()) + assert match is not None + before = match.group(1) + + second = client.get(f"/site/changes?before={before}", base_url="http://zeta.jottit.test/") + + assert second.status_code == 200 + assert "Older" not in second.data.decode() + + +# ---- Auth gating ---- + + +def test_changes_on_private_site_redirects_to_signin( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_site(db_engine, secret_url="c7", public_url="eta", security="private", password="hunter2") + + response = client.get("/site/changes", base_url="http://eta.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_changes_on_public_site_redirects_to_signin(client: FlaskClient, db_engine: Engine) -> None: + _seed_site( + db_engine, secret_url="c8", public_url="theta", security="public", password="hunter2" + ) + + response = client.get("/site/changes", base_url="http://theta.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_changes_on_open_site_allowed_anonymously(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="c9", public_url="iota", security="open", password="hunter2") + + response = client.get("/site/changes", base_url="http://iota.jottit.test/") + + assert response.status_code == 200 + + +# ---- Secret-URL routing ---- + + +def test_changes_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + _seed_site(db_engine, secret_url="abc12") + + response = client.get("/abc12/site/changes", base_url=APEX) + + assert response.status_code == 200 + # Revision links should resolve under the secret prefix. + assert 'href="/abc12/?r=' in response.data.decode() diff --git a/tests/test_subdomain_routing.py b/tests/test_subdomain_routing.py index da88ccb..1d4206e 100644 --- a/tests/test_subdomain_routing.py +++ b/tests/test_subdomain_routing.py @@ -10,11 +10,9 @@ ("method", "path", "expected_substring"), [ # /site/claim, /site/signin, /site/signout, /site/forgot-password, - # and /site/change-password are exercised end-to-end in - # tests/test_claim.py, tests/test_signin.py, and - # tests/test_password_recovery.py. - ("GET", "/site/changes", "site/changes GET"), - ("GET", "/site/changes.atom", "site/changes.atom GET"), + # /site/change-password, /site/changes, /site/changes.rss, and + # /site/changes.json are exercised end-to-end in their respective + # test files. ("POST", "/site/hide-primer", "site/hide-primer POST"), ], )