fujin/backlog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add per-project API namespace with legacy default aliases

Browse Source
This commit is contained in:
Alexander
2026-08-24 12:33:06 +02:00
parent fcb207b228
commit 41d419e5f8
2 changed files with 569 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/server.py
+224 -42
View File
@@ -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/<name>[/<action>...]' → (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()
tests/test_server.py
+345 -4
View File
@@ -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"
"<!-- SECTION: ENTRIES -->\n\n"
"- [ ] [P1] Alpha item\n\n"
"<!-- SECTION: HISTORY -->\n\n"
"| Timestamp | Item ID | Action | Details |\n"
"|-----------|---------|--------|---------|\n\n"
"<!-- SECTION: INTEGRITY -->\n\n"
"<!-- saved: 2025-01-01T00:00:00Z | checksum: sha256:test | entries: 1 | history: 1 -->\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("<!-- SECTION: ENTRIES -->", 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()