§22 S2: registry mirror reads .collection.yaml manifests

Discover named collections by walking each project's content-repo root for
<subdir>/.collection.yaml; parse + upsert with immutable-type enforcement
(§22.4a) and project-visibility inheritance. The default collection still
flows from projects.yaml.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/tests/test_collection_registry.py

@@ -0,0 +1,146 @@
+"""§22 S2 — the registry mirror reads `.collection.yaml` manifests inside each
+project's content repo and upserts a named collection per manifest. The default
+collection still flows from projects.yaml (test_registry.py)."""
+from __future__ import annotations
+
+import asyncio
+import base64
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from app import db, registry
+from app.config import Config
+
+
+def _db() -> Config:
+    cfg = Config(
+        gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="wiggleverse",
+        registry_repo="registry", oauth_client_id="x",
+        oauth_client_secret="x", app_url="x", secret_key="x",
+        database_path=Path(tempfile.mkdtemp(prefix="colreg-")) / "t.db",
+        owner_gitea_login="x", webhook_secret="x",
+    )
+    db.run_migrations(cfg)
+    if db._CONN is not None:
+        db._CONN.close()
+        db._CONN = None
+    db.init(cfg)
+    return cfg
+
+
+# --- pure parser --------------------------------------------------------------
+
+
+def test_parse_collection_manifest_minimal():
+    doc = registry.parse_collection_manifest("type: bdd\n")
+    assert doc.type == "bdd"
+    # §22.4b: bdd defaults to 'active'; visibility inherits (None == inherit).
+    assert doc.initial_state == "active"
+    assert doc.visibility is None
+    assert doc.name is None
+
+
+def test_parse_collection_manifest_full():
+    doc = registry.parse_collection_manifest(
+        "type: document\nvisibility: public\ninitial_state: active\nname: Model\n"
+    )
+    assert (doc.type, doc.visibility, doc.initial_state, doc.name) == (
+        "document", "public", "active", "Model",
+    )
+
+
+def test_parse_collection_manifest_rejects_bad_type():
+    with pytest.raises(registry.RegistryError):
+        registry.parse_collection_manifest("type: nonsense\n")
+
+
+def test_parse_collection_manifest_rejects_bad_visibility():
+    with pytest.raises(registry.RegistryError):
+        registry.parse_collection_manifest("type: bdd\nvisibility: nope\n")
+
+
+# --- mirror discovery ---------------------------------------------------------
+
+
+class _FakeGitea:
+    """Minimal Gitea stub: projects.yaml in the registry repo + a content repo
+    whose root holds a `features/` subdir carrying a `.collection.yaml`."""
+
+    def __init__(self, projects_yaml: str, repo_tree: dict[str, dict[str, str]]):
+        self._projects_yaml = projects_yaml
+        self._repo_tree = repo_tree  # {repo: {path: text}}
+
+    async def get_contents(self, org, repo, path, ref="main"):
+        if path == "projects.yaml":
+            return {"type": "file",
+                    "content": base64.b64encode(self._projects_yaml.encode()).decode(),
+                    "sha": "regsha-test"}
+        text = self._repo_tree.get(repo, {}).get(path)
+        if text is None:
+            return None
+        return {"type": "file",
+                "content": base64.b64encode(text.encode()).decode(), "sha": "c0ffee"}
+
+    async def list_dir(self, org, repo, path, ref="main"):
+        # Root listing: surface each top-level segment as a 'dir' entry.
+        prefix = (path.rstrip("/") + "/") if path else ""
+        dirs = set()
+        for p in self._repo_tree.get(repo, {}):
+            if not p.startswith(prefix):
+                continue
+            rest = p[len(prefix):]
+            if "/" in rest:
+                dirs.add(rest.split("/", 1)[0])
+        return [{"type": "dir", "name": n, "path": prefix + n} for n in sorted(dirs)]
+
+
+_PROJECTS = (
+    "deployment:\n  name: Ohm\n  tagline: t\n"
+    "projects:\n  - id: ohm\n    name: Ohm\n    type: document\n"
+    "    content_repo: ohm-rfc\n    visibility: public\n"
+)
+
+
+def test_refresh_registry_mirrors_named_collection():
+    cfg = _db()
+    gitea = _FakeGitea(
+        projects_yaml=_PROJECTS,
+        repo_tree={"ohm-rfc": {"features/.collection.yaml": "type: bdd\nname: Features\n"}},
+    )
+    asyncio.run(registry.refresh_registry(cfg, gitea))
+    row = db.conn().execute(
+        "SELECT type, subfolder, name, project_id, visibility FROM collections WHERE id='features'"
+    ).fetchone()
+    assert row is not None
+    assert (row["type"], row["subfolder"], row["project_id"]) == ("bdd", "features", "ohm")
+    assert row["name"] == "Features"
+    # visibility inherits the project's (public) when the manifest omits it.
+    assert row["visibility"] == "public"
+
+
+def test_refresh_registry_leaves_default_collection_intact():
+    cfg = _db()
+    gitea = _FakeGitea(
+        projects_yaml=_PROJECTS,
+        repo_tree={"ohm-rfc": {"features/.collection.yaml": "type: bdd\n"}},
+    )
+    asyncio.run(registry.refresh_registry(cfg, gitea))
+    # The default collection (from projects.yaml) and the named one coexist.
+    ids = {r["id"] for r in db.conn().execute("SELECT id FROM collections")}
+    assert {"default", "features"} <= ids
+
+
+def test_refresh_registry_immutable_type_on_named_collection():
+    cfg = _db()
+    gitea = _FakeGitea(
+        projects_yaml=_PROJECTS,
+        repo_tree={"ohm-rfc": {"features/.collection.yaml": "type: bdd\n"}},
+    )
+    asyncio.run(registry.refresh_registry(cfg, gitea))
+    # A later manifest that flips the type is refused (§22.4a immutable type).
+    gitea._repo_tree["ohm-rfc"]["features/.collection.yaml"] = "type: document\n"
+    asyncio.run(registry.refresh_registry(cfg, gitea))
+    t = db.conn().execute("SELECT type FROM collections WHERE id='features'").fetchone()["type"]
+    assert t == "bdd"