ackandClaude Fable 5.1
0d01e23d7c
From the development repository at 299f73d, 2026-10-11: hosting from the site documented; a page loaded from a notebook and left unchanged is not kept again; notebooks and pages are deleted from the chooser, to the system's Trash (--trash); the Notebooks button. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
370 lines
21 KiB
Python
370 lines
21 KiB
Python
"""Tests of the local program's notebook requests. Run from the project
|
|
directory:
|
|
|
|
.venv/bin/python -m unittest tests.test_local_program -v
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
|
|
import kokkoja # noqa: E402
|
|
|
|
TOKEN = "test-token"
|
|
PAGE = b'<svg xmlns="http://www.w3.org/2000/svg" data-kokkoja="1"/>\n'
|
|
PNG = b"\x89PNG\r\n\x1a\n" + b"the rest of a picture"
|
|
TEXT_ONLY_LAYER = (b'<g data-layer="abc" data-name="Andy"><g data-text="t1" data-x="10" data-y="10">'
|
|
b'<desc>hello</desc><text><tspan>hello</tspan></text></g></g>')
|
|
|
|
WRITTEN_PAGE = b'<svg data-kokkoja="1"><g data-layer="a"><g data-stroke="s"/></g></svg>\n'
|
|
|
|
|
|
def page_named(session, page, number, body=b'<g data-layer="a"><g data-stroke="s"/></g>'):
|
|
"""A page file as the client writes it: named on the root element and
|
|
in the title, then the paper and the content."""
|
|
return (b'<?xml version="1.0" encoding="UTF-8"?>\n'
|
|
b'<svg xmlns="http://www.w3.org/2000/svg" data-kokkoja="1" data-session="' + session +
|
|
b'" data-host="h" data-page="' + page + b'" data-number="' + number +
|
|
b'" data-base="2970 2100" viewBox="0 0 2970 2100">\n'
|
|
b'<title>Kokkoja page ' + number + b'</title>\n'
|
|
b'<rect x="0" y="0" width="2970" height="2100" fill="#ffffff"/>\n' + body + b'</svg>\n')
|
|
|
|
|
|
class LocalProgramTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.notebooks = pathlib.Path(self.directory.name) / "notebooks"
|
|
self.config = pathlib.Path(self.directory.name) / "config"
|
|
# Deleted things go to a Trash of the test's own, a plain
|
|
# directory as macOS's Trash is; the freedesktop layout has its
|
|
# own test.
|
|
self.trash = pathlib.Path(self.directory.name) / "trash"
|
|
self.server = kokkoja.make_server(0, "ws://relay.example:8770/", self.notebooks, TOKEN,
|
|
self.config, trash=kokkoja.Trash(self.trash))
|
|
self.port = self.server.server_address[1]
|
|
thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
thread.start()
|
|
self.addCleanup(thread.join)
|
|
self.addCleanup(self.server.server_close)
|
|
self.addCleanup(self.server.shutdown)
|
|
|
|
def request(self, method, path, body=None, token=TOKEN, headers=None):
|
|
"""Returns (status, body)."""
|
|
all_headers = {} if token is None else {"Authorization": f"Bearer {token}"}
|
|
all_headers.update(headers or {})
|
|
request = urllib.request.Request(f"http://127.0.0.1:{self.port}{path}", data=body,
|
|
method=method, headers=all_headers)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
return response.status, response.read()
|
|
except urllib.error.HTTPError as error:
|
|
with error:
|
|
return error.code, error.read()
|
|
|
|
def json(self, method, path, **arguments):
|
|
status, body = self.request(method, path, **arguments)
|
|
return status, json.loads(body)
|
|
|
|
def test_client_and_config_need_no_token(self):
|
|
status, body = self.request("GET", "/", token=None)
|
|
self.assertEqual(status, 200)
|
|
self.assertIn(b"Kokkoja", body)
|
|
self.assertEqual(self.json("GET", "/config.json", token=None),
|
|
(200, {"relay": "ws://relay.example:8770", "storage": "files", "site": ""}))
|
|
|
|
def test_the_relay_is_remembered(self):
|
|
config = pathlib.Path(self.directory.name) / "relay_config"
|
|
self.assertEqual(kokkoja.remembered_relay(config), (None, False))
|
|
self.assertEqual(kokkoja.remembered_relay(config, "wss://kokkoja.example/relay"),
|
|
("wss://kokkoja.example/relay", False))
|
|
self.assertEqual(kokkoja.remembered_relay(config), ("wss://kokkoja.example/relay", True))
|
|
self.assertEqual(kokkoja.remembered_relay(config, " ws://127.0.0.1:8770 "),
|
|
("ws://127.0.0.1:8770", False))
|
|
self.assertEqual((config / "relay").read_text(), "ws://127.0.0.1:8770\n")
|
|
|
|
def test_site_comes_from_a_kokkoja_server_relay_or_is_given(self):
|
|
self.assertEqual(kokkoja.site_of("wss://kokkoja.example/relay", ""), "https://kokkoja.example/")
|
|
self.assertEqual(kokkoja.site_of("wss://kokkoja.example/relay/", ""), "https://kokkoja.example/")
|
|
self.assertEqual(kokkoja.site_of("ws://127.0.0.1:8770", ""), "")
|
|
self.assertEqual(kokkoja.site_of("wss://relay.example:8770", ""), "")
|
|
self.assertEqual(kokkoja.site_of("ws://127.0.0.1:8770", "http://127.0.0.1:9000"),
|
|
"http://127.0.0.1:9000/")
|
|
|
|
def test_write_list_and_read_a_page(self):
|
|
self.assertEqual(self.json("GET", "/api/notebooks")[1]["notebooks"], [])
|
|
page = "/api/notebooks/" + urllib.parse.quote("Übungen A1") + "/2026_10_02_page.svg"
|
|
self.assertEqual(self.json("PUT", page, body=PAGE), (200, {"saved": "2026_10_02_page.svg"}))
|
|
self.assertEqual(self.request("GET", page), (200, PAGE))
|
|
self.assertEqual((self.notebooks / "Übungen A1" / "2026_10_02_page.svg").read_bytes(), PAGE)
|
|
self.assertEqual(self.json("GET", "/api/notebooks")[1]["notebooks"], ["Übungen A1"])
|
|
self.assertEqual(self.json("GET", "/api/notebooks/" + urllib.parse.quote("Übungen A1")),
|
|
(200, {"pages": ["2026_10_02_page.svg"]}))
|
|
|
|
def test_writing_again_replaces_the_page_and_leaves_nothing_else(self):
|
|
page = "/api/notebooks/Andy/page.svg"
|
|
self.request("PUT", page, body=b"first")
|
|
self.request("PUT", page, body=PAGE)
|
|
self.assertEqual(self.request("GET", page), (200, PAGE))
|
|
self.assertEqual([entry.name for entry in (self.notebooks / "Andy").iterdir()], ["page.svg"])
|
|
|
|
def test_token_is_required(self):
|
|
for token in (None, "wrong", ""):
|
|
with self.subTest(token=token):
|
|
self.assertEqual(self.request("GET", "/api/notebooks", token=token)[0], 401)
|
|
self.assertEqual(
|
|
self.request("PUT", "/api/notebooks/Andy/page.svg", body=PAGE, token=token)[0], 401)
|
|
self.assertFalse(self.notebooks.exists())
|
|
|
|
def test_request_from_another_web_site_is_refused(self):
|
|
for headers in ({"Origin": "https://evil.example"}, {"Host": "evil.example:8760"}):
|
|
with self.subTest(headers=headers):
|
|
status, _ = self.request("PUT", "/api/notebooks/Andy/page.svg", body=PAGE,
|
|
headers=headers)
|
|
self.assertEqual(status, 403)
|
|
status, _ = self.request("PUT", "/api/notebooks/Andy/page.svg", body=PAGE,
|
|
headers={"Origin": f"http://localhost:{self.port}"})
|
|
self.assertEqual(status, 200)
|
|
|
|
def test_names_cannot_leave_the_notebooks_directory(self):
|
|
outside = pathlib.Path(self.directory.name) / "outside.svg"
|
|
for path in ("/api/notebooks/../outside.svg",
|
|
"/api/notebooks/%2E%2E/outside.svg",
|
|
"/api/notebooks/Andy/..%2F..%2Foutside.svg",
|
|
"/api/notebooks/Andy%2F..%2F../outside.svg",
|
|
"/api/notebooks/.hidden/page.svg",
|
|
"/api/notebooks/Andy/.hidden.svg",
|
|
"/api/notebooks/Andy/sub/page.svg"):
|
|
with self.subTest(path=path):
|
|
status, _ = self.request("PUT", path, body=PAGE)
|
|
self.assertIn(status, (400, 404))
|
|
self.assertFalse(outside.exists())
|
|
self.assertFalse(self.notebooks.exists())
|
|
|
|
def test_only_pages_can_be_written(self):
|
|
self.assertEqual(self.request("PUT", "/api/notebooks/Andy/notes.txt", body=b"x")[0], 400)
|
|
self.assertEqual(self.request("PUT", "/api/notebooks/Andy", body=b"x")[0], 405)
|
|
|
|
def test_keeping_the_current_page(self):
|
|
current = "/api/notebooks/Andy/Current.svg"
|
|
keep = "/api/notebooks/Andy/keep"
|
|
today = time.strftime("%Y_%m_%d")
|
|
|
|
# Nothing to keep: no notebook, then a blank page.
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": None, "new": False}))
|
|
self.request("PUT", current, body=PAGE)
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": None, "new": False}))
|
|
|
|
# A page with writing is kept once, however often it is asked.
|
|
self.request("PUT", current, body=WRITTEN_PAGE)
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_01.svg", "new": True}))
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_01.svg", "new": False}))
|
|
self.assertEqual((self.notebooks / "Andy" / f"{today}_01.svg").read_bytes(), WRITTEN_PAGE)
|
|
|
|
# A changed page is kept under the next serial number.
|
|
changed = WRITTEN_PAGE + b"<!-- more -->"
|
|
self.request("PUT", current, body=changed)
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_02.svg", "new": True}))
|
|
self.assertEqual(self.json("GET", "/api/notebooks/Andy")[1]["pages"],
|
|
[f"{today}_01.svg", f"{today}_02.svg", "Current.svg"])
|
|
|
|
# A page with typed text and nothing else is not blank either.
|
|
self.request("PUT", current, body=b'<svg data-kokkoja="1">' + TEXT_ONLY_LAYER + b"</svg>\n")
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_03.svg", "new": True}))
|
|
|
|
# The same page under other names is the same page: loaded from
|
|
# the notebook into another session, it is a new page with a new
|
|
# id and number, and is not kept again when left unchanged. A
|
|
# different sheet, or typed text that happens to name a page, is
|
|
# not the same.
|
|
self.request("PUT", current, body=page_named(b"s1", b"p1", b"4"))
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_04.svg", "new": True}))
|
|
self.request("PUT", current, body=page_named(b"s2", b"p2", b"11"))
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_04.svg", "new": False}))
|
|
self.request("PUT", current, body=page_named(b"s2", b"p2", b"11").replace(b'viewBox="0 0', b'viewBox="-50 0'))
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_05.svg", "new": True}))
|
|
typed = page_named(b"s1", b"p1", b"4", body=b'<g data-layer="a"><g data-text="t"><desc>data-page="p1"</desc></g></g>')
|
|
self.request("PUT", current, body=typed)
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_06.svg", "new": True}))
|
|
self.request("PUT", current, body=typed.replace(b'<desc>data-page="p1"', b'<desc>data-page="p2"'))
|
|
self.assertEqual(self.json("POST", keep), (200, {"kept": f"{today}_07.svg", "new": True}))
|
|
|
|
def test_deleting_pages_and_notebooks(self):
|
|
notebook = self.notebooks / "Andy"
|
|
notebook.mkdir(parents=True)
|
|
for name in ("Current.svg", "2026_10_01_01.svg", "2026_10_01_02.svg"):
|
|
(notebook / name).write_bytes(WRITTEN_PAGE)
|
|
(self.notebooks / "Other").mkdir()
|
|
(self.notebooks / "Other" / "2026_10_02_01.svg").write_bytes(WRITTEN_PAGE)
|
|
self.assertTrue(self.json("GET", "/api/notebooks")[1]["trash"])
|
|
|
|
# A kept page goes to the Trash; the page being written stays.
|
|
self.assertEqual(self.json("DELETE", "/api/notebooks/Andy/2026_10_01_01.svg"),
|
|
(200, {"discarded": "2026_10_01_01.svg", "trash": True}))
|
|
self.assertEqual(self.json("GET", "/api/notebooks/Andy")[1]["pages"], ["2026_10_01_02.svg", "Current.svg"])
|
|
self.assertEqual((self.trash / "2026_10_01_01.svg").read_bytes(), WRITTEN_PAGE)
|
|
self.assertEqual(self.request("DELETE", "/api/notebooks/Andy/Current.svg")[0], 400)
|
|
self.assertEqual(self.request("DELETE", "/api/notebooks/Andy/none.svg")[0], 404)
|
|
self.assertEqual(self.request("DELETE", "/api/notebooks")[0], 405)
|
|
self.assertEqual(self.request("DELETE", "/api/notebooks/Andy/2026_10_01_02.svg", token=None)[0], 401)
|
|
|
|
# A whole notebook goes as one, and a name already taken in the
|
|
# Trash is not overwritten.
|
|
(self.notebooks / "Andy" / "2026_10_01_01.svg").write_bytes(b"kept again")
|
|
self.assertEqual(self.json("DELETE", "/api/notebooks/Andy"),
|
|
(200, {"discarded": "Andy", "trash": True}))
|
|
self.assertEqual(self.json("GET", "/api/notebooks")[1]["notebooks"], ["Other"])
|
|
self.assertEqual(sorted(p.name for p in (self.trash / "Andy").iterdir()),
|
|
["2026_10_01_01.svg", "2026_10_01_02.svg", "Current.svg"])
|
|
(self.notebooks / "Andy").mkdir()
|
|
(self.notebooks / "Andy" / "Current.svg").write_bytes(b"a second Andy")
|
|
self.assertEqual(self.json("DELETE", "/api/notebooks/Andy")[1]["discarded"], "Andy")
|
|
self.assertEqual((self.trash / "Andy.2" / "Current.svg").read_bytes(), b"a second Andy")
|
|
self.assertEqual((self.trash / "2026_10_01_01.svg").read_bytes(), WRITTEN_PAGE)
|
|
|
|
def test_the_freedesktop_trash(self):
|
|
basket = pathlib.Path(self.directory.name) / "Trash"
|
|
trash = kokkoja.Trash(basket, freedesktop=True)
|
|
page = self.notebooks / "Andy" / "2026_10_01_01.svg"
|
|
page.parent.mkdir(parents=True)
|
|
page.write_bytes(WRITTEN_PAGE)
|
|
trash.take(page)
|
|
self.assertFalse(page.exists())
|
|
self.assertEqual((basket / "files" / page.name).read_bytes(), WRITTEN_PAGE)
|
|
info = (basket / "info" / f"{page.name}.trashinfo").read_text()
|
|
self.assertTrue(info.startswith("[Trash Info]\nPath="), info)
|
|
self.assertIn(f"Path={urllib.parse.quote(str(page.resolve()))}\n", info)
|
|
self.assertRegex(info, r"DeletionDate=\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\n")
|
|
self.assertEqual(basket.stat().st_mode & 0o777, 0o700)
|
|
# The same name again: the info file reserves the first, so the
|
|
# second gets a serial.
|
|
page.write_bytes(b"another")
|
|
trash.take(page)
|
|
self.assertEqual((basket / "files" / "2026_10_01_01.2.svg").read_bytes(), b"another")
|
|
self.assertTrue((basket / "info" / "2026_10_01_01.2.svg.trashinfo").exists())
|
|
# Without a Trash, a deletion is for good.
|
|
page.write_bytes(b"gone")
|
|
self.assertFalse(kokkoja.discard(page, None))
|
|
self.assertFalse(page.exists())
|
|
self.assertTrue(kokkoja.discard(page.parent, kokkoja.Trash(basket)))
|
|
self.assertTrue((basket / "Andy").is_dir())
|
|
|
|
def test_a_kept_page_is_dated_when_it_was_last_changed(self):
|
|
notebook = self.notebooks / "Andy"
|
|
notebook.mkdir(parents=True)
|
|
(notebook / "Current.svg").write_bytes(WRITTEN_PAGE)
|
|
last_week = time.mktime((2026, 9, 25, 16, 30, 0, 0, 0, -1))
|
|
os.utime(notebook / "Current.svg", (last_week, last_week))
|
|
(notebook / "2026_09_25_01.svg").write_bytes(b"an earlier page of that day")
|
|
self.assertEqual(self.json("POST", "/api/notebooks/Andy/keep"),
|
|
(200, {"kept": "2026_09_25_02.svg", "new": True}))
|
|
self.assertEqual((notebook / "2026_09_25_02.svg").stat().st_mtime, last_week)
|
|
|
|
def test_identity_is_kept_once_and_privately(self):
|
|
self.assertEqual(self.request("GET", "/api/identity")[0], 404)
|
|
self.assertEqual(self.request("GET", "/api/identity", token=None)[0], 401)
|
|
identity = b'{"publicKey": "abc", "privateKey": {"kty": "EC"}}'
|
|
self.assertEqual(self.json("PUT", "/api/identity", body=identity), (200, {"kept": True}))
|
|
self.assertEqual(self.request("GET", "/api/identity"), (200, identity))
|
|
self.assertEqual(self.request("PUT", "/api/identity", body=b'{"other": 1}')[0], 409)
|
|
self.assertEqual(self.request("GET", "/api/identity"), (200, identity))
|
|
self.assertEqual(self.request("PUT", "/api/identity", body=b"not json")[0], 400)
|
|
path = self.config / "identity.json"
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_missing_page(self):
|
|
self.assertEqual(self.request("GET", "/api/notebooks/Andy/none.svg")[0], 404)
|
|
self.assertEqual(self.json("GET", "/api/notebooks/Andy"), (200, {"pages": []}))
|
|
|
|
# The file chooser and exporting.
|
|
|
|
def files(self, directory=None):
|
|
query = "" if directory is None else "?directory=" + urllib.parse.quote(str(directory))
|
|
return self.json("GET", "/api/files" + query)
|
|
|
|
def test_files_lists_directories_and_image_files(self):
|
|
home = pathlib.Path(self.directory.name) / "Bilder"
|
|
(home / "Übungen").mkdir(parents=True)
|
|
(home / ".hidden").mkdir()
|
|
(home / "b.PNG").write_bytes(PNG)
|
|
(home / "a.jpg").write_bytes(b"jpeg")
|
|
(home / "notes.txt").write_bytes(b"text")
|
|
(home / ".secret.png").write_bytes(PNG)
|
|
status, listing = self.files(home)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(listing, {
|
|
"directory": str(home), "parent": self.directory.name,
|
|
"directories": [{"name": "Übungen", "path": str(home / "Übungen")}],
|
|
"files": [{"name": "a.jpg", "path": str(home / "a.jpg"), "size": 4},
|
|
{"name": "b.PNG", "path": str(home / "b.PNG"), "size": len(PNG)}]})
|
|
# A path is tidied, so that going up always leads somewhere.
|
|
self.assertEqual(self.files(home / "Übungen" / "..")[1]["directory"], str(home))
|
|
|
|
def test_files_begins_in_the_notebooks_directory(self):
|
|
# Before there are notebooks: the user's home directory.
|
|
self.assertEqual(self.files()[1]["directory"], str(pathlib.Path.home()))
|
|
self.notebooks.mkdir()
|
|
self.assertEqual(self.files()[1]["directory"], str(self.notebooks))
|
|
|
|
def test_files_refusals(self):
|
|
self.assertEqual(self.files("relative/path")[0], 400)
|
|
self.assertEqual(self.files(pathlib.Path(self.directory.name) / "missing")[0], 404)
|
|
self.assertEqual(self.request("GET", "/api/files", token=None)[0], 401)
|
|
self.assertEqual(self.request("GET", "/api/file?path=/etc/hostname", token=None)[0], 401)
|
|
|
|
def test_an_image_file_is_read_and_nothing_else(self):
|
|
directory = pathlib.Path(self.directory.name)
|
|
(directory / "scan.png").write_bytes(PNG)
|
|
(directory / "letter.txt").write_bytes(b"private")
|
|
read = lambda path: self.request("GET", "/api/file?path=" + urllib.parse.quote(str(path)))
|
|
self.assertEqual(read(directory / "scan.png"), (200, PNG))
|
|
self.assertEqual(read(directory / "letter.txt")[0], 400)
|
|
self.assertEqual(read("scan.png")[0], 400)
|
|
self.assertEqual(read(directory / "missing.png")[0], 404)
|
|
|
|
def export(self, directory, name, body=None, replace=False):
|
|
query = urllib.parse.urlencode({"directory": str(directory), "name": name,
|
|
**({"replace": "1"} if replace else {})})
|
|
return self.json("PUT", "/api/file?" + query, body=PNG if body is None else body)
|
|
|
|
def test_export_writes_a_png_where_asked(self):
|
|
directory = pathlib.Path(self.directory.name) / "Export nach Hause"
|
|
directory.mkdir()
|
|
self.assertEqual(self.export(directory, "Seite 1.png"),
|
|
(200, {"saved": str(directory / "Seite 1.png")}))
|
|
self.assertEqual((directory / "Seite 1.png").read_bytes(), PNG)
|
|
self.assertEqual([entry.name for entry in directory.iterdir()], ["Seite 1.png"])
|
|
|
|
def test_export_replaces_a_file_only_when_told_to(self):
|
|
directory = pathlib.Path(self.directory.name)
|
|
(directory / "page.png").write_bytes(b"earlier")
|
|
self.assertEqual(self.export(directory, "page.png"), (409, {"error": "exists"}))
|
|
self.assertEqual((directory / "page.png").read_bytes(), b"earlier")
|
|
self.assertEqual(self.export(directory, "page.png", replace=True)[0], 200)
|
|
self.assertEqual((directory / "page.png").read_bytes(), PNG)
|
|
|
|
def test_export_refusals(self):
|
|
directory = pathlib.Path(self.directory.name)
|
|
self.assertEqual(self.export(directory, "page.svg")[0], 400)
|
|
self.assertEqual(self.export(directory, "../page.png")[0], 400)
|
|
self.assertEqual(self.export(directory, ".hidden.png")[0], 400)
|
|
self.assertEqual(self.export(directory, "page.png", body=b"#!/bin/sh\n")[0], 400)
|
|
self.assertEqual(self.export(directory / "missing", "page.png")[0], 404)
|
|
self.assertEqual(self.export("relative", "page.png")[0], 404)
|
|
query = urllib.parse.urlencode({"directory": str(directory), "name": "page.png"})
|
|
self.assertEqual(self.request("PUT", "/api/file?" + query, body=PNG, token=None)[0], 401)
|
|
self.assertEqual(list(directory.glob("*.png")), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|