benstull/rfc-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Slice 1: scaffolding + propose-to-super-draft vertical

Browse Source 
Brings the §1 bot wrapper, the §4 cache (webhook + reconciler), the
§5 schema (six numbered migrations), Gitea OAuth + §6 user
provisioning, the §7 catalog left pane, and the propose-to-merge
vertical: propose modal opens an idea PR against the meta repo, an
owner merges from the pending-idea view, the cache picks it up via
webhook or reconciler sweep, and the catalog renders the new
super-draft.

Per §1 the bot is the only Git writer; every commit, branch
creation, and PR merge carries the §6.5 On-behalf-of: trailer and
an `actions` audit row. Per §4 the cache is never written from a
user action — it's webhook+reconciler only.

Covered by `backend/tests/test_propose_vertical.py` against an
in-process Gitea simulator.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-05-24 04:31:11 -07:00
commit 779ba6db59
42 changed files with 10385 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_propose_vertical.py
+440
View File
@@ -0,0 +1,440 @@
"""End-to-end integration test for the Slice 1 vertical.
Stands up the FastAPI app against a mocked Gitea transport that
simulates the meta repo and the propose-to-merge lifecycle. The test
walks the same path a user would: sign in (a forged session cookie
substitutes for the OAuth round-trip, since OAuth itself is not in
scope to mock end-to-end), open a propose modal
(POST /api/rfcs/propose), exercise the bot wrapper through to the
Gitea HTTP layer, merge the PR as an owner, refresh the cache, and
verify the super-draft surfaces in GET /api/rfcs and
GET /api/rfcs/<slug>.
The mocked Gitea is intentionally narrow — it only honors the
endpoints the slice actually exercises. Adding routes to it as later
slices land is the right shape: the test surface tracks the production
surface.
"""
from __future__ import annotations
import base64
import json
import os
import re
import tempfile
from pathlib import Path
import httpx
import pytest
# ---------------------------------------------------------------------------
# Fake Gitea
# ---------------------------------------------------------------------------
class FakeGitea:
"""A narrow in-memory simulation of the Gitea API the slice uses."""
def __init__(self):
# files: (owner, repo, branch, path) -> {"content": str, "sha": str}
self.files: dict[tuple[str, str, str, str], dict] = {}
# branches: (owner, repo) -> {branch_name -> {"sha": str}}
self.branches: dict[tuple[str, str], dict[str, dict]] = {}
# pulls: (owner, repo) -> list[pull-dict]
self.pulls: dict[tuple[str, str], list[dict]] = {}
self._pr_counter = 0
self._commit_counter = 0
self._seed_repo("wiggleverse", "meta")
def _seed_repo(self, owner, repo):
self.branches[(owner, repo)] = {"main": {"sha": "initial"}}
self.pulls[(owner, repo)] = []
def _next_sha(self):
self._commit_counter += 1
return f"sha{self._commit_counter:04d}"
def handle(self, request: httpx.Request) -> httpx.Response:
path = request.url.path.replace("/api/v1", "", 1)
method = request.method
body = request.read().decode() if request.content else ""
payload = json.loads(body) if body else {}
# GET /repos/{owner}/{repo}
if method == "GET" and re.fullmatch(r"/repos/[^/]+/[^/]+", path):
return httpx.Response(200, json={"name": path.split("/")[-1]})
# GET /repos/{owner}/{repo}/branches/{branch}
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/branches/([^/]+)", path)
if method == "GET" and m:
owner, repo, branch = m.groups()
b = self.branches.get((owner, repo), {}).get(branch)
if not b:
return httpx.Response(404, json={"message": "not found"})
return httpx.Response(200, json={"name": branch, "commit": {"id": b["sha"]}})
# POST /repos/{owner}/{repo}/branches
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/branches", path)
if method == "POST" and m:
owner, repo = m.groups()
new = payload["new_branch_name"]
old = payload["old_branch_name"]
old_sha = self.branches[(owner, repo)][old]["sha"]
self.branches[(owner, repo)][new] = {"sha": old_sha}
# Copy main's files into the new branch
for (o, r, br, p), data in list(self.files.items()):
if (o, r, br) == (owner, repo, old):
self.files[(owner, repo, new, p)] = dict(data)
return httpx.Response(201, json={"name": new})
# GET /repos/{owner}/{repo}/contents/{path}?ref=...
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/contents/(.+)", path)
if method == "GET" and m:
owner, repo, fpath = m.groups()
ref = request.url.params.get("ref", "main")
key = (owner, repo, ref, fpath)
if key in self.files:
f = self.files[key]
return httpx.Response(200, json={
"name": fpath.rsplit("/", 1)[-1],
"path": fpath,
"type": "file",
"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"],
})
if children:
return httpx.Response(200, json=children)
return httpx.Response(404, json={"message": "not found"})
# POST /repos/{owner}/{repo}/contents/{path}
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/contents/(.+)", path)
if method == "POST" and m:
owner, repo, fpath = m.groups()
branch = payload["branch"]
content = base64.b64decode(payload["content"]).decode()
sha = self._next_sha()
self.files[(owner, repo, branch, fpath)] = {"content": content, "sha": sha}
self.branches[(owner, repo)][branch]["sha"] = sha
return httpx.Response(201, json={"commit": {"sha": sha}})
# GET /repos/{owner}/{repo}/pulls?state=...
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/pulls", path)
if method == "GET" and m:
owner, repo = m.groups()
state = request.url.params.get("state", "open")
items = self.pulls.get((owner, repo), [])
filtered = [p for p in items if (state == "all") or (p["state"] == state)]
return httpx.Response(200, json=filtered)
# POST /repos/{owner}/{repo}/pulls
if method == "POST" and m:
owner, repo = m.groups()
self._pr_counter += 1
head_branch = payload["head"]
pr = {
"number": self._pr_counter,
"title": payload["title"],
"body": payload["body"],
"head": {"ref": head_branch, "sha": self.branches[(owner, repo)][head_branch]["sha"]},
"base": {"ref": payload["base"]},
"state": "open",
"merged": False,
"merged_at": None,
"closed_at": None,
"created_at": "2026-05-23T00:00:00Z",
"user": {"login": "rfc-bot"},
}
self.pulls[(owner, repo)].append(pr)
return httpx.Response(201, json=pr)
# POST /repos/{owner}/{repo}/pulls/{number}/merge
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/pulls/(\d+)/merge", path)
if method == "POST" and m:
owner, repo, num = m.groups()
for pr in self.pulls[(owner, repo)]:
if pr["number"] == int(num):
head_branch = pr["head"]["ref"]
for (o, r, br, p), data in list(self.files.items()):
if (o, r, br) == (owner, repo, head_branch):
self.files[(owner, repo, "main", p)] = dict(data)
# Real Gitea: state becomes "closed" with merged=true.
pr["state"] = "closed"
pr["merged"] = True
pr["merged_at"] = "2026-05-23T01:00:00Z"
pr["closed_at"] = "2026-05-23T01:00:00Z"
new_sha = self._next_sha()
self.branches[(owner, repo)]["main"]["sha"] = new_sha
return httpx.Response(200, json={"merged": True})
return httpx.Response(404, json={"message": "not found"})
# GET /repos/{owner}/{repo}/hooks
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/hooks", path)
if method == "GET" and m:
return httpx.Response(200, json=[])
# PATCH /repos/{owner}/{repo}/issues/{number} — Gitea close path.
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/issues/(\d+)", path)
if method == "PATCH" and m:
owner, repo, num = m.groups()
for pr in self.pulls.get((owner, repo), []):
if pr["number"] == int(num) and payload.get("state") == "closed":
pr["state"] = "closed"
pr["closed_at"] = "2026-05-23T02:00:00Z"
return httpx.Response(200, json={"state": "closed"})
return httpx.Response(200, json={})
# POST /repos/{owner}/{repo}/issues/{number}/comments
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/issues/(\d+)/comments", path)
if method == "POST" and m:
return httpx.Response(201, json={"id": 1, "body": payload.get("body", "")})
return httpx.Response(404, json={"message": f"unmocked {method} {path}"})
# ---------------------------------------------------------------------------
# Session helpers — forge a SessionMiddleware cookie directly to skip OAuth.
# ---------------------------------------------------------------------------
def _sign_session(session_data: dict, secret: str) -> str:
from itsdangerous import TimestampSigner
data = base64.b64encode(json.dumps(session_data).encode("utf-8"))
signer = TimestampSigner(secret)
return signer.sign(data).decode("utf-8")
def sign_in_as(client, *, user_id, gitea_login, display_name, role, email=""):
payload = {
"user": {
"user_id": user_id,
"gitea_id": user_id,
"gitea_login": gitea_login,
"display_name": display_name,
"email": email,
"avatar_url": "",
"role": role,
}
}
cookie = _sign_session(payload, os.environ["SECRET_KEY"])
client.cookies.set("rfc_session", cookie)
def provision_user_row(*, user_id: int, login: str, role: str) -> None:
from app import db
db.conn().execute(
"""
INSERT OR REPLACE INTO users (id, gitea_id, gitea_login, email, display_name, avatar_url, role)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(user_id, user_id, login, f"{login}@test", login.capitalize(), "", role),
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def tmp_env(monkeypatch):
tmpdir = tempfile.mkdtemp(prefix="rfc-app-test-")
db_path = Path(tmpdir) / "test.db"
env = {
"GITEA_URL": "http://gitea.test",
"GITEA_BOT_USER": "rfc-bot",
"GITEA_BOT_TOKEN": "bot-token",
"GITEA_ORG": "wiggleverse",
"META_REPO": "meta",
"OAUTH_CLIENT_ID": "cid",
"OAUTH_CLIENT_SECRET": "csec",
"APP_URL": "http://localhost:8000",
"SECRET_KEY": "test-secret-key-for-cookies",
"DATABASE_PATH": str(db_path),
"OWNER_GITEA_LOGIN": "ben",
"GITEA_WEBHOOK_SECRET": "",
"ENABLED_MODELS": "claude",
}
for k, v in env.items():
monkeypatch.setenv(k, v)
yield env
@pytest.fixture
def app_with_fake_gitea(tmp_env, monkeypatch):
fake = FakeGitea()
real_client_cls = httpx.AsyncClient
def patched_client(*args, **kwargs):
kwargs["transport"] = httpx.MockTransport(fake.handle)
return real_client_cls(*args, **kwargs)
monkeypatch.setattr("app.gitea.httpx.AsyncClient", patched_client)
# The db module memoizes its connection — reset across tests so each
# test gets the tmpdir db its env points at, not a previous test's.
from app import db
if db._CONN is not None:
db._CONN.close()
db._CONN = None
from app.main import create_app
app = create_app()
return app, fake
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_propose_to_super_draft_vertical(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
app, _fake = app_with_fake_gitea
with TestClient(app) as client:
# The catalog is empty before anything happens.
r = client.get("/api/rfcs")
assert r.status_code == 200
assert r.json()["items"] == []
# A contributor proposes a new RFC.
provision_user_row(user_id=2, login="alice", role="contributor")
provision_user_row(user_id=1, login="ben", role="owner")
sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor", email="alice@test")
r = client.post("/api/rfcs/propose", json={
"title": "Open Human Model",
"slug": "open-human-model",
"pitch": "A shared definition of what we mean by *human*.",
"tags": ["identity", "schema"],
})
assert r.status_code == 200, r.text
pr_number = r.json()["pr_number"]
assert r.json()["slug"] == "open-human-model"
# The proposal surfaces on the pending-ideas list.
r = client.get("/api/proposals")
items = r.json()["items"]
assert len(items) == 1
assert items[0]["slug"] == "open-human-model"
assert items[0]["pr_number"] == pr_number
# A contributor cannot merge.
r = client.post(f"/api/proposals/{pr_number}/merge")
assert r.status_code == 403
# Switch to the owner. The pending-idea view exposes the merge affordance.
sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner", email="ben@test")
r = client.get(f"/api/proposals/{pr_number}")
assert r.status_code == 200, r.text
proposal = r.json()
assert proposal["entry"]["title"] == "Open Human Model"
assert proposal["entry"]["state"] == "super-draft"
assert proposal["affordances"]["merge"] is True
# Owner merges. The catalog picks up the new super-draft.
r = client.post(f"/api/proposals/{pr_number}/merge")
assert r.status_code == 200, r.text
assert r.json()["slug"] == "open-human-model"
r = client.get("/api/rfcs")
items = r.json()["items"]
assert len(items) == 1
assert items[0]["slug"] == "open-human-model"
assert items[0]["state"] == "super-draft"
assert "identity" in items[0]["tags"]
# The super-draft view renders the body.
r = client.get("/api/rfcs/open-human-model")
assert r.status_code == 200
view = r.json()
assert view["state"] == "super-draft"
assert "shared definition" in view["body"]
# The pending-ideas list no longer carries the merged proposal.
r = client.get("/api/proposals")
assert r.json()["items"] == []
# The bot's actions are recorded in the audit log per §6.5.
actions = db.conn().execute(
"SELECT action_kind, on_behalf_of FROM actions ORDER BY id"
).fetchall()
kinds = [(a["action_kind"], a["on_behalf_of"]) for a in actions]
assert ("propose_rfc", "alice") in kinds
assert ("merge_proposal", "ben") in kinds
def test_slug_uniqueness_enforced(app_with_fake_gitea):
from fastapi.testclient import TestClient
app, _fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=5, login="alice", role="contributor")
sign_in_as(client, user_id=5, gitea_login="alice", display_name="Alice", role="contributor")
r = client.post("/api/rfcs/propose", json={
"title": "First", "slug": "first", "pitch": "p", "tags": [],
})
assert r.status_code == 200, r.text
r = client.post("/api/rfcs/propose", json={
"title": "First Again", "slug": "first", "pitch": "p", "tags": [],
})
assert r.status_code == 409
def test_invalid_slug_rejected(app_with_fake_gitea):
from fastapi.testclient import TestClient
app, _fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=7, login="alice", role="contributor")
sign_in_as(client, user_id=7, gitea_login="alice", display_name="Alice", role="contributor")
r = client.post("/api/rfcs/propose", json={
"title": "Bad slug", "slug": "Bad Slug!", "pitch": "p", "tags": [],
})
assert r.status_code == 422
def test_anonymous_cannot_propose(app_with_fake_gitea):
from fastapi.testclient import TestClient
app, _fake = app_with_fake_gitea
with TestClient(app) as client:
r = client.post("/api/rfcs/propose", json={
"title": "A", "slug": "a", "pitch": "p", "tags": [],
})
assert r.status_code == 401
def test_withdraw_by_proposer_works(app_with_fake_gitea):
from fastapi.testclient import TestClient
app, _fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=9, login="alice", role="contributor")
provision_user_row(user_id=10, login="bob", role="contributor")
sign_in_as(client, user_id=9, gitea_login="alice", display_name="Alice", role="contributor")
r = client.post("/api/rfcs/propose", json={
"title": "X", "slug": "x", "pitch": "p", "tags": [],
})
pr_number = r.json()["pr_number"]
# A different contributor cannot withdraw someone else's proposal.
sign_in_as(client, user_id=10, gitea_login="bob", display_name="Bob", role="contributor")
r = client.post(f"/api/proposals/{pr_number}/withdraw")
assert r.status_code == 403
# The proposer can.
sign_in_as(client, user_id=9, gitea_login="alice", display_name="Alice", role="contributor")
r = client.post(f"/api/proposals/{pr_number}/withdraw")
assert r.status_code == 200, r.text
r = client.get("/api/proposals")
assert r.json()["items"] == []