Add project registry core with default auto-registration
This commit is contained in:
+53
-6
@@ -26,12 +26,27 @@ class Config:
|
||||
self.archive = self.dir / "archive.md"
|
||||
self.backups_dir = self.dir / "backups"
|
||||
self.stats_file = self.dir / "stats.jsonl"
|
||||
self.registry_file = self.dir / "projects.json"
|
||||
self.web_dir = web_dir.resolve() if web_dir else Path(__file__).parent.parent / "webapp"
|
||||
|
||||
def ensure_dirs(self):
|
||||
self.dir.mkdir(parents=True, exist_ok=True)
|
||||
self.backups_dir.mkdir(exist_ok=True)
|
||||
|
||||
def load_registry(self) -> dict:
|
||||
if not self.registry_file.exists():
|
||||
return {}
|
||||
try:
|
||||
return json.loads(self.registry_file.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"[registry] projects.json unreadable, starting empty: {e}")
|
||||
return {}
|
||||
|
||||
def save_registry(self, registry: dict):
|
||||
tmp = self.registry_file.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(registry, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
tmp.replace(self.registry_file)
|
||||
|
||||
|
||||
CONFIG: Config = None # type: ignore[assignment]
|
||||
|
||||
@@ -220,6 +235,34 @@ def read_stats(from_iso: str = None, to_iso: str = None) -> list:
|
||||
return events
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Project registry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def safe_project_name(name: str) -> str:
|
||||
"""Collapse to [a-zA-Z0-9_-]; empty result means the name is invalid."""
|
||||
return re.sub(r"[^a-zA-Z0-9_-]", "-", name.strip())
|
||||
|
||||
|
||||
def get_project_path(name: str) -> Path:
|
||||
registry = CONFIG.load_registry()
|
||||
if name not in registry:
|
||||
raise KeyError(name)
|
||||
return Path(registry[name]["path"]).resolve()
|
||||
|
||||
|
||||
def auto_register_default():
|
||||
registry = CONFIG.load_registry()
|
||||
if registry:
|
||||
return
|
||||
if not CONFIG.master.exists():
|
||||
return
|
||||
path = str(CONFIG.master.resolve())
|
||||
registry["default"] = {"path": path, "name": "default"}
|
||||
CONFIG.save_registry(registry)
|
||||
print(f"[init] Registered default project: {path}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTTP Handler
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -450,6 +493,15 @@ class Handler(BaseHTTPRequestHandler):
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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")
|
||||
print(f"[init] Created blank {CONFIG.master}")
|
||||
auto_register_default()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Personal Backlog Server")
|
||||
parser.add_argument("--port", type=int, default=8080, help="Port to listen on")
|
||||
@@ -461,12 +513,7 @@ def main():
|
||||
global CONFIG
|
||||
web_dir = Path(args.web_dir) if args.web_dir else None
|
||||
CONFIG = Config(Path(args.dir), args.port, web_dir)
|
||||
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")
|
||||
print(f"[init] Created blank {CONFIG.master}")
|
||||
initialize_storage()
|
||||
|
||||
server = HTTPServer(("0.0.0.0", args.port), Handler)
|
||||
print(f"[server] Listening on http://0.0.0.0:{args.port}")
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Tests for the Personal Backlog server (registry core, todo 1).
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
import io
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
from pathlib import Path
|
||||
|
||||
import server.server as ss
|
||||
|
||||
|
||||
class TestIsolatedServer(unittest.TestCase):
|
||||
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), 8080)
|
||||
ss.CONFIG.ensure_dirs()
|
||||
|
||||
def tearDown(self):
|
||||
ss.CONFIG = self._orig_config
|
||||
|
||||
# (1) fresh dir -> startup registers exactly the default project
|
||||
def test_initialize_storage_registers_default_on_fresh_dir(self):
|
||||
ss.initialize_storage()
|
||||
registry_path = ss.CONFIG.registry_file
|
||||
self.assertTrue(registry_path.exists(), "projects.json must exist after startup")
|
||||
with open(registry_path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
self.assertEqual(
|
||||
data,
|
||||
{"default": {"path": str(ss.CONFIG.master.resolve()), "name": "default"}},
|
||||
)
|
||||
|
||||
# (2) registry with existing entries -> auto_register_default is a no-op
|
||||
def test_auto_register_default_noop_when_registry_has_entries(self):
|
||||
existing = {"work": {"path": "/somewhere/work/backlog.md", "name": "work"}}
|
||||
ss.CONFIG.save_registry(existing)
|
||||
ss.auto_register_default()
|
||||
with open(ss.CONFIG.registry_file, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
self.assertEqual(data, existing)
|
||||
|
||||
# (3) corrupt projects.json -> load_registry returns {} without raising
|
||||
def test_load_registry_corrupt_file_returns_empty(self):
|
||||
ss.CONFIG.registry_file.write_text("{bad", encoding="utf-8")
|
||||
captured = io.StringIO()
|
||||
with redirect_stdout(captured):
|
||||
result = ss.CONFIG.load_registry()
|
||||
self.assertEqual(result, {})
|
||||
self.assertIn("[registry] projects.json unreadable", captured.getvalue())
|
||||
|
||||
# (4) safe_project_name sanitizes and rejects empty
|
||||
def test_safe_project_name(self):
|
||||
self.assertEqual(ss.safe_project_name("My Proj!"), "My-Proj-")
|
||||
self.assertEqual(ss.safe_project_name(" "), "")
|
||||
|
||||
# (5) save_registry -> load_registry roundtrip preserves the dict
|
||||
def test_registry_roundtrip_preserves_dict(self):
|
||||
registry = {
|
||||
"default": {"path": "/data/backlog.md", "name": "default"},
|
||||
"work": {"path": "/home/me/work/backlog.md", "name": "work"},
|
||||
}
|
||||
ss.CONFIG.save_registry(registry)
|
||||
self.assertEqual(ss.CONFIG.load_registry(), registry)
|
||||
|
||||
# (6) get_project_path resolves registered paths, KeyError otherwise
|
||||
def test_get_project_path(self):
|
||||
master = ss.CONFIG.master.resolve()
|
||||
ss.CONFIG.save_registry({"default": {"path": str(master), "name": "default"}})
|
||||
self.assertEqual(ss.get_project_path("default"), master)
|
||||
with self.assertRaises(KeyError):
|
||||
ss.get_project_path("nope")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user