rfc-app/backend/tests/test_collection_scoped_serve.py

"""§22 S2 — collection-grained corpus mirror + collection-scoped serve/propose.

The mirror test drives cache.refresh_meta_repo against an in-memory content repo
holding entries under both the default `rfcs/` and a named collection's
`features/rfcs/`, and asserts cached_rfcs is keyed by the right collection_id."""
from __future__ import annotations

import asyncio
import tempfile
from pathlib import Path

from app import cache, db
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="colserve-")) / "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


class _CorpusGitea:
    """A content repo modelled as a flat {path: text} map, listing files under a
    directory prefix and reading them back."""

    def __init__(self, tree: dict[str, str]):
        self._tree = tree

    async def list_dir(self, org, repo, path, ref="main"):
        out = []
        prefix = (path.rstrip("/") + "/") if path else ""
        for p in self._tree:
            if p.startswith(prefix) and "/" not in p[len(prefix):]:
                out.append({"type": "file", "name": p.split("/")[-1], "path": p})
        return out

    async def read_file(self, org, repo, path, ref="main"):
        t = self._tree.get(path)
        return (t, "sha-" + path) if t is not None else None


def _entry_md(slug, title):
    return f"---\nslug: {slug}\ntitle: {title}\nstate: active\n---\nbody\n"


def _seed_project_with_two_collections():
    db.conn().execute(
        "INSERT OR REPLACE INTO projects (id, name, content_repo, visibility, updated_at) "
        "VALUES ('ohm','Ohm','ohm-rfc','public', datetime('now'))")
    for cid, sub in [("default", ""), ("features", "features")]:
        db.conn().execute(
            "INSERT OR REPLACE INTO collections (id, project_id, type, subfolder, initial_state, "
            "visibility, created_at, updated_at) VALUES (?, 'ohm','document',?, "
            "'super-draft','public', datetime('now'), datetime('now'))", (cid, sub))


def test_mirror_keys_entries_by_collection():
    cfg = _db()
    _seed_project_with_two_collections()
    gitea = _CorpusGitea({
        "rfcs/a.md": _entry_md("a", "Default A"),
        "features/rfcs/b.md": _entry_md("b", "Feature B"),
    })
    asyncio.run(cache.refresh_meta_repo(cfg, gitea))
    got = {(r["collection_id"], r["slug"]) for r in
           db.conn().execute("SELECT collection_id, slug FROM cached_rfcs")}
    assert got == {("default", "a"), ("features", "b")}


# --- collection-scoped serve + propose (full app) -----------------------------

from fastapi.testclient import TestClient  # noqa: E402
from test_propose_vertical import (  # noqa: E402,F401
    app_with_fake_gitea, tmp_env, provision_user_row, sign_in_as,
)


def _add_features_collection(content_repo="meta"):
    """Add a named 'features' collection (subfolder 'features') under the seeded
    default project, plus a single entry under features/rfcs/ in the db cache."""
    db.conn().execute(
        "INSERT OR REPLACE INTO collections (id, project_id, type, subfolder, "
        "initial_state, visibility, name, created_at, updated_at) VALUES "
        "('features','default','document','features','super-draft','public','Features', "
        "datetime('now'), datetime('now'))")


def test_scoped_list_returns_only_that_collection(app_with_fake_gitea):
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        _add_features_collection()
        # Seed one entry under each collection's rfcs dir + mirror them in.
        fake.files[("wiggleverse", "meta", "main", "rfcs/a.md")] = {
            "content": _entry_md("a", "Default A"), "sha": "sa"}
        fake.files[("wiggleverse", "meta", "main", "features/rfcs/b.md")] = {
            "content": _entry_md("b", "Feature B"), "sha": "sb"}
        from app import cache as cache_mod, gitea as gitea_mod
        from app.config import load_config
        cfg = load_config()
        asyncio.run(cache_mod.refresh_meta_repo(cfg, gitea_mod.Gitea(cfg)))

        r = client.get("/api/projects/default/collections/features/rfcs")
        assert r.status_code == 200, r.text
        assert [i["slug"] for i in r.json()["items"]] == ["b"]
        # The default collection still serves only its own entry.
        r2 = client.get("/api/projects/default/collections/default/rfcs")
        assert [i["slug"] for i in r2.json()["items"]] == ["a"]


def test_scoped_propose_writes_into_collection_subfolder(app_with_fake_gitea):
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        _add_features_collection()
        provision_user_row(user_id=3, login="alice", role="contributor")
        # §22 S3: an explicitly-created collection requires an explicit scope
        # grant to write (the grandfathered baseline covers only `default`).
        db.conn().execute(
            "INSERT OR REPLACE INTO memberships (scope_type, scope_id, user_id, role) "
            "VALUES ('collection', 'features', 3, 'contributor')")
        sign_in_as(client, user_id=3, gitea_login="alice", display_name="Alice",
                   role="contributor", email="alice@test")
        r = client.post(
            "/api/projects/default/collections/features/rfcs/propose",
            json={"title": "New B", "slug": "newb", "pitch": "x", "tags": []})
        assert r.status_code == 200, r.text
        # The bot wrote the entry under features/rfcs/, not rfcs/.
        keys = {(k[1], k[3]) for k in fake.files
                if k[1] == "meta" and k[3].endswith("newb.md")}
        assert ("meta", "features/rfcs/newb.md") in keys


def test_scoped_routes_404_for_collection_outside_project(app_with_fake_gitea):
    app, _ = app_with_fake_gitea
    with TestClient(app) as client:
        r = client.get("/api/projects/default/collections/nope/rfcs")
        assert r.status_code == 404


def test_s2_anonymous_empty_public_collection(app_with_fake_gitea):
    """C3.6 (@S2): a public collection with no entries; an anonymous visitor
    lands on its catalog → an empty catalog (200, no items), and the propose
    action is not available to them (the propose route rejects anonymous)."""
    app, _ = app_with_fake_gitea
    with TestClient(app) as client:
        _add_features_collection()  # public, no entries
        # Anonymous (no session cookie) reads the empty catalog — 200, [].
        r = client.get("/api/projects/default/collections/features/rfcs")
        assert r.status_code == 200, r.text
        assert r.json()["items"] == []
        # No propose action for an anonymous visitor.
        r2 = client.post(
            "/api/projects/default/collections/features/rfcs/propose",
            json={"title": "X", "slug": "x", "pitch": "p", "tags": []})
        assert r2.status_code == 401