feat(slice4): POST .../meta single-entry edit endpoint + GET RFC meta/can_edit_meta (§22.4a PUC-1)

Direct-commit to the sidecar (D7), contributor+ gated (INV-4), schema-validated
at the write boundary, lazy-migrates a legacy entry, re-ingests. GET RFC now
returns the per-entry meta mapping and a can_edit_meta capability.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-06-07 19:00:40 -07:00
parent aee9b582e5
commit d687a65470
3 changed files with 245 additions and 0 deletions
+12
View File
@@ -29,6 +29,7 @@ from . import (
api_invitations,
api_join_requests,
api_memberships,
api_metadata,
api_notifications,
api_prs,
auth,
@@ -130,6 +131,8 @@ def make_router(
router.include_router(api_prs.make_router(config, gitea, bot, providers))
# Slice 5: §13 graduation + §13.1 claim.
router.include_router(api_graduation.make_router(config, gitea, bot))
# §22.4a SLICE-4/5: entry metadata edit + Owner-gated collection migrate.
router.include_router(api_metadata.make_router(config, gitea, bot))
# Slice 6: §15 notifications surface (inbox, watches, prefs,
# quiet hours, per-user mute, email unsubscribe, bounce webhook).
router.include_router(api_notifications.make_router(config))
@@ -721,6 +724,9 @@ def make_router(
(slug,),
).fetchone()
payload["proposed_use_case"] = uc["use_case"] if uc else None
# §22.4a SLICE-4: contributor+ on the entry's collection may edit metadata.
payload["can_edit_meta"] = bool(
auth.can_contribute_in_collection(viewer, auth.collection_of_rfc(slug)))
return payload
# ---------------------------------------------------------------
@@ -839,6 +845,9 @@ def make_router(
(slug, collection_id),
).fetchone()
payload["proposed_use_case"] = uc["use_case"] if uc else None
# §22.4a SLICE-4: contributor+ on the collection may edit metadata (INV-4).
payload["can_edit_meta"] = bool(
auth.can_contribute_in_collection(viewer, collection_id))
return payload
@router.get("/api/projects/{project_id}/rfcs")
@@ -1387,6 +1396,9 @@ def _serialize_rfc(row) -> dict[str, Any]:
"tags": json.loads(row["tags_json"] or "[]"),
"body": row["body"] or "",
"metadata_malformed": bool(row["metadata_malformed"]),
# §22.4a SLICE-4: the full per-entry metadata mapping (known + custom
# fields) so the detail panel can render schema-driven controls.
"meta": json.loads(row["meta_json"] or "{}"),
}
+87
View File
@@ -0,0 +1,87 @@
"""§22.4a SLICE-4/5 — entry metadata edit endpoints.
`POST .../rfcs/<slug>/meta` writes schema-defined metadata to an entry's sidecar
with a direct commit (D7: direct commit for authorized roles), validated against
the collection's field schema at the write boundary (INV-4), lazy-migrating a
legacy entry to a clean body-only `.md` on first edit. The Owner-gated
`metadata.migrate_collection` operator endpoint also lives here (SLICE-4 carried
work); SLICE-5's bulk endpoint will join it.
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel
from . import (auth, cache, collections as collections_mod,
metadata as metadata_mod, metadata_schema,
projects as projects_mod)
from .bot import Bot
from .config import Config
from .gitea import Gitea, GiteaError
class MetaEditBody(BaseModel):
values: dict[str, Any]
def make_router(config: Config, gitea: Gitea, bot: Bot) -> APIRouter:
router = APIRouter()
def _content_repo() -> tuple[str, str]:
return config.gitea_org, (projects_mod.default_content_repo(config) or "")
def _md_path(collection_id: str, slug: str) -> str:
sub = collections_mod.subfolder_of(collection_id) or ""
rfcs_dir = f"{sub}/rfcs" if sub else "rfcs"
return f"{rfcs_dir}/{slug}.md"
@router.post("/api/projects/{project_id}/collections/{collection_id}/rfcs/{slug}/meta")
async def edit_meta(
project_id: str, collection_id: str, slug: str,
body: MetaEditBody, request: Request,
) -> dict[str, Any]:
viewer = auth.current_user(request)
if collections_mod.project_of_collection(collection_id) != project_id:
raise HTTPException(404, "Collection not in project")
# INV-4: contributor+ on the collection (returns False for anonymous).
if not auth.can_contribute_in_collection(viewer, collection_id):
raise HTTPException(403, "Contributor access required to edit metadata")
col = collections_mod.get_collection(collection_id)
fields = (col or {}).get("fields") or {}
if not fields:
raise HTTPException(422, "Collection declares no editable fields")
if not body.values:
raise HTTPException(422, "Provide at least one field value")
unknown = [k for k in body.values if k not in fields]
if unknown:
raise HTTPException(422, f"Unknown field(s): {', '.join(sorted(unknown))}")
org, repo = _content_repo()
md_path = _md_path(collection_id, slug)
st = await metadata_mod.read_entry_from_git(gitea, org, repo, md_path)
if st is None:
raise HTTPException(404, f"{md_path} not found")
new_entry = metadata_mod.apply_values(st.entry, body.values)
problems = metadata_schema.validate(
metadata_mod.metadata_dict(new_entry), fields)
if problems:
raise HTTPException(422, {"problems": [p.as_dict() for p in problems]})
files = metadata_mod.write_entry_files(md_path, new_entry, st)
try:
await bot.commit_entry_files(
viewer.as_actor(), org=org, repo=repo, files=files,
message=f"Edit metadata: {slug}", branch="main")
except GiteaError as e:
raise HTTPException(502, f"Gitea: {e.detail}")
await cache.refresh_meta_repo(config, gitea)
return {
"ok": True, "slug": slug,
"meta": metadata_mod.metadata_dict(new_entry),
}
return router
@@ -0,0 +1,146 @@
"""SLICE-4 — single-entry metadata edit endpoint (PUC-1, §6.4).
Through the real API: contributor+ gating (INV-4), schema validation at the
write boundary, direct commit to the sidecar with lazy migration, re-ingest,
and the GET RFC `meta` + `can_edit_meta` exposure. Reuses the fake-Gitea harness.
"""
from __future__ import annotations
import asyncio
import json
import yaml
from fastapi.testclient import TestClient
from app import cache, db, gitea as gitea_mod
from app.config import load_config
from test_propose_vertical import ( # noqa: F401 (fixtures)
app_with_fake_gitea,
provision_user_row,
sign_in_as,
tmp_env,
)
PID = "default"
CID = "default"
META = f"/api/projects/{PID}/collections/{CID}"
def _refresh():
cfg = load_config()
asyncio.run(cache.refresh_meta_repo(cfg, gitea_mod.Gitea(cfg)))
def _set_fields(schema):
db.conn().execute(
"UPDATE collections SET config_json = ? WHERE id = 'default'",
(json.dumps({"fields": schema}),))
def _seed_legacy(fake, slug, *, state="active", **front):
fm = {"slug": slug, "title": slug.title(), "state": state, **front}
body = yaml.safe_dump(fm, sort_keys=False).strip()
fake.files[("wiggleverse", "meta", "main", f"rfcs/{slug}.md")] = {
"content": f"---\n{body}\n---\n\nBody.\n", "sha": slug}
def _seed_migrated(fake, slug, sidecar):
fake.files[("wiggleverse", "meta", "main", f"rfcs/{slug}.md")] = {
"content": "Body.\n", "sha": f"{slug}-md"}
fake.files[("wiggleverse", "meta", "main", f"rfcs/{slug}.meta.yaml")] = {
"content": sidecar, "sha": f"{slug}-sc"}
def _login_owner(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")
def test_edit_meta_sets_value_commits_sidecar_and_lazy_migrates(app_with_fake_gitea):
app, fake = app_with_fake_gitea
with TestClient(app) as client:
_set_fields({"priority": {"type": "enum", "values": ["P0", "P1", "P2"]},
"tags": {"type": "tags"}})
_seed_legacy(fake, "a", priority="P1", tags=["x"])
_refresh()
_login_owner(client)
r = client.post(f"{META}/rfcs/a/meta", json={"values": {"priority": "P0"}})
assert r.status_code == 200, r.text
assert r.json()["meta"]["priority"] == "P0"
# sidecar written, .md lazy-migrated to body-only
sc = yaml.safe_load(
fake.files[("wiggleverse", "meta", "main", "rfcs/a.meta.yaml")]["content"])
assert sc["priority"] == "P0"
assert "---" not in fake.files[("wiggleverse", "meta", "main", "rfcs/a.md")]["content"]
# cache reflects the new value
row = db.conn().execute(
"SELECT meta_json FROM cached_rfcs WHERE slug='a'").fetchone()
assert json.loads(row["meta_json"])["priority"] == "P0"
def test_edit_meta_rejects_value_outside_enum(app_with_fake_gitea):
app, fake = app_with_fake_gitea
with TestClient(app) as client:
_set_fields({"priority": {"type": "enum", "values": ["P0", "P1"]}})
_seed_legacy(fake, "a", priority="P1")
_refresh()
_login_owner(client)
r = client.post(f"{META}/rfcs/a/meta", json={"values": {"priority": "ZZZ"}})
assert r.status_code == 422, r.text
# nothing committed
assert ("wiggleverse", "meta", "main", "rfcs/a.meta.yaml") not in fake.files
def test_edit_meta_rejects_unknown_field(app_with_fake_gitea):
app, fake = app_with_fake_gitea
with TestClient(app) as client:
_set_fields({"priority": {"type": "enum", "values": ["P0"]}})
_seed_legacy(fake, "a", priority="P0")
_refresh()
_login_owner(client)
r = client.post(f"{META}/rfcs/a/meta", json={"values": {"nope": "x"}})
assert r.status_code == 422, r.text
def test_edit_meta_forbidden_for_anonymous(app_with_fake_gitea):
app, fake = app_with_fake_gitea
with TestClient(app) as client:
_set_fields({"priority": {"type": "enum", "values": ["P0"]}})
_seed_legacy(fake, "a", priority="P0")
_refresh()
r = client.post(f"{META}/rfcs/a/meta", json={"values": {"priority": "P0"}})
assert r.status_code == 403, r.text
def test_edit_meta_on_already_migrated_entry(app_with_fake_gitea):
app, fake = app_with_fake_gitea
with TestClient(app) as client:
_set_fields({"priority": {"type": "enum", "values": ["P0", "P1"]}})
_seed_migrated(fake, "a", "slug: a\ntitle: A\nstate: active\npriority: P1\n")
_refresh()
_login_owner(client)
r = client.post(f"{META}/rfcs/a/meta", json={"values": {"priority": "P0"}})
assert r.status_code == 200, r.text
sc = yaml.safe_load(
fake.files[("wiggleverse", "meta", "main", "rfcs/a.meta.yaml")]["content"])
assert sc["priority"] == "P0"
# .md untouched (still body-only)
assert fake.files[("wiggleverse", "meta", "main", "rfcs/a.md")]["content"] == "Body.\n"
def test_get_rfc_exposes_meta_and_can_edit(app_with_fake_gitea):
app, fake = app_with_fake_gitea
with TestClient(app) as client:
_set_fields({"priority": {"type": "enum", "values": ["P0", "P1"]}})
_seed_legacy(fake, "a", priority="P1")
_refresh()
# anonymous: meta present, can_edit_meta False
r = client.get(f"{META}/rfcs/a")
assert r.status_code == 200, r.text
assert r.json()["meta"]["priority"] == "P1"
assert r.json()["can_edit_meta"] is False
# owner: can_edit_meta True
_login_owner(client)
r = client.get(f"{META}/rfcs/a")
assert r.json()["can_edit_meta"] is True