Kokkoja: first public assembly
Assembled from the development repository by tools/make_release.sh. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
ackandClaude Fable 5.1 committed
commit
95c80d04c9
39 files changed
+9905
No files matched your search
@@ -0,0 +1,260 @@
|
||||
"""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"
|
||||
WRITTEN_PAGE = b'<svg data-kokkoja="1"><g data-layer="a"><g data-stroke="s"/></g></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"
|
||||
self.server = kokkoja.make_server(0, "ws://relay.example:8770/", self.notebooks, TOKEN,
|
||||
self.config)
|
||||
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_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"])
|
||||
|
||||
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()
|
||||
Reference in new issue
Block a user