"""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'\n' PNG = b"\x89PNG\r\n\x1a\n" + b"the rest of a picture" TEXT_ONLY_LAYER = (b'' b'hellohello') WRITTEN_PAGE = b'\n' def page_named(session, page, number, body=b''): """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'\n' b'\n' b'Kokkoja page ' + number + b'\n' b'\n' + body + b'\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"]) listed = self.json("GET", "/api/notebooks/" + urllib.parse.quote("Übungen A1")) self.assertEqual(listed[1]["pages"], ["2026_10_02_page.svg"]) self.assertAlmostEqual(listed[1]["changed"]["2026_10_02_page.svg"] / 1000, time.time(), delta=10) 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"" 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'' + TEXT_ONLY_LAYER + b"\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'data-page="p1"') 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'data-page="p1"', b'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": [], "changed": {}})) # 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()