{{ error }}
{% endif %} + + + + diff --git a/jottit/templates/admin_change_site_address.html b/jottit/templates/admin_change_site_address.html new file mode 100644 index 0000000..3dc546e --- /dev/null +++ b/jottit/templates/admin_change_site_address.html @@ -0,0 +1,21 @@ + + + + +{{ error }}
{% endif %} + + + + diff --git a/jottit/templates/admin_delete.html b/jottit/templates/admin_delete.html new file mode 100644 index 0000000..1bf449f --- /dev/null +++ b/jottit/templates/admin_delete.html @@ -0,0 +1,15 @@ + + + + +This will remove the site from public listings and free its subdomain. Pages and revisions stay on file in case we ever add a restore path.
+ + + + diff --git a/jottit/templates/admin_design.html b/jottit/templates/admin_design.html new file mode 100644 index 0000000..8e3c353 --- /dev/null +++ b/jottit/templates/admin_design.html @@ -0,0 +1,44 @@ + + + + +{{ error }}
{% endif %} + + + + diff --git a/jottit/templates/admin_settings.html b/jottit/templates/admin_settings.html new file mode 100644 index 0000000..5049629 --- /dev/null +++ b/jottit/templates/admin_settings.html @@ -0,0 +1,37 @@ + + + + +{{ error }}
{% endif %} + + + + + diff --git a/jottit/views/admin.py b/jottit/views/admin.py index ffe0120..324742a 100644 --- a/jottit/views/admin.py +++ b/jottit/views/admin.py @@ -1,31 +1,339 @@ from __future__ import annotations -from flask import request +import io +import re +import zipfile +from flask import abort, current_app, g, jsonify, redirect, render_template, request, send_file +from flask.typing import ResponseReturnValue +from sqlalchemy import Connection -def settings(site_slug: str) -> str: - return f"admin:{site_slug} admin/settings {request.method} (TODO)" +from jottit import auth +from jottit.db import ( + change_public_url, + delete_site, + get_design, + get_pages_for_export, + get_request_conn, + is_public_url_available, + set_password, + update_design, + update_site, +) +from jottit.urls import site_root +_ALLOWED_SECURITY_LEVELS = {"private", "public", "open"} +_PUBLIC_URL_RE = re.compile(r"^[a-z0-9-]+$") +_HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{3,8}$") -def design(site_slug: str) -> str: - return f"admin:{site_slug} admin/design {request.method} (TODO)" +# Field groupings for /admin/design. These tuples are the source of truth +# for the rendered form, the POST parser, and the validator. +_DESIGN_FONT_FIELDS = ("title_font", "subtitle_font", "headings_font", "content_font") +_DESIGN_COLOR_FIELDS = ("header_color", "title_color", "subtitle_color") +_DESIGN_SIZE_FIELDS = ("title_size", "subtitle_size", "headings_size", "content_size") -def url_available(site_slug: str) -> str: - return f"admin:{site_slug} admin/url-available POST (TODO)" +def settings(site_slug: str) -> ResponseReturnValue: + if (response := _gate_admin()) is not None: + return response + if request.method == "GET": + return render_template( + "admin_settings.html", + title=g.site.title or "", + subtitle=g.site.subtitle or "", + email=g.site.email or "", + security=g.site.security or "private", + public_url=g.site.public_url or "", + error=None, + ) -def delete(site_slug: str) -> str: - return f"admin:{site_slug} admin/delete {request.method} (TODO)" + title = request.form.get("title", "") + subtitle = request.form.get("subtitle", "") + email = request.form.get("email", "") + security = request.form.get("security", g.site.security or "private") + error = _validate_settings(email=email, security=security) + if error is not None: + return render_template( + "admin_settings.html", + title=title, + subtitle=subtitle, + email=email, + security=security, + public_url=g.site.public_url or "", + error=error, + ), 400 -def change_site_address(site_slug: str) -> str: - return f"admin:{site_slug} admin/change-site-address {request.method} (TODO)" + conn = _conn() + update_site( + conn, + site_id=g.site.id, + title=title, + subtitle=subtitle, + email=email, + security=security, + ) + return redirect(site_root(), code=303) -def change_password(site_slug: str) -> str: - return f"admin:{site_slug} admin/change-password {request.method} (TODO)" +def _validate_settings(*, email: str, security: str) -> str | None: + if email and "@" not in email: + return "Please enter a valid email address." + if security not in _ALLOWED_SECURITY_LEVELS: + return "Pick a valid security level." + return None -def export(site_slug: str) -> str: - return f"admin:{site_slug} admin/export GET (TODO)" +def _gate_admin() -> ResponseReturnValue | None: + if g.site is None: + abort(404) + return auth.gate("admin") + + +def design(site_slug: str) -> ResponseReturnValue: + if (response := _gate_admin()) is not None: + return response + + conn = _conn() + design_row = get_design(conn, site_id=g.site.id) + + if request.method == "GET": + return render_template( + "admin_design.html", + design=_design_view_model(design_row), + error=None, + ) + + submitted = {field: request.form.get(field, "") for field in _all_design_fields()} + error = _validate_design(submitted) + if error is not None: + return render_template( + "admin_design.html", + design=submitted, + error=error, + ), 400 + + update_design(conn, site_id=g.site.id, **_coerce_design(submitted)) + return redirect(f"{site_root()}admin/design", code=303) + + +def _all_design_fields() -> tuple[str, ...]: + return ( + *_DESIGN_FONT_FIELDS, + *_DESIGN_COLOR_FIELDS, + *_DESIGN_SIZE_FIELDS, + "hue", + "brightness", + ) + + +def _design_view_model(row: object) -> dict[str, str]: + """Surface design fields as strings for the template (sizes get str-coerced).""" + if row is None: + return {f: "" for f in _all_design_fields()} + return { + f: ("" if (v := getattr(row, f, None)) is None else str(v)) for f in _all_design_fields() + } + + +def _validate_design(values: dict[str, str]) -> str | None: + for f in _DESIGN_FONT_FIELDS: + # Free-form for now (M9 will introduce a real picker). Block stray HTML + # so a stored value can't smuggle markup into the rendered page. + if any(c in values[f] for c in "<>\"'"): + return f"{f.replace('_', ' ').capitalize()} contains invalid characters." + for f in _DESIGN_COLOR_FIELDS: + if values[f] and not _HEX_COLOR_RE.fullmatch(values[f]): + return f"{f.replace('_', ' ').capitalize()} must be a hex color like #fff or #abcdef." + for f in _DESIGN_SIZE_FIELDS: + if values[f] and not _is_int_in_range(values[f], 25, 500): + return f"{f.replace('_', ' ').capitalize()} must be a number between 25 and 500." + if values["hue"] and not _is_int_in_range(values["hue"], 0, 360): + return "Hue must be a number between 0 and 360." + if values["brightness"] and not _is_int_in_range(values["brightness"], 0, 300): + return "Brightness must be a number between 0 and 300." + return None + + +def _is_int_in_range(value: str, lo: int, hi: int) -> bool: + try: + n = int(value) + except ValueError: + return False + return lo <= n <= hi + + +def _coerce_design(values: dict[str, str]) -> dict[str, object]: + """Turn validated form strings into typed values for the DB layer. + + Empty strings are dropped so unchanged columns aren't overwritten with + nulls; sizes are coerced to int (the column is Integer). + """ + out: dict[str, object] = {} + for f in (*_DESIGN_FONT_FIELDS, *_DESIGN_COLOR_FIELDS, "hue", "brightness"): + if values[f]: + out[f] = values[f] + for f in _DESIGN_SIZE_FIELDS: + if values[f]: + out[f] = int(values[f]) + return out + + +def url_available(site_slug: str) -> ResponseReturnValue: + """JSON probe used by the change-site-address form: is this slug free?""" + if (response := _gate_admin()) is not None: + return response + + url = request.form.get("url", "").strip().lower() + if not url: + return jsonify(available=False) + + if not _PUBLIC_URL_RE.fullmatch(url): + return jsonify(available=False) + + conn = _conn() + return jsonify(available=is_public_url_available(conn, public_url=url)) + + +def delete(site_slug: str) -> ResponseReturnValue: + if (response := _gate_admin()) is not None: + return response + + if request.method == "GET": + return render_template("admin_delete.html") + + conn = _conn() + delete_site(conn, site_id=g.site.id) + auth.sign_out(g.site.id) + # The site no longer resolves at its old URL; bounce the user back to + # the apex homepage instead of an immediate 404 loop. + domain = current_app.config["SERVER_NAME"] + return redirect(f"{request.scheme}://{domain}/", code=303) + + +def change_site_address(site_slug: str) -> ResponseReturnValue: + if (response := _gate_admin()) is not None: + return response + + if request.method == "GET": + return render_template( + "admin_change_site_address.html", + public_url=g.site.public_url or "", + error=None, + ) + + new_url = request.form.get("public_url", "").strip().lower() + + # No-op: same slug as before — skip the DB write and just redirect home. + if new_url == (g.site.public_url or ""): + return redirect(_admin_settings_url(new_url), code=303) + + conn = _conn() + error = _validate_public_url(new_url, conn) + if error is not None: + return render_template( + "admin_change_site_address.html", + public_url=new_url, + error=error, + ), 400 + + change_public_url(conn, site_id=g.site.id, public_url=new_url) + return redirect(_admin_settings_url(new_url), code=303) + + +def _validate_public_url(url: str, conn: Connection) -> str | None: + if not url: + # Empty value clears the slug — site stays reachable via secret URL only. + return None + if not _PUBLIC_URL_RE.fullmatch(url): + return "Site address can only contain lowercase letters, numbers, and dashes." + if not is_public_url_available(conn, public_url=url): + return "That site address is already taken." + return None + + +def _admin_settings_url(new_public_url: str) -> str: + """Absolute URL for /admin/settings after a public_url change. + + Goes to the new subdomain when a public_url is set, otherwise to the + secret-URL path. Cross-subdomain redirect is required because the + current request's host (the OLD subdomain) no longer resolves to a + site after the DB update. + """ + scheme = request.scheme + domain = current_app.config["SERVER_NAME"] + if new_public_url: + return f"{scheme}://{new_public_url}.{domain}/admin/settings" + return f"{scheme}://{domain}/{g.site.secret_url}/admin/settings" + + +def _conn() -> Connection: + conn = get_request_conn() + if conn is None: + abort(500) + return conn + + +def change_password(site_slug: str) -> ResponseReturnValue: + """Change the site password while signed in. + + Distinct from the token-based recovery flow in /site/change-password — + this path requires the current password to be supplied and re-typed. + """ + if (response := _gate_admin()) is not None: + return response + + if request.method == "GET": + return render_template("admin_change_password.html", error=None) + + current_password = request.form.get("current_password", "") + new_password = request.form.get("new_password", "") + + if not auth.verify_password(current_password, g.site.password): + return render_template( + "admin_change_password.html", + error="That isn't your current password.", + ), 401 + + if not new_password: + return render_template( + "admin_change_password.html", + error="Please enter a new password.", + ), 400 + + conn = _conn() + set_password(conn, site_id=g.site.id, password_hash=auth.hash_password(new_password)) + return redirect(f"{site_root()}admin/settings", code=303) + + +def export(site_slug: str) -> ResponseReturnValue: + """Stream a zip with one .md per page (latest revision only) as a download.""" + if (response := _gate_admin()) is not None: + return response + + conn = _conn() + rows = get_pages_for_export(conn, site_id=g.site.id) + + buf = io.BytesIO() + with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: + for row in rows: + zf.writestr(_export_filename(row.page_name), row.content) + buf.seek(0) + + slug = g.site.public_url or g.site.secret_url + return send_file( + buf, + mimetype="application/zip", + as_attachment=True, + download_name=f"{slug}-export.zip", + ) + + +def _export_filename(page_name: str) -> str: + """Filename inside the export zip. The empty name (home page) becomes 'home.md'.""" + safe = page_name or "home" + # Keep names predictable: stripped of path separators and trailing dots + # (Windows hates trailing dots on filenames). + safe = safe.replace("/", "_").replace("\\", "_").rstrip(".") or "home" + return f"{safe}.md" diff --git a/tests/test_admin_change_password.py b/tests/test_admin_change_password.py new file mode 100644 index 0000000..5302a0b --- /dev/null +++ b/tests/test_admin_change_password.py @@ -0,0 +1,173 @@ +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, verify_password +from jottit.db import claim_site, get_site, metadata, new_site + +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_claimed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + password: str = "current-pw", +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content="hi", secret_url=secret_url, public_url=public_url) + claim_site( + conn, + site_id=site_id, + password_hash=hash_password(password), + email="owner@example.com", + security="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] + + +# ---- Auth gating ---- + + +def test_get_redirects_when_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="c1", public_url="alpha") + + response = client.get("/admin/change-password", base_url="http://alpha.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_post_returns_401_when_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="c2", public_url="beta") + + response = client.post( + "/admin/change-password", + base_url="http://beta.jottit.test/", + data={"current_password": "current-pw", "new_password": "new-pw"}, + ) + + assert response.status_code == 401 + + +# ---- GET ---- + + +def test_get_renders_form(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="c3", public_url="gamma") + _sign_in(client, base_url="http://gamma.jottit.test/", site_id=site_id) + + response = client.get("/admin/change-password", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + assert 'name="current_password"' in body + assert 'name="new_password"' in body + + +# ---- POST ---- + + +def test_post_changes_password_when_current_correct(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="c4", public_url="delta") + _sign_in(client, base_url="http://delta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-password", + base_url="http://delta.jottit.test/", + data={"current_password": "current-pw", "new_password": "shiny-new"}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "/admin/settings" + + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert verify_password("shiny-new", row.password) + assert not verify_password("current-pw", row.password) + + +def test_post_rejects_wrong_current_password(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="c5", public_url="epsilon") + _sign_in(client, base_url="http://epsilon.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-password", + base_url="http://epsilon.jottit.test/", + data={"current_password": "WRONG", "new_password": "shiny-new"}, + ) + + assert response.status_code == 401 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + # Password unchanged. + assert verify_password("current-pw", row.password) + + +def test_post_rejects_empty_new_password(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="c6", public_url="zeta") + _sign_in(client, base_url="http://zeta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-password", + base_url="http://zeta.jottit.test/", + data={"current_password": "current-pw", "new_password": ""}, + ) + + assert response.status_code == 400 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert verify_password("current-pw", row.password) + + +def test_post_keeps_user_signed_in(client: FlaskClient, db_engine: Engine) -> None: + """Sessions are keyed by site_id, not password — a password change shouldn't bump the user out.""" + site_id = _seed_claimed_site(db_engine, secret_url="c7", public_url="eta") + _sign_in(client, base_url="http://eta.jottit.test/", site_id=site_id) + + client.post( + "/admin/change-password", + base_url="http://eta.jottit.test/", + data={"current_password": "current-pw", "new_password": "next-pw"}, + ) + + with client.session_transaction(base_url="http://eta.jottit.test/") as sess: + assert site_id in sess["signed_in_sites"] + + +# ---- Secret-URL routing ---- + + +def test_change_password_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="abc12") + _sign_in(client, base_url=APEX, site_id=site_id) + + response = client.post( + "/abc12/admin/change-password", + base_url=APEX, + data={"current_password": "current-pw", "new_password": "next-pw"}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "/abc12/admin/settings" diff --git a/tests/test_admin_change_site_address.py b/tests/test_admin_change_site_address.py new file mode 100644 index 0000000..b527bb0 --- /dev/null +++ b/tests/test_admin_change_site_address.py @@ -0,0 +1,326 @@ +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_site, metadata, new_site + +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_claimed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: 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) + claim_site( + conn, + site_id=site_id, + password_hash=hash_password("hunter2"), + email="owner@example.com", + security="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] + + +# ---- Auth gating ---- + + +def test_get_change_site_address_redirects_anonymous( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_claimed_site(db_engine, secret_url="a1", public_url="alpha") + + response = client.get("/admin/change-site-address", base_url="http://alpha.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_post_change_site_address_returns_401_anonymous( + client: FlaskClient, db_engine: Engine +) -> None: + _seed_claimed_site(db_engine, secret_url="a2", public_url="beta") + + response = client.post( + "/admin/change-site-address", + base_url="http://beta.jottit.test/", + data={"public_url": "newbeta"}, + ) + + assert response.status_code == 401 + + +# ---- GET ---- + + +def test_get_change_site_address_prefills_current_value( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a3", public_url="gamma") + _sign_in(client, base_url="http://gamma.jottit.test/", site_id=site_id) + + response = client.get("/admin/change-site-address", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + assert 'value="gamma"' in response.data.decode() + + +# ---- POST: change ---- + + +def test_post_change_site_address_updates_and_redirects_to_new_subdomain( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a4", public_url="delta") + _sign_in(client, base_url="http://delta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://delta.jottit.test/", + data={"public_url": "deltatwo"}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "http://deltatwo.jottit.test/admin/settings" + + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.public_url == "deltatwo" + + +def test_post_change_site_address_clearing_redirects_to_secret_url( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a5", public_url="epsilon") + _sign_in(client, base_url="http://epsilon.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://epsilon.jottit.test/", + data={"public_url": ""}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "http://jottit.test/a5/admin/settings" + + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.public_url is None + + +def test_post_change_site_address_same_value_is_noop_redirect( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a6", public_url="zeta") + _sign_in(client, base_url="http://zeta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://zeta.jottit.test/", + data={"public_url": "zeta"}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "http://zeta.jottit.test/admin/settings" + + +def test_post_change_site_address_lowercases_and_trims( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a7", public_url="eta") + _sign_in(client, base_url="http://eta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://eta.jottit.test/", + data={"public_url": " ETA-TWO "}, + ) + + assert response.status_code == 303 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.public_url == "eta-two" + + +# ---- POST: rejection ---- + + +def test_post_change_site_address_rejects_invalid_chars( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a8", public_url="theta") + _sign_in(client, base_url="http://theta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://theta.jottit.test/", + data={"public_url": "has spaces"}, + ) + + assert response.status_code == 400 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.public_url == "theta" + + +def test_post_change_site_address_rejects_taken_slug( + client: FlaskClient, db_engine: Engine +) -> None: + # Pre-seed another site that already owns "claimed". + with db_engine.begin() as conn: + new_site(conn, content="x", secret_url="other", public_url="claimed") + site_id = _seed_claimed_site(db_engine, secret_url="a9", public_url="iota") + _sign_in(client, base_url="http://iota.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://iota.jottit.test/", + data={"public_url": "claimed"}, + ) + + assert response.status_code == 400 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.public_url == "iota" + + +def test_post_change_site_address_rejects_reserved_slug( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="a10", public_url="kappa") + _sign_in(client, base_url="http://kappa.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/change-site-address", + base_url="http://kappa.jottit.test/", + data={"public_url": "www"}, + ) + + assert response.status_code == 400 + + +# ---- /admin/url-available ---- + + +def test_url_available_true_for_free_slug(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="ua1", public_url="lambda") + _sign_in(client, base_url="http://lambda.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/url-available", + base_url="http://lambda.jottit.test/", + data={"url": "fresh-slug"}, + ) + + assert response.status_code == 200 + assert json.loads(response.data) == {"available": True} + + +def test_url_available_false_for_taken_slug(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="ua2", public_url="mu") + with db_engine.begin() as conn: + new_site(conn, content="x", secret_url="ua2other", public_url="grabbed") + _sign_in(client, base_url="http://mu.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/url-available", + base_url="http://mu.jottit.test/", + data={"url": "grabbed"}, + ) + + assert json.loads(response.data) == {"available": False} + + +def test_url_available_false_for_reserved(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="ua3", public_url="nu") + _sign_in(client, base_url="http://nu.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/url-available", + base_url="http://nu.jottit.test/", + data={"url": "www"}, + ) + + assert json.loads(response.data) == {"available": False} + + +def test_url_available_false_for_invalid_chars(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="ua4", public_url="xi") + _sign_in(client, base_url="http://xi.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/url-available", + base_url="http://xi.jottit.test/", + data={"url": "has spaces"}, + ) + + assert json.loads(response.data) == {"available": False} + + +def test_url_available_false_for_empty(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="ua5", public_url="omicron") + _sign_in(client, base_url="http://omicron.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/url-available", + base_url="http://omicron.jottit.test/", + data={"url": ""}, + ) + + assert json.loads(response.data) == {"available": False} + + +def test_url_available_requires_auth(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="ua6", public_url="pi") + + response = client.post( + "/admin/url-available", + base_url="http://pi.jottit.test/", + data={"url": "anything"}, + ) + + assert response.status_code == 401 + + +# ---- Secret-URL routing ---- + + +def test_change_site_address_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="abc12") + _sign_in(client, base_url=APEX, site_id=site_id) + + response = client.post( + "/abc12/admin/change-site-address", + base_url=APEX, + data={"public_url": "newpub"}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "http://newpub.jottit.test/admin/settings" diff --git a/tests/test_admin_delete_and_export.py b/tests/test_admin_delete_and_export.py new file mode 100644 index 0000000..f8c5f94 --- /dev/null +++ b/tests/test_admin_delete_and_export.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import io +import zipfile +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_site, + is_public_url_available, + 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_claimed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + content: str = "home page", +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content=content, secret_url=secret_url, public_url=public_url) + claim_site( + conn, + site_id=site_id, + password_hash=hash_password("hunter2"), + email="owner@example.com", + security="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] + + +# ---- /admin/delete: auth ---- + + +def test_delete_get_redirects_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="dl1", public_url="alpha") + + response = client.get("/admin/delete", base_url="http://alpha.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_delete_post_returns_401_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="dl2", public_url="beta") + + response = client.post("/admin/delete", base_url="http://beta.jottit.test/") + + assert response.status_code == 401 + + +# ---- /admin/delete: GET ---- + + +def test_delete_get_renders_confirmation(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="dl3", public_url="gamma") + _sign_in(client, base_url="http://gamma.jottit.test/", site_id=site_id) + + response = client.get("/admin/delete", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + assert "Delete this site" in body + + +# ---- /admin/delete: POST ---- + + +def test_delete_post_marks_site_deleted_and_frees_slug( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="dl4", public_url="delta") + _sign_in(client, base_url="http://delta.jottit.test/", site_id=site_id) + + response = client.post("/admin/delete", base_url="http://delta.jottit.test/") + + assert response.status_code == 303 + assert response.headers["Location"] == "http://jottit.test/" + + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.deleted is True + assert row.public_url is None + assert is_public_url_available(conn, public_url="delta") is True + + +def test_delete_post_signs_user_out_of_that_site(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="dl5", public_url="epsilon") + _sign_in(client, base_url="http://epsilon.jottit.test/", site_id=site_id) + + client.post("/admin/delete", base_url="http://epsilon.jottit.test/") + + with client.session_transaction(base_url="http://epsilon.jottit.test/") as sess: + assert site_id not in sess.get("signed_in_sites", []) + + +def test_deleted_site_no_longer_resolves_for_subsequent_requests( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="dl6", public_url="zeta") + _sign_in(client, base_url="http://zeta.jottit.test/", site_id=site_id) + + client.post("/admin/delete", base_url="http://zeta.jottit.test/") + + # The site row still exists, but the resolver hides it now. + response = client.get("/", base_url="http://zeta.jottit.test/") + assert response.status_code == 404 + + +def test_deleted_site_secret_url_also_404s(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="dl7") + _sign_in(client, base_url=APEX, site_id=site_id) + + client.post("/dl7/admin/delete", base_url=APEX) + + response = client.get("/dl7/", base_url=APEX) + assert response.status_code == 404 + + +# ---- /admin/export ---- + + +def test_export_redirects_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="ex1", public_url="eta") + + response = client.get("/admin/export", base_url="http://eta.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_export_returns_zip_with_pages(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="ex2", public_url="theta", content="welcome") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="notes", content="my notes") + new_page(conn, site_id=site_id, name="about", content="about me") + _sign_in(client, base_url="http://theta.jottit.test/", site_id=site_id) + + response = client.get("/admin/export", base_url="http://theta.jottit.test/") + + assert response.status_code == 200 + assert response.mimetype == "application/zip" + assert "theta-export.zip" in response.headers["Content-Disposition"] + + with zipfile.ZipFile(io.BytesIO(response.data)) as zf: + names = sorted(zf.namelist()) + assert names == ["about.md", "home.md", "notes.md"] + assert zf.read("home.md").decode() == "welcome" + assert zf.read("notes.md").decode() == "my notes" + assert zf.read("about.md").decode() == "about me" + + +def test_export_uses_latest_revision_for_each_page(client: FlaskClient, db_engine: Engine) -> None: + from jottit.db import get_page + + site_id = _seed_claimed_site(db_engine, secret_url="ex3", public_url="iota", content="v1") + with db_engine.begin() as conn: + home = get_page(conn, site_id=site_id, page_name="") + assert home is not None + update_page(conn, page_id=home.id, content="v2") + _sign_in(client, base_url="http://iota.jottit.test/", site_id=site_id) + + response = client.get("/admin/export", base_url="http://iota.jottit.test/") + + with zipfile.ZipFile(io.BytesIO(response.data)) as zf: + assert zf.read("home.md").decode() == "v2" + + +def test_export_filename_falls_back_to_secret_url_when_no_public_url( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="abc12") + _sign_in(client, base_url=APEX, site_id=site_id) + + response = client.get("/abc12/admin/export", base_url=APEX) + + assert "abc12-export.zip" in response.headers["Content-Disposition"] + + +def test_export_skips_deleted_pages(client: FlaskClient, db_engine: Engine) -> None: + from jottit.db import delete_page, get_page + + site_id = _seed_claimed_site(db_engine, secret_url="ex4", public_url="kappa") + 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) + _sign_in(client, base_url="http://kappa.jottit.test/", site_id=site_id) + + response = client.get("/admin/export", base_url="http://kappa.jottit.test/") + + with zipfile.ZipFile(io.BytesIO(response.data)) as zf: + assert "gone.md" not in zf.namelist() + + +def test_export_export_filename_strips_slashes(client: FlaskClient, db_engine: Engine) -> None: + """Path separators in page names mustn't survive into the zip — that's the + actual zip-slip vector. Leading dots are cosmetic and stay as-is.""" + site_id = _seed_claimed_site(db_engine, secret_url="ex5", public_url="lambda") + with db_engine.begin() as conn: + new_page(conn, site_id=site_id, name="../oops", content="x") + _sign_in(client, base_url="http://lambda.jottit.test/", site_id=site_id) + + response = client.get("/admin/export", base_url="http://lambda.jottit.test/") + + with zipfile.ZipFile(io.BytesIO(response.data)) as zf: + for name in zf.namelist(): + assert "/" not in name + assert "\\" not in name diff --git a/tests/test_admin_design.py b/tests/test_admin_design.py new file mode 100644 index 0000000..24d9fce --- /dev/null +++ b/tests/test_admin_design.py @@ -0,0 +1,237 @@ +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_design, metadata, new_site + +APEX = "http://jottit.test/" + +GOOD_DESIGN = { + "title_font": "Georgia", + "subtitle_font": "Georgia", + "headings_font": "Georgia", + "content_font": "Verdana", + "header_color": "#003452", + "title_color": "#ffffff", + "subtitle_color": "#bfe8ff", + "title_size": "120", + "subtitle_size": "110", + "headings_size": "130", + "content_size": "100", + "hue": "143", + "brightness": "214", +} + + +@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_claimed_site(db_engine: Engine, *, secret_url: str, public_url: 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) + claim_site( + conn, + site_id=site_id, + password_hash=hash_password("hunter2"), + email="owner@example.com", + security="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] + + +# ---- Auth gating ---- + + +def test_get_design_redirects_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="d1", public_url="alpha") + + response = client.get("/admin/design", base_url="http://alpha.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_post_design_returns_401_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="d2", public_url="beta") + + response = client.post("/admin/design", base_url="http://beta.jottit.test/", data=GOOD_DESIGN) + + assert response.status_code == 401 + + +# ---- GET ---- + + +def test_get_renders_form_with_current_design(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d3", public_url="gamma") + _sign_in(client, base_url="http://gamma.jottit.test/", site_id=site_id) + + response = client.get("/admin/design", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + # new_site populates the design row from a random ColorScheme; the + # title font default is Lucida_Grande. + assert "Lucida_Grande" in body + + +# ---- POST: happy path ---- + + +def test_post_design_saves_all_thirteen_fields(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d4", public_url="delta") + _sign_in(client, base_url="http://delta.jottit.test/", site_id=site_id) + + response = client.post("/admin/design", base_url="http://delta.jottit.test/", data=GOOD_DESIGN) + + assert response.status_code == 303 + assert response.headers["Location"] == "/admin/design" + + with db_engine.connect() as conn: + d = get_design(conn, site_id=site_id) + assert d is not None + assert d.title_font == "Georgia" + assert d.content_font == "Verdana" + assert d.header_color == "#003452" + assert d.title_color == "#ffffff" + assert d.subtitle_color == "#bfe8ff" + assert d.title_size == 120 + assert d.subtitle_size == 110 + assert d.headings_size == 130 + assert d.content_size == 100 + assert d.hue == "143" + assert d.brightness == "214" + + +def test_post_design_accepts_short_hex_colors(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d5", public_url="epsilon") + _sign_in(client, base_url="http://epsilon.jottit.test/", site_id=site_id) + data = {**GOOD_DESIGN, "header_color": "#fff", "title_color": "#000"} + + response = client.post("/admin/design", base_url="http://epsilon.jottit.test/", data=data) + + assert response.status_code == 303 + with db_engine.connect() as conn: + d = get_design(conn, site_id=site_id) + assert d is not None + assert d.header_color == "#fff" + + +def test_post_design_empty_string_leaves_existing_value( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d6", public_url="zeta") + _sign_in(client, base_url="http://zeta.jottit.test/", site_id=site_id) + with db_engine.connect() as conn: + before = get_design(conn, site_id=site_id) + assert before is not None + original_subtitle_font = before.subtitle_font + + data = {**GOOD_DESIGN, "subtitle_font": ""} + client.post("/admin/design", base_url="http://zeta.jottit.test/", data=data) + + with db_engine.connect() as conn: + after = get_design(conn, site_id=site_id) + assert after is not None + assert after.subtitle_font == original_subtitle_font + + +# ---- POST: validation ---- + + +def test_post_design_rejects_invalid_hex_color(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d7", public_url="eta") + _sign_in(client, base_url="http://eta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/design", + base_url="http://eta.jottit.test/", + data={**GOOD_DESIGN, "header_color": "red; evil"}, + ) + + assert response.status_code == 400 + with db_engine.connect() as conn: + d = get_design(conn, site_id=site_id) + assert d is not None + # Color unchanged from the new_site default. + assert d.header_color != "red; evil" + + +def test_post_design_rejects_size_out_of_range(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d8", public_url="theta") + _sign_in(client, base_url="http://theta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/design", + base_url="http://theta.jottit.test/", + data={**GOOD_DESIGN, "title_size": "99999"}, + ) + + assert response.status_code == 400 + + +def test_post_design_rejects_non_numeric_size(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d9", public_url="iota") + _sign_in(client, base_url="http://iota.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/design", + base_url="http://iota.jottit.test/", + data={**GOOD_DESIGN, "content_size": "huge"}, + ) + + assert response.status_code == 400 + + +def test_post_design_rejects_font_with_html_chars(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d10", public_url="kappa") + _sign_in(client, base_url="http://kappa.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/design", + base_url="http://kappa.jottit.test/", + data={**GOOD_DESIGN, "title_font": ""}, + ) + + assert response.status_code == 400 + + +def test_post_design_rejects_hue_out_of_range(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="d11", public_url="lambda") + _sign_in(client, base_url="http://lambda.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/design", + base_url="http://lambda.jottit.test/", + data={**GOOD_DESIGN, "hue": "9999"}, + ) + + assert response.status_code == 400 + + +# ---- Secret-URL routing ---- + + +def test_design_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="abc12") + _sign_in(client, base_url=APEX, site_id=site_id) + + response = client.post("/abc12/admin/design", base_url=APEX, data=GOOD_DESIGN) + + assert response.status_code == 303 + assert response.headers["Location"] == "/abc12/admin/design" diff --git a/tests/test_admin_settings.py b/tests/test_admin_settings.py new file mode 100644 index 0000000..c43e051 --- /dev/null +++ b/tests/test_admin_settings.py @@ -0,0 +1,213 @@ +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_site, metadata, new_site, update_site + +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_claimed_site( + db_engine: Engine, + *, + secret_url: str, + public_url: str | None = None, + security: str = "private", +) -> int: + with db_engine.begin() as conn: + site_id = new_site(conn, content="hi", secret_url=secret_url, public_url=public_url) + claim_site( + conn, + site_id=site_id, + password_hash=hash_password("hunter2"), + email="owner@example.com", + security=security, + ) + 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] + + +# ---- Auth gating ---- + + +def test_get_settings_redirects_when_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="s1", public_url="alpha") + + response = client.get("/admin/settings", base_url="http://alpha.jottit.test/") + + assert response.status_code == 303 + assert "site/signin" in response.headers["Location"] + + +def test_post_settings_returns_401_when_anonymous(client: FlaskClient, db_engine: Engine) -> None: + _seed_claimed_site(db_engine, secret_url="s2", public_url="beta") + + response = client.post( + "/admin/settings", + base_url="http://beta.jottit.test/", + data={"title": "X", "subtitle": "Y", "email": "o@example.com", "security": "private"}, + ) + + assert response.status_code == 401 + + +# ---- GET ---- + + +def test_get_settings_renders_form_with_current_values( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="s3", public_url="gamma") + with db_engine.begin() as conn: + update_site(conn, site_id=site_id, title="My Site", subtitle="Of jottings", email="o@x.com") + _sign_in(client, base_url="http://gamma.jottit.test/", site_id=site_id) + + response = client.get("/admin/settings", base_url="http://gamma.jottit.test/") + + assert response.status_code == 200 + body = response.data.decode() + assert 'value="My Site"' in body + assert 'value="Of jottings"' in body + assert 'value="o@x.com"' in body + + +# ---- POST ---- + + +def test_post_settings_updates_fields_and_redirects(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="s4", public_url="delta") + _sign_in(client, base_url="http://delta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/settings", + base_url="http://delta.jottit.test/", + data={ + "title": "New Title", + "subtitle": "New Subtitle", + "email": "new@example.com", + "security": "public", + }, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "/" + + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.title == "New Title" + assert row.subtitle == "New Subtitle" + assert row.email == "new@example.com" + assert row.security == "public" + + +def test_post_settings_rejects_invalid_email(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="s5", public_url="epsilon") + _sign_in(client, base_url="http://epsilon.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/settings", + base_url="http://epsilon.jottit.test/", + data={ + "title": "x", + "subtitle": "y", + "email": "not-an-email", + "security": "private", + }, + ) + + assert response.status_code == 400 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + # Original values untouched. + assert row.email == "owner@example.com" + + +def test_post_settings_rejects_invalid_security_level( + client: FlaskClient, db_engine: Engine +) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="s6", public_url="zeta") + _sign_in(client, base_url="http://zeta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/settings", + base_url="http://zeta.jottit.test/", + data={"title": "x", "subtitle": "y", "email": "", "security": "superadmin"}, + ) + + assert response.status_code == 400 + + +def test_post_settings_allows_empty_email(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="s7", public_url="eta") + _sign_in(client, base_url="http://eta.jottit.test/", site_id=site_id) + + response = client.post( + "/admin/settings", + base_url="http://eta.jottit.test/", + data={"title": "x", "subtitle": "y", "email": "", "security": "private"}, + ) + + assert response.status_code == 303 + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.email == "" + + +def test_post_settings_preserves_unrelated_fields(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="s8", public_url="theta") + _sign_in(client, base_url="http://theta.jottit.test/", site_id=site_id) + + client.post( + "/admin/settings", + base_url="http://theta.jottit.test/", + data={"title": "T", "subtitle": "S", "email": "n@example.com", "security": "public"}, + ) + + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + # public_url is managed by /admin/change-site-address, not here. + assert row.public_url == "theta" + # Password is untouched. + assert row.password is not None + + +# ---- Secret-URL routing ---- + + +def test_admin_settings_via_secret_url(client: FlaskClient, db_engine: Engine) -> None: + site_id = _seed_claimed_site(db_engine, secret_url="abc12") + _sign_in(client, base_url=APEX, site_id=site_id) + + response = client.post( + "/abc12/admin/settings", + base_url=APEX, + data={"title": "T", "subtitle": "S", "email": "o@example.com", "security": "open"}, + ) + + assert response.status_code == 303 + assert response.headers["Location"] == "/abc12/" + with db_engine.connect() as conn: + row = get_site(conn, site_id=site_id) + assert row is not None + assert row.security == "open" diff --git a/tests/test_db_queries.py b/tests/test_db_queries.py index b9862ac..8a197fc 100644 --- a/tests/test_db_queries.py +++ b/tests/test_db_queries.py @@ -5,14 +5,20 @@ from jottit.db import ( _AMBIGUOUS_CHARS, + RESERVED_PUBLIC_URLS, + change_public_url, claim_site, delete_draft, delete_page, + delete_site, designs, drafts, + get_design, get_page, + get_pages_for_export, get_revision, get_site, + is_public_url_available, new_page, new_site, pages, @@ -22,6 +28,7 @@ set_password, undelete_page, update_caret_pos, + update_design, update_page, update_site, ) @@ -472,3 +479,179 @@ def test_update_site_can_change_security_level(db_conn: Connection) -> None: row = get_site(db_conn, site_id=site_id) assert row is not None assert row.security == "open" + + +# ---- change_public_url ---- + + +def test_change_public_url_sets_value(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="cp1") + + change_public_url(db_conn, site_id=site_id, public_url="myblog") + + row = get_site(db_conn, site_id=site_id) + assert row is not None + assert row.public_url == "myblog" + + +def test_change_public_url_empty_string_clears_value(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="cp2", public_url="orig") + + change_public_url(db_conn, site_id=site_id, public_url="") + + row = get_site(db_conn, site_id=site_id) + assert row is not None + assert row.public_url is None + + +def test_change_public_url_none_clears_value(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="cp3", public_url="orig") + + change_public_url(db_conn, site_id=site_id, public_url=None) + + row = get_site(db_conn, site_id=site_id) + assert row is not None + assert row.public_url is None + + +# ---- is_public_url_available ---- + + +def test_is_public_url_available_true_for_unused_slug(db_conn: Connection) -> None: + assert is_public_url_available(db_conn, public_url="brand-new") is True + + +def test_is_public_url_available_false_when_taken(db_conn: Connection) -> None: + new_site(db_conn, content="hi", secret_url="av1", public_url="takenslug") + + assert is_public_url_available(db_conn, public_url="takenslug") is False + + +def test_is_public_url_available_false_for_reserved(db_conn: Connection) -> None: + for reserved in RESERVED_PUBLIC_URLS: + assert is_public_url_available(db_conn, public_url=reserved) is False + + +# ---- delete_site ---- + + +def test_delete_site_marks_deleted_and_frees_public_url(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="ds1", public_url="goinggone") + + delete_site(db_conn, site_id=site_id) + + row = get_site(db_conn, site_id=site_id) + assert row is not None + assert row.deleted is True + assert row.public_url is None + # The slug should now be available for someone else. + assert is_public_url_available(db_conn, public_url="goinggone") is True + + +def test_delete_site_does_not_drop_pages_or_revisions(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="surviving content", secret_url="ds2") + + delete_site(db_conn, site_id=site_id) + + # Pages and revisions still queryable for a potential restore path. + page = get_page(db_conn, site_id=site_id, page_name="") + assert page is not None + rev = get_revision(db_conn, page_id=page.id) + assert rev is not None + assert rev.content == "surviving content" + + +# ---- get_design / update_design ---- + + +def test_get_design_returns_row_created_by_new_site(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="gd1") + + design = get_design(db_conn, site_id=site_id) + + assert design is not None + assert design.site_id == site_id + assert design.title_font == "Lucida_Grande" + + +def test_update_design_patches_provided_fields(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="ud1") + original = get_design(db_conn, site_id=site_id) + assert original is not None + original_subtitle_font = original.subtitle_font + + update_design( + db_conn, + site_id=site_id, + title_font="Georgia", + header_color="#abcdef", + title_size=140, + ) + + updated = get_design(db_conn, site_id=site_id) + assert updated is not None + assert updated.title_font == "Georgia" + assert updated.header_color == "#abcdef" + assert updated.title_size == 140 + # Unprovided field untouched. + assert updated.subtitle_font == original_subtitle_font + + +def test_update_design_ignores_unknown_fields(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="ud2") + + update_design(db_conn, site_id=site_id, evil_column="payload", title_font="Helvetica") + + design = get_design(db_conn, site_id=site_id) + assert design is not None + assert design.title_font == "Helvetica" + + +def test_update_design_no_provided_fields_is_noop(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="hi", secret_url="ud3") + + update_design(db_conn, site_id=site_id) # should not raise + + +# ---- get_pages_for_export ---- + + +def test_get_pages_for_export_returns_latest_revisions(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="home v1", secret_url="ex1") + home = get_page(db_conn, site_id=site_id, page_name="") + assert home is not None + # New revision on home; export should show v2. + update_page(db_conn, page_id=home.id, content="home v2") + + new_page(db_conn, site_id=site_id, name="notes", content="notes body") + + rows = get_pages_for_export(db_conn, site_id=site_id) + + by_name = {r.page_name: r for r in rows} + assert by_name[""].content == "home v2" + assert by_name["notes"].content == "notes body" + + +def test_get_pages_for_export_skips_deleted_pages(db_conn: Connection) -> None: + site_id = new_site(db_conn, content="home", secret_url="ex2") + new_page(db_conn, site_id=site_id, name="gone", content="bye") + page = get_page(db_conn, site_id=site_id, page_name="gone") + assert page is not None + delete_page(db_conn, page_id=page.id) + + rows = get_pages_for_export(db_conn, site_id=site_id) + + names = [r.page_name for r in rows] + assert "gone" not in names + assert "" in names + + +def test_get_pages_for_export_scoped_to_site(db_conn: Connection) -> None: + site_a = new_site(db_conn, content="a-home", secret_url="ex3a") + site_b = new_site(db_conn, content="b-home", secret_url="ex3b") + new_page(db_conn, site_id=site_b, name="b-page", content="b-only") + + rows = get_pages_for_export(db_conn, site_id=site_a) + + names = [r.page_name for r in rows] + assert "b-page" not in names diff --git a/tests/test_secret_routing.py b/tests/test_secret_routing.py index 0ab4f3a..c4f4bd0 100644 --- a/tests/test_secret_routing.py +++ b/tests/test_secret_routing.py @@ -28,52 +28,10 @@ def test_secret_site_routes( assert expected_substring in body -@pytest.mark.parametrize( - ("method", "path", "expected_substring"), - [ - ("GET", "/abc123/admin/settings", "admin/settings GET"), - ("POST", "/abc123/admin/settings", "admin/settings POST"), - ("GET", "/abc123/admin/design", "admin/design GET"), - ("POST", "/abc123/admin/design", "admin/design POST"), - ("POST", "/abc123/admin/url-available", "admin/url-available POST"), - ("GET", "/abc123/admin/delete", "admin/delete GET"), - ("POST", "/abc123/admin/delete", "admin/delete POST"), - ("GET", "/abc123/admin/change-site-address", "admin/change-site-address GET"), - ("POST", "/abc123/admin/change-site-address", "admin/change-site-address POST"), - ("GET", "/abc123/admin/change-password", "admin/change-password GET"), - ("POST", "/abc123/admin/change-password", "admin/change-password POST"), - ("GET", "/abc123/admin/export", "admin/export GET"), - ], -) -def test_secret_admin_routes( - client: FlaskClient, method: str, path: str, expected_substring: str -) -> None: - response = client.open(path, method=method, base_url=APEX) - assert response.status_code == 200 - body = response.data.decode() - assert body.startswith("admin:abc123 ") - assert expected_substring in body - - -# Page routes via the secret blueprint are exercised end-to-end in -# tests/test_page_view.py. The site/admin parity checks below exercise that -# the secret and subdomain blueprints share the same handler functions. - - -@pytest.mark.parametrize( - ("secret_path", "subdomain_path"), - [ - ("/abc123/admin/settings", "/admin/settings"), - ("/abc123/admin/design", "/admin/design"), - ], -) -def test_secret_and_subdomain_share_handler( - client: FlaskClient, secret_path: str, subdomain_path: str -) -> None: - secret = client.get(secret_path, base_url=APEX) - subdomain = client.get(subdomain_path, base_url="http://abc123.jottit.test/") - assert secret.status_code == subdomain.status_code == 200 - assert secret.data == subdomain.data +# /admin/* routes via the secret blueprint are exercised end-to-end in +# tests/test_admin_*.py. Page routes are in tests/test_page_view.py. +# Handler-parity between subdomain and secret blueprints is implicit in +# those files: each admin test has a "via secret URL" case. def test_apex_static_routes_still_win_over_secret_prefix(client: FlaskClient) -> None: diff --git a/tests/test_subdomain_routing.py b/tests/test_subdomain_routing.py index 571d09e..da88ccb 100644 --- a/tests/test_subdomain_routing.py +++ b/tests/test_subdomain_routing.py @@ -28,33 +28,7 @@ def test_site_blueprint( assert expected_substring in body -@pytest.mark.parametrize( - ("method", "path", "expected_substring"), - [ - ("GET", "/admin/settings", "admin/settings GET"), - ("POST", "/admin/settings", "admin/settings POST"), - ("GET", "/admin/design", "admin/design GET"), - ("POST", "/admin/design", "admin/design POST"), - ("POST", "/admin/url-available", "admin/url-available POST"), - ("GET", "/admin/delete", "admin/delete GET"), - ("POST", "/admin/delete", "admin/delete POST"), - ("GET", "/admin/change-site-address", "admin/change-site-address GET"), - ("POST", "/admin/change-site-address", "admin/change-site-address POST"), - ("GET", "/admin/change-password", "admin/change-password GET"), - ("POST", "/admin/change-password", "admin/change-password POST"), - ("GET", "/admin/export", "admin/export GET"), - ], -) -def test_admin_blueprint( - client: FlaskClient, method: str, path: str, expected_substring: str -) -> None: - response = client.open(path, method=method, base_url=SITE_BASE) - assert response.status_code == 200 - body = response.data.decode() - assert body.startswith("admin:mysite ") - assert expected_substring in body - - +# /admin/* routes are exercised end-to-end in tests/test_admin_*.py. # Page routes are exercised end-to-end in tests/test_page_view.py.