§22 S2: create-collection endpoint (bot commit + registry refresh)

POST /api/projects/<id>/collections, owner/admin-gated, commits a
.collection.yaml to the content repo main via bot.create_collection, then
re-mirrors the registry so the collections row appears (§22.2). Adds GET
list/one collection routes. Extends FakeGitea to model directory listings so
the mirror's content-repo walk is exercised end to end.

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

backend/tests/test_collection_create_vertical.py

@@ -0,0 +1,83 @@
+"""§22 S2 — create-collection vertical: a deployment owner/admin POSTs, the bot
+commits a `.collection.yaml`, and the registry mirror upserts the collections
+row (registry stays the source of truth)."""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from app import db
+from test_propose_vertical import (  # noqa: F401
+    app_with_fake_gitea, tmp_env, provision_user_row, sign_in_as,
+)
+
+
+def test_create_collection_commits_manifest_and_mirrors(app_with_fake_gitea):
+    app, fake = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=1, login="ben", role="owner")
+        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben",
+                   role="owner", email="ben@test")
+        r = client.post("/api/projects/default/collections",
+                        json={"collection_id": "features", "type": "bdd", "name": "Features"})
+        assert r.status_code == 200, r.text
+        assert r.json()["type"] == "bdd"
+
+        # The bot committed the manifest to the content repo's main.
+        f = fake.files.get(("wiggleverse", "meta", "main", "features/.collection.yaml"))
+        assert f is not None
+        assert "type: bdd" in f["content"]
+
+        # The registry refresh mirrored it into a collections row.
+        row = db.conn().execute(
+            "SELECT type, project_id, subfolder FROM collections WHERE id='features'"
+        ).fetchone()
+        assert (row["type"], row["project_id"], row["subfolder"]) == ("bdd", "default", "features")
+
+        # It is now navigable via the directory + scoped serve.
+        items = client.get("/api/projects/default/collections").json()["items"]
+        assert any(c["id"] == "features" for c in items)
+        assert client.get("/api/projects/default/collections/features/rfcs").status_code == 200
+
+
+def test_create_collection_requires_admin(app_with_fake_gitea):
+    app, _ = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=2, login="alice", role="contributor")
+        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice",
+                   role="contributor", email="alice@test")
+        r = client.post("/api/projects/default/collections",
+                        json={"collection_id": "x", "type": "bdd"})
+        assert r.status_code in (401, 403)
+
+
+def test_create_collection_anonymous_rejected(app_with_fake_gitea):
+    app, _ = app_with_fake_gitea
+    with TestClient(app) as client:
+        r = client.post("/api/projects/default/collections",
+                        json={"collection_id": "x", "type": "bdd"})
+        assert r.status_code in (401, 403)
+
+
+def test_create_collection_rejects_duplicate(app_with_fake_gitea):
+    app, _ = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=1, login="ben", role="owner")
+        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben",
+                   role="owner", email="ben@test")
+        ok = client.post("/api/projects/default/collections",
+                         json={"collection_id": "features", "type": "bdd"})
+        assert ok.status_code == 200, ok.text
+        dup = client.post("/api/projects/default/collections",
+                          json={"collection_id": "features", "type": "bdd"})
+        assert dup.status_code == 409
+
+
+def test_create_collection_rejects_reserved_default_id(app_with_fake_gitea):
+    app, _ = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=1, login="ben", role="owner")
+        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben",
+                   role="owner", email="ben@test")
+        r = client.post("/api/projects/default/collections",
+                        json={"collection_id": "default", "type": "bdd"})
+        assert r.status_code == 422

backend/tests/test_propose_vertical.py

@@ -90,6 +90,28 @@ class FakeGitea:
         self._commit_counter += 1
         return f"sha{self._commit_counter:04d}"
 
+    def _dir_listing(self, owner, repo, ref, dirpath):
+        """Children directly under `dirpath` on (owner, repo, ref): files as
+        `type: file` and immediate subdirectories as `type: dir` (the shape real
+        Gitea returns for a contents listing)."""
+        prefix = (dirpath.rstrip("/") + "/") if dirpath else ""
+        files: dict[str, dict] = {}
+        dirs: set[str] = set()
+        for (o, r, br, p), data in self.files.items():
+            if (o, r, br) != (owner, repo, ref) or not p.startswith(prefix):
+                continue
+            rest = p[len(prefix):]
+            if "/" in rest:
+                dirs.add(rest.split("/", 1)[0])
+            elif rest:
+                files[p] = data
+        children = [{"name": n, "path": prefix + n, "type": "dir"} for n in sorted(dirs)]
+        children += [
+            {"name": p.rsplit("/", 1)[-1], "path": p, "type": "file", "sha": d["sha"]}
+            for p, d in sorted(files.items())
+        ]
+        return children
+
     def _enrich_pr(self, owner: str, repo: str, pr: dict) -> dict:
         """Return the PR with mergeability fields filled in.
 
@@ -227,6 +249,16 @@ class FakeGitea:
             }
             return httpx.Response(201, json={"name": new})
 
+        # GET /repos/{owner}/{repo}/contents (root listing, empty path). §22 S2:
+        # the registry mirror walks the content-repo root for collection
+        # subfolders, so the simulator models a root directory listing that
+        # surfaces both file and `dir` children.
+        m_root = re.fullmatch(r"/repos/([^/]+)/([^/]+)/contents/?", path)
+        if method == "GET" and m_root:
+            owner, repo = m_root.groups()
+            ref = request.url.params.get("ref", "main")
+            return httpx.Response(200, json=self._dir_listing(owner, repo, ref, ""))
+
         # GET /repos/{owner}/{repo}/contents/{path}?ref=...
         m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/contents/(.+)", path)
         if method == "GET" and m:
@@ -242,17 +274,8 @@ class FakeGitea:
                     "sha": f["sha"],
                     "content": base64.b64encode(f["content"].encode()).decode(),
                 })
-            # Directory listing
-            prefix = fpath.rstrip("/") + "/"
-            children = []
-            for (o, r, br, p), data in self.files.items():
-                if (o, r, br) == (owner, repo, ref) and p.startswith(prefix) and "/" not in p[len(prefix):]:
-                    children.append({
-                        "name": p.rsplit("/", 1)[-1],
-                        "path": p,
-                        "type": "file",
-                        "sha": data["sha"],
-                    })
+            # Directory listing — both file and subdir children.
+            children = self._dir_listing(owner, repo, ref, fpath)
             if children:
                 return httpx.Response(200, json=children)
             return httpx.Response(404, json={"message": "not found"})