diff --git a/server/server.py b/server/server.py index dd9f379..8bdac6c 100644 --- a/server/server.py +++ b/server/server.py @@ -15,7 +15,8 @@ import sys from datetime import datetime, timezone from http.server import HTTPServer, BaseHTTPRequestHandler from pathlib import Path -from urllib.parse import parse_qs, urlparse +from typing import NamedTuple +from urllib.parse import parse_qs, urlparse, unquote class Config: @@ -113,60 +114,71 @@ def build_markdown(entries_text: str, history_text: str) -> str: # --------------------------------------------------------------------------- # File operations # --------------------------------------------------------------------------- +# Path-parameterized cores (*_at) serve any project; the bare-name wrappers +# keep the legacy single-project API bound to the resolved default project. -def read_master() -> dict: - if not CONFIG.master.exists(): - return {"content": build_markdown("", "| Timestamp | Item ID | Action | Details |\n|-----------|---------|--------|---------|"), "checksum": "", "size": 0} - text = CONFIG.master.read_text(encoding="utf-8") - entries, history, meta = parse_markdown_sections(text) +BLANK_HISTORY = "| Timestamp | Item ID | Action | Details |\n|-----------|---------|--------|---------|" + + +class ProjectPaths(NamedTuple): + master: Path + archive: Path + backups_dir: Path + + +def read_master_at(master: Path) -> dict: + if not master.exists(): + return {"content": build_markdown("", BLANK_HISTORY), "checksum": "", "size": 0} + text = master.read_text(encoding="utf-8") + _, _, meta = parse_markdown_sections(text) checksum = meta["checksum"] if meta else "" return {"content": text, "checksum": checksum, "size": len(text.encode("utf-8"))} -def write_master(content: str) -> dict: - """Atomic write with backup.""" - tmp = CONFIG.master.with_suffix(".md.tmp") - # Write temp +def write_master_at(content: str, paths: ProjectPaths) -> dict: + """Atomic write with backup. Creates the project's parent/backups dirs on demand.""" + paths.master.parent.mkdir(parents=True, exist_ok=True) + tmp = paths.master.with_suffix(".md.tmp") tmp.write_text(content, encoding="utf-8") - # Verify it parses - parse_markdown_sections(content) + parse_markdown_sections(content) # verify it parses # Create backup (millis to avoid collisions) timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S") millis = datetime.now(timezone.utc).strftime("%f")[:3] backup_name = f"backlog_{timestamp}-{millis}.md" - backup_path = CONFIG.backups_dir / backup_name - shutil.copy2(tmp, backup_path) - rotate_backups() + paths.backups_dir.mkdir(parents=True, exist_ok=True) + shutil.copy2(tmp, paths.backups_dir / backup_name) + rotate_backups_at(paths.backups_dir) # Atomic rename - tmp.replace(CONFIG.master) - # Stats + tmp.replace(paths.master) + # Stats stay global (single analytics log in the data dir) append_stats({"t": datetime.now(timezone.utc).isoformat(), "e": "save_completed", "d": {"size": len(content.encode("utf-8"))}}) _, _, meta = parse_markdown_sections(content) return {"ok": True, "checksum": meta["checksum"] if meta else "", "saved": meta["saved"] if meta else ""} -def read_archive() -> dict: +def read_archive_at(archive: Path) -> dict: """Read archive.md, return empty structure if not exists.""" - if not CONFIG.archive.exists(): + if not archive.exists(): return {"content": "", "checksum": "", "size": 0, "exists": False} - text = CONFIG.archive.read_text(encoding="utf-8") - entries, history, meta = parse_markdown_sections(text) + text = archive.read_text(encoding="utf-8") + _, _, meta = parse_markdown_sections(text) checksum = meta["checksum"] if meta else "" return {"content": text, "checksum": checksum, "size": len(text.encode("utf-8")), "exists": True} -def write_archive(content: str) -> dict: +def write_archive_at(content: str, paths: ProjectPaths) -> dict: """Write archive.md (append-only semantics enforced at API level).""" - tmp = CONFIG.archive.with_suffix(".md.tmp") + paths.archive.parent.mkdir(parents=True, exist_ok=True) + tmp = paths.archive.with_suffix(".md.tmp") tmp.write_text(content, encoding="utf-8") - parse_markdown_sections(content) # Verify it parses - tmp.replace(CONFIG.archive) + parse_markdown_sections(content) # verify it parses + tmp.replace(paths.archive) _, _, meta = parse_markdown_sections(content) return {"ok": True, "checksum": meta["checksum"] if meta else "", "saved": meta["saved"] if meta else ""} -def rotate_backups(): - files = sorted(CONFIG.backups_dir.glob("backlog_*.md"), key=lambda p: p.stat().st_mtime) +def rotate_backups_at(backups_dir: Path): + files = sorted(backups_dir.glob("backlog_*.md"), key=lambda p: p.stat().st_mtime) if not files: return now = datetime.now(timezone.utc) @@ -180,9 +192,9 @@ def rotate_backups(): f.unlink() -def list_backups() -> list: +def list_backups_at(backups_dir: Path) -> list: result = [] - for f in sorted(CONFIG.backups_dir.glob("backlog_*.md"), key=lambda p: p.stat().st_mtime, reverse=True): + for f in sorted(backups_dir.glob("backlog_*.md"), key=lambda p: p.stat().st_mtime, reverse=True): text = f.read_text(encoding="utf-8") _, _, meta = parse_markdown_sections(text) result.append({ @@ -194,16 +206,45 @@ def list_backups() -> list: return result -def restore_backup(name: str) -> dict: - src = CONFIG.backups_dir / name +def restore_backup_at(name: str, paths: ProjectPaths) -> dict: + src = paths.backups_dir / name if not src.exists(): return {"ok": False, "error": "Backup not found"} text = src.read_text(encoding="utf-8") parse_markdown_sections(text) # validate readable - shutil.copy2(src, CONFIG.master) + paths.master.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src, paths.master) return {"ok": True} +def read_master() -> dict: + return read_master_at(resolve_default_project().master) + + +def write_master(content: str) -> dict: + return write_master_at(content, resolve_default_project()) + + +def read_archive() -> dict: + return read_archive_at(resolve_default_project().archive) + + +def write_archive(content: str) -> dict: + return write_archive_at(content, resolve_default_project()) + + +def rotate_backups(): + rotate_backups_at(resolve_default_project().backups_dir) + + +def list_backups() -> list: + return list_backups_at(resolve_default_project().backups_dir) + + +def restore_backup(name: str) -> dict: + return restore_backup_at(name, resolve_default_project()) + + # --------------------------------------------------------------------------- # Stats # --------------------------------------------------------------------------- @@ -251,6 +292,45 @@ def get_project_path(name: str) -> Path: return Path(registry[name]["path"]).resolve() +def project_paths(name: str) -> ProjectPaths: + """Paths for a registered project; KeyError when the name is unknown.""" + master = get_project_path(name) + return ProjectPaths(master=master, archive=master.parent / "archive.md", backups_dir=master.parent / "backups") + + +def resolve_default_project_name() -> str: + """'default' when registered, else the first registry entry (insertion order).""" + registry = CONFIG.load_registry() + if "default" in registry: + return "default" + return next(iter(registry), "default") + + +def resolve_default_project() -> ProjectPaths: + """Paths legacy routes operate on; CONFIG paths when the registry is empty/corrupt.""" + try: + return project_paths(resolve_default_project_name()) + except KeyError: + return ProjectPaths(master=CONFIG.master, archive=CONFIG.archive, backups_dir=CONFIG.backups_dir) + + +def list_projects() -> list: + """Summary entries for GET /api/projects, in registry insertion order.""" + projects = [] + for name, entry in CONFIG.load_registry().items(): + master = Path(entry["path"]).resolve() + missing = not master.exists() + checksum = "" + size = 0 + if not missing: + size = master.stat().st_size + _, _, meta = parse_markdown_sections(master.read_text(encoding="utf-8")) + if meta: + checksum = meta["checksum"] + projects.append({"name": name, "path": str(master), "size": size, "checksum": checksum, "missing": missing}) + return projects + + def auto_register_default(): registry = CONFIG.load_registry() if registry: @@ -311,6 +391,30 @@ class Handler(BaseHTTPRequestHandler): body = self.rfile.read(length).decode("utf-8") return json.loads(body) + def _read_json_body_or_400(self): + """Parsed JSON body, or None after responding 400 to malformed JSON.""" + try: + return self._read_json_body() + except json.JSONDecodeError as e: + self._json_response({"error": f"Invalid JSON: {e}"}, 400) + return None + + def _project_route_parts(self, path: str) -> tuple: + """'/api/projects/[/...]' → (url-decoded name, action).""" + rest = path[len("/api/projects/"):] + name, _, action = rest.partition("/") + return unquote(name), action + + def _resolve_project_or_404(self, name: str): + """ProjectPaths for name, or None after responding 404 (empty/unknown).""" + try: + if name: + return project_paths(name) + except KeyError: + pass + self._json_response({"error": "Project not found"}, 404) + return None + def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") @@ -347,21 +451,60 @@ class Handler(BaseHTTPRequestHandler): self._json_response({"error": "Not found"}, 404) return + if path == "/api/projects": + self._json_response({"projects": list_projects()}) + return + + if path.startswith("/api/projects/"): + name, action = self._project_route_parts(path) + paths = self._resolve_project_or_404(name) + if paths is None: + return + if action == "backlog": + info = read_master_at(paths.master) + client_checksum = qs.get("checksum", [None])[0] + if client_checksum and client_checksum == info["checksum"]: + self.send_response(304) + self.send_header("Access-Control-Allow-Origin", "*") + self.end_headers() + return + self._json_response({"content": info["content"], "checksum": info["checksum"]}) + return + if action == "archive": + info = read_archive_at(paths.archive) + if not info["exists"]: + self._json_response({"content": "", "checksum": "", "exists": False}) + return + client_checksum = qs.get("checksum", [None])[0] + if client_checksum and client_checksum == info["checksum"]: + self.send_response(304) + self.send_header("Access-Control-Allow-Origin", "*") + self.end_headers() + return + self._json_response({"content": info["content"], "checksum": info["checksum"], "exists": True}) + return + if action == "backups": + self._json_response({"backups": list_backups_at(paths.backups_dir)}) + return + self._json_response({"error": "Not found"}, 404) + return + if path == "/api/health": - info = read_master() - archive_info = read_archive() - backups = list_backups() + paths = resolve_default_project() + info = read_master_at(paths.master) + archive_info = read_archive_at(paths.archive) + backups = list_backups_at(paths.backups_dir) self._json_response({ "status": "ok", "lastSave": info.get("meta", {}).get("saved", "") if isinstance(info, dict) else "", "lastBackup": backups[0]["timestamp"] if backups else "", - "masterSize": CONFIG.master.stat().st_size if CONFIG.master.exists() else 0, + "masterSize": paths.master.stat().st_size if paths.master.exists() else 0, "backupCount": len(backups), - "masterPath": str(CONFIG.master), - "backupsPath": str(CONFIG.backups_dir), + "masterPath": str(paths.master), + "backupsPath": str(paths.backups_dir), "archiveExists": archive_info["exists"], "archiveSize": archive_info["size"], - "archivePath": str(CONFIG.archive), + "archivePath": str(paths.archive), }) return @@ -396,7 +539,7 @@ class Handler(BaseHTTPRequestHandler): if path.startswith("/api/backups/"): name = path[len("/api/backups/"):] - backup_path = CONFIG.backups_dir / name + backup_path = resolve_default_project().backups_dir / name self._file_response(backup_path, "text/markdown") return @@ -411,6 +554,46 @@ class Handler(BaseHTTPRequestHandler): parsed = urlparse(self.path) path = parsed.path + if path.startswith("/api/projects/"): + name, action = self._project_route_parts(path) + body = self._read_json_body_or_400() + if body is None: + return + paths = self._resolve_project_or_404(name) + if paths is None: + return + if action == "backlog": + result = write_master_at(body.get("content", ""), paths) + self._json_response(result) + return + if action == "archive": + # Archive items: save both backlog and archive atomically + backlog_content = body.get("backlog", "") + archive_content = body.get("archive", "") + parse_markdown_sections(backlog_content) + parse_markdown_sections(archive_content) + # Write archive first (append-only), then backlog + archive_result = write_archive_at(archive_content, paths) + backlog_result = write_master_at(backlog_content, paths) + self._json_response({"ok": True, "backlog": backlog_result, "archive": archive_result}) + return + if action == "archive/restore": + # Restore items from archive: save both backlog and archive atomically + backlog_content = body.get("backlog", "") + archive_content = body.get("archive", "") + parse_markdown_sections(backlog_content) + parse_markdown_sections(archive_content) + backlog_result = write_master_at(backlog_content, paths) + archive_result = write_archive_at(archive_content, paths) + self._json_response({"ok": True, "backlog": backlog_result, "archive": archive_result}) + return + if action == "backups/restore": + result = restore_backup_at(body.get("name", ""), paths) + self._json_response(result) + return + self._json_response({"error": "Not found"}, 404) + return + if path == "/api/backlog": body = self._read_json_body() content = body.get("content", "") @@ -496,8 +679,7 @@ class Handler(BaseHTTPRequestHandler): def initialize_storage(): CONFIG.ensure_dirs() if not CONFIG.master.exists(): - blank = build_markdown("", "| Timestamp | Item ID | Action | Details |\n|-----------|---------|--------|---------|") - CONFIG.master.write_text(blank, encoding="utf-8") + CONFIG.master.write_text(build_markdown("", BLANK_HISTORY), encoding="utf-8") print(f"[init] Created blank {CONFIG.master}") auto_register_default() diff --git a/tests/test_server.py b/tests/test_server.py index a90d2ec..0e4ef27 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,20 +1,50 @@ -"""Tests for the Personal Backlog server (registry core, todo 1). +"""Tests for the Personal Backlog server. -stdlib unittest only. Runs the server's startup logic (initialize_storage / -auto_register_default) against an isolated temporary data dir — no port is -bound, no serve_forever loop is started. +stdlib unittest only. + +Two layers: +- TestIsolatedServer: startup/registry logic against an isolated tmp dir (todo 1). +- TestLegacyRoutesBaseline: pins the CURRENT legacy HTTP behavior over a real + HTTPServer thread bound to an ephemeral port. Written BEFORE the per-project + namespace routes exist — it must stay green through the todo-2 refactor so + legacy behavior stays byte-identical. """ +import http.client import io import json import tempfile +import threading import unittest from contextlib import redirect_stdout +from http.server import HTTPServer from pathlib import Path import server.server as ss +MINIMAL_MARKDOWN = ( + "# Backlog\n\n" + "\n\n" + "- [ ] [P1] Alpha item\n\n" + "\n\n" + "| Timestamp | Item ID | Action | Details |\n" + "|-----------|---------|--------|---------|\n\n" + "\n\n" + "\n" +) + +# Pinned by the web frontend (admin.jsx health card) — must never change silently. +HEALTH_KEYS = { + "status", "lastSave", "lastBackup", "masterSize", "backupCount", + "masterPath", "backupsPath", "archiveExists", "archiveSize", "archivePath", +} + + +def make_content(label: str) -> str: + return MINIMAL_MARKDOWN.replace("Alpha item", f"{label} item") + + class TestIsolatedServer(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -79,5 +109,316 @@ class TestIsolatedServer(unittest.TestCase): ss.get_project_path("nope") +class TestLegacyRoutesBaseline(unittest.TestCase): + """Pins legacy single-project HTTP behavior (must hold before AND after todo 2).""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self._orig_config = ss.CONFIG + ss.CONFIG = ss.Config(Path(self.tmp.name), 0) + ss.initialize_storage() + self.httpd = HTTPServer(("127.0.0.1", 0), ss.Handler) + self.port = self.httpd.server_address[1] + self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self.thread.start() + # LIFO: shutdown the loop, then close the socket, then restore CONFIG. + self.addCleanup(setattr, ss, "CONFIG", self._orig_config) + self.addCleanup(self.httpd.server_close) + self.addCleanup(self.httpd.shutdown) + + def request(self, method: str, path: str, json_body=None, raw_body=None): + """One HTTP request → (status, decoded text). json_body dict or raw_body str.""" + if raw_body is not None: + payload = raw_body.encode("utf-8") + elif json_body is not None: + payload = json.dumps(json_body).encode("utf-8") + else: + payload = None + headers = {"Content-Type": "application/json"} if payload is not None else {} + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + conn.request(method, path, body=payload, headers=headers) + resp = conn.getresponse() + data = resp.read() + conn.close() + return resp.status, data.decode("utf-8") + + def request_json(self, method: str, path: str, json_body=None, raw_body=None): + status, text = self.request(method, path, json_body=json_body, raw_body=raw_body) + return status, json.loads(text) + + # GET /api/backlog on a never-written fresh server → blank template + def test_get_backlog_fresh_returns_blank_template(self): + status, data = self.request_json("GET", "/api/backlog") + self.assertEqual(status, 200) + self.assertEqual(set(data.keys()), {"content", "checksum"}) + self.assertIn("# Backlog", data["content"]) + self.assertIn("", data["content"]) + self.assertIn("| Timestamp | Item ID | Action | Details |", data["content"]) + self.assertTrue(data["checksum"].startswith("sha256:")) + + # POST then GET roundtrip returns the written content with its checksum + def test_post_backlog_roundtrip(self): + status, resp = self.request_json("POST", "/api/backlog", json_body={"content": MINIMAL_MARKDOWN}) + self.assertEqual(status, 200) + self.assertTrue(resp["ok"]) + status, data = self.request_json("GET", "/api/backlog") + self.assertEqual(status, 200) + self.assertEqual(data["content"], MINIMAL_MARKDOWN) + self.assertEqual(data["checksum"], resp["checksum"]) + + # Matching ?checksum= → 304 Not Modified (polled by the web SyncPoller) + def test_get_backlog_304_on_matching_checksum(self): + _, data = self.request_json("GET", "/api/backlog") + status, _text = self.request("GET", f"/api/backlog?checksum={data['checksum']}") + self.assertEqual(status, 304) + + # GET unknown API route → 404 JSON error + def test_get_unknown_api_route_404(self): + status, data = self.request_json("GET", "/api/nope") + self.assertEqual(status, 404) + self.assertIn("error", data) + + # /api/health key set is pinned EXACTLY (web consumes these keys) + def test_health_shape_pinned(self): + status, data = self.request_json("GET", "/api/health") + self.assertEqual(status, 200) + self.assertEqual(set(data.keys()), HEALTH_KEYS) + + # Fresh archive is reported as absent + def test_get_archive_fresh_absent(self): + status, data = self.request_json("GET", "/api/archive") + self.assertEqual(status, 200) + self.assertEqual(data, {"content": "", "checksum": "", "exists": False}) + + # Fresh backups listing is empty + def test_get_backups_fresh_empty(self): + status, data = self.request_json("GET", "/api/backups") + self.assertEqual(status, 200) + self.assertEqual(data, {"backups": []}) + + +class TestProjectNamespace(unittest.TestCase): + """Todo 2: /api/projects namespace + legacy aliases (failing-first).""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self._orig_config = ss.CONFIG + ss.CONFIG = ss.Config(Path(self.tmp.name), 0) + ss.initialize_storage() + self.httpd = HTTPServer(("127.0.0.1", 0), ss.Handler) + self.port = self.httpd.server_address[1] + self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self.thread.start() + self.addCleanup(setattr, ss, "CONFIG", self._orig_config) + self.addCleanup(self.httpd.server_close) + self.addCleanup(self.httpd.shutdown) + + def request(self, method: str, path: str, json_body=None, raw_body=None): + if raw_body is not None: + payload = raw_body.encode("utf-8") + elif json_body is not None: + payload = json.dumps(json_body).encode("utf-8") + else: + payload = None + headers = {"Content-Type": "application/json"} if payload is not None else {} + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + conn.request(method, path, body=payload, headers=headers) + resp = conn.getresponse() + data = resp.read() + conn.close() + return resp.status, data.decode("utf-8") + + def request_json(self, method: str, path: str, json_body=None, raw_body=None): + status, text = self.request(method, path, json_body=json_body, raw_body=raw_body) + return status, json.loads(text) + + def register_project(self, name: str, master: Path, content: str): + master.write_text(content, encoding="utf-8") + registry = ss.CONFIG.load_registry() + registry[name] = {"path": str(master), "name": name} + ss.CONFIG.save_registry(registry) + + # (a) legacy GET /api/backlog body == GET /api/projects/default/backlog body + def test_legacy_backlog_equals_default_project_backlog(self): + _, legacy = self.request_json("GET", "/api/backlog") + status, namespaced = self.request_json("GET", "/api/projects/default/backlog") + self.assertEqual(status, 200) + self.assertEqual(legacy, namespaced) + + # (b) POST then GET roundtrip on the namespace route + def test_project_backlog_roundtrip_default(self): + content = make_content("namespaced") + status, resp = self.request_json("POST", "/api/projects/default/backlog", + json_body={"content": content}) + self.assertEqual(status, 200) + self.assertTrue(resp["ok"]) + self.assertTrue(resp["checksum"]) + status, data = self.request_json("GET", "/api/projects/default/backlog") + self.assertEqual(status, 200) + self.assertEqual(data["content"], content) + self.assertEqual(data["checksum"], resp["checksum"]) + + # (c) unknown project → 404 {"error": "Project not found"} on GET and POST + def test_unknown_project_404(self): + status, data = self.request_json("GET", "/api/projects/nope/backlog") + self.assertEqual(status, 404) + self.assertEqual(data, {"error": "Project not found"}) + status, data = self.request_json("POST", "/api/projects/nope/backlog", + json_body={"content": MINIMAL_MARKDOWN}) + self.assertEqual(status, 404) + self.assertEqual(data, {"error": "Project not found"}) + + # (e) two extra registered projects: listing + strict write isolation + def test_projects_listing_and_write_isolation(self): + tmp_a = tempfile.TemporaryDirectory() + self.addCleanup(tmp_a.cleanup) + tmp_b = tempfile.TemporaryDirectory() + self.addCleanup(tmp_b.cleanup) + alpha = Path(tmp_a.name) / "backlog.md" + beta = Path(tmp_b.name) / "backlog.md" + self.register_project("alpha", alpha, make_content("alpha-original")) + self.register_project("beta", beta, make_content("beta-original")) + + status, data = self.request_json("GET", "/api/projects") + self.assertEqual(status, 200) + self.assertEqual([p["name"] for p in data["projects"]], ["default", "alpha", "beta"]) + alpha_entry = data["projects"][1] + self.assertEqual(alpha_entry["path"], str(alpha)) + self.assertEqual(alpha_entry["size"], alpha.stat().st_size) + self.assertTrue(alpha_entry["checksum"].startswith("sha256:")) + self.assertFalse(alpha_entry["missing"]) + + default_before = ss.CONFIG.master.read_bytes() + beta_before = beta.read_bytes() + rewritten = make_content("alpha-rewritten") + status, resp = self.request_json("POST", "/api/projects/alpha/backlog", + json_body={"content": rewritten}) + self.assertEqual(status, 200) + self.assertTrue(resp["ok"]) + + # Isolation: only alpha's file changed. + self.assertEqual(alpha.read_text(encoding="utf-8"), rewritten) + self.assertEqual(ss.CONFIG.master.read_bytes(), default_before) + self.assertEqual(beta.read_bytes(), beta_before) + + # alpha's backups live under alpha's own directory, and are listed. + backups_dir = Path(tmp_a.name) / "backups" + self.assertTrue(backups_dir.is_dir()) + self.assertTrue(any(backups_dir.glob("backlog_*.md"))) + status, data = self.request_json("GET", "/api/projects/alpha/backups") + self.assertEqual(status, 200) + self.assertGreaterEqual(len(data["backups"]), 1) + self.assertEqual(list(ss.CONFIG.backups_dir.glob("backlog_*.md")), []) + + # Per-project archive dual-write: both files land next to the project file + def test_per_project_archive_dual_write(self): + tmp_a = tempfile.TemporaryDirectory() + self.addCleanup(tmp_a.cleanup) + alpha = Path(tmp_a.name) / "backlog.md" + self.register_project("alpha", alpha, make_content("alpha-original")) + + backlog_content = make_content("after-archive") + archive_content = MINIMAL_MARKDOWN.replace("Alpha item", "archived item") + status, resp = self.request_json("POST", "/api/projects/alpha/archive", + json_body={"backlog": backlog_content, + "archive": archive_content}) + self.assertEqual(status, 200) + self.assertTrue(resp["ok"]) + self.assertTrue(resp["backlog"]["ok"]) + self.assertTrue(resp["archive"]["ok"]) + self.assertEqual(alpha.read_text(encoding="utf-8"), backlog_content) + self.assertEqual((Path(tmp_a.name) / "archive.md").read_text(encoding="utf-8"), + archive_content) + # default project untouched + self.assertFalse(ss.CONFIG.archive.exists()) + + status, data = self.request_json("GET", "/api/projects/alpha/archive") + self.assertEqual(status, 200) + self.assertEqual(data["exists"], True) + self.assertEqual(data["content"], archive_content) + + # Per-project backup restore puts the chosen backup back into the project file + def test_per_project_backup_restore(self): + tmp_a = tempfile.TemporaryDirectory() + self.addCleanup(tmp_a.cleanup) + alpha = Path(tmp_a.name) / "backlog.md" + first = make_content("first") + second = make_content("second") + self.register_project("alpha", alpha, make_content("seed")) + status, resp = self.request_json("POST", "/api/projects/alpha/backlog", + json_body={"content": first}) + self.assertTrue(resp["ok"]) + status, resp = self.request_json("POST", "/api/projects/alpha/backlog", + json_body={"content": second}) + self.assertTrue(resp["ok"]) + + status, data = self.request_json("GET", "/api/projects/alpha/backups") + self.assertEqual(status, 200) + names = [b["name"] for b in data["backups"]] # newest first + self.assertEqual(len(names), 2) + + status, resp = self.request_json("POST", "/api/projects/alpha/backups/restore", + json_body={"name": names[-1]}) # oldest + self.assertEqual(status, 200) + self.assertTrue(resp["ok"]) + self.assertEqual(alpha.read_text(encoding="utf-8"), first) + + # (f) rename default → main: legacy routes alias the first registry entry + def test_first_entry_alias_after_default_rename(self): + content = make_content("alias-target") + status, resp = self.request_json("POST", "/api/backlog", json_body={"content": content}) + self.assertTrue(resp["ok"]) + master = ss.CONFIG.master + ss.CONFIG.save_registry({"main": {"path": str(master), "name": "main"}}) + + status, data = self.request_json("GET", "/api/backlog") + self.assertEqual(status, 200) + self.assertEqual(data["content"], content) + + status, namespaced = self.request_json("GET", "/api/projects/main/backlog") + self.assertEqual(status, 200) + self.assertEqual(namespaced, data) + + # health keeps its exact shape and still describes the aliased project + status, health = self.request_json("GET", "/api/health") + self.assertEqual(status, 200) + self.assertEqual(set(health.keys()), HEALTH_KEYS) + self.assertEqual(health["masterPath"], str(master)) + + # Adversarial: malformed JSON body on a namespace POST → graceful 400, server stays up + def test_post_project_backlog_invalid_json_400(self): + status, data = self.request_json("POST", "/api/projects/default/backlog", + raw_body='{"content": not-json') + self.assertEqual(status, 400) + self.assertIn("error", data) + # server still alive afterwards + status, _ = self.request("GET", "/api/backlog") + self.assertEqual(status, 200) + + # Adversarial: empty project name → 404 + def test_empty_project_name_404(self): + status, _ = self.request("GET", "/api/projects/") + self.assertEqual(status, 404) + status, _ = self.request("GET", "/api/projects//backlog") + self.assertEqual(status, 404) + status, _ = self.request("POST", "/api/projects//backlog", + json_body={"content": MINIMAL_MARKDOWN}) + self.assertEqual(status, 404) + + # Registry edits on disk are picked up by the NEXT request (no caching) + def test_registry_reload_between_requests(self): + tmp_a = tempfile.TemporaryDirectory() + self.addCleanup(tmp_a.cleanup) + alpha = Path(tmp_a.name) / "backlog.md" + status, _ = self.request("GET", "/api/projects/alpha/backlog") + self.assertEqual(status, 404) + self.register_project("alpha", alpha, make_content("late-registered")) + status, data = self.request_json("GET", "/api/projects/alpha/backlog") + self.assertEqual(status, 200) + self.assertEqual(data["content"], make_content("late-registered")) + + if __name__ == "__main__": unittest.main()