rfc-app/backend/tests/test_collection_registry.py

"""§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"