0c654b173d
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>
84 lines
3.8 KiB
Python
84 lines
3.8 KiB
Python
"""§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
|