Files
rfc-app/backend/tests/test_graduation_vertical.py
T
Ben Stull 0c972c8af5 v0.31.0: meta-only repository topology (ROADMAP #36)
Retire the per-RFC-repo model. RFCs now live in their meta-repo entry
(rfcs/<slug>.md) for their whole life; graduation is an in-place
super-draft → active state flip that keeps the body in the entry — no
repo creation, no body-strip, no five-step transaction, no rollback.

SPEC: §1 topology rewritten (one meta/content repository, no per-RFC
repos) with a deployer-facing "single content repository" framing; §2
(repo: always-null), §3 (active is in-place), §4, §9.8 (handoff
frictions dissolve), and §13 fully rewritten (two-field dialog, "The
flip", §13.6 RFC-0001 fold-back record).

Code: graduation collapses to open+merge one frontmatter PR; branch/PR/
chat dispatch re-keyed on meta-residency (repo IS NULL) so active RFCs
edit on the meta repo exactly as super-drafts do; the two "RFC has no
repo" 409 guards removed; promote-to-branch slug-embeds an active RFC's
auto-branch (edit-<slug>-<hex>) for shared-repo cache attribution;
refresh_meta_branches + hygiene branch-resolution include meta-resident
actives; the §9.8 read-only guard + pre_graduation_history scoped to
legacy per-repo only; dead bot primitives + GraduateDialog repo field +
blocking-PR popover removed. The repo: frontmatter field and the
/blocking-prs endpoint are retained (schema stability / informational).

Tests: graduation suite rewritten to the flip model; e2e + hygiene
updated. Full backend suite 375 passed; frontend builds.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 07:12:06 -07:00

546 lines
22 KiB
Python

"""End-to-end integration tests for the §13 graduation flow under the
meta-only topology (SPEC §1, ROADMAP #36).
Graduation is an in-place state flip on the meta entry — no per-RFC repo
is created, the body is kept, and there is no multi-step transaction or
rollback (§13.3). These tests walk it against the in-process FakeGitea
from test_propose_vertical.py:
* Seed an owned super-draft (the §13.1 claim flow is exercised
separately in test_claim_opens_meta_pr).
* GET /api/rfcs/<slug>/graduate/check returns per-field validity for
the two-field dialog (integer id + owners; no repo name).
* POST /api/rfcs/<slug>/graduate?_sync=1 opens + merges the flip PR
inline. On success: NO per-RFC repo, the meta entry is `state:
active` with the body KEPT and `repo` null, cached_rfcs.state flips
to 'active'.
* An open body-edit PR no longer blocks graduation (§9.8) — they
coexist.
* An open-PR failure leaves the entry a super-draft (nothing created);
a merge failure cleans up the half-open PR/branch and leaves the
entry a super-draft.
* §13.4: chat threads + edit branches stay put — the slug is the
canonical key per §2.3, so nothing moves at the flip.
The orchestrator's `?_sync=1` seam awaits the flip inline so the test can
assert post-conditions on the same event loop tick.
"""
from __future__ import annotations
import json as _json
import pytest
from test_propose_vertical import ( # noqa: F401
FakeGitea,
app_with_fake_gitea,
grant_rfc_collaborator,
provision_user_row,
sign_in_as,
tmp_env,
)
from test_super_draft_vertical import seed_super_draft # noqa: F401
PITCH = (
"Open Human Model is a framework for representing humans.\n\n"
"It defines consent, trait, and agency in compatible terms."
)
def seed_owned_super_draft(fake: FakeGitea, *, slug: str, title: str, pitch: str,
owners: list[str], arbiters: list[str] | None = None,
proposed_by: str = "alice", tags: list[str] | None = None) -> None:
"""Seed a super-draft directly with owners already filled in — the
§13.1 claim flow is exercised separately."""
import yaml
from app import db
fm = {
"slug": slug,
"title": title,
"state": "super-draft",
"id": None,
"repo": None,
"proposed_by": proposed_by,
"proposed_at": "2026-05-23",
"graduated_at": None,
"graduated_by": None,
"owners": owners,
"arbiters": arbiters or owners[:1],
"tags": tags or [],
}
body = pitch.strip() + "\n"
entry_text = f"---\n{yaml.safe_dump(fm, sort_keys=False).rstrip()}\n---\n\n{body}"
sha = fake._next_sha()
fake.files[("wiggleverse", "meta", "main", f"rfcs/{slug}.md")] = {
"content": entry_text, "sha": sha,
}
fake.branches[("wiggleverse", "meta")]["main"]["sha"] = sha
db.conn().execute(
"""
INSERT OR REPLACE INTO cached_rfcs
(slug, title, state, rfc_id, repo, proposed_by, proposed_at,
owners_json, arbiters_json, tags_json,
body, body_sha, last_main_commit_at, last_entry_commit_at)
VALUES (?, ?, 'super-draft', NULL, NULL, ?, '2026-05-23',
?, ?, ?, ?, ?, datetime('now'), datetime('now'))
""",
(
slug, title, proposed_by,
_json.dumps(owners),
_json.dumps(arbiters or owners[:1]),
_json.dumps(tags or []),
body, sha,
),
)
db.conn().execute(
"""
INSERT OR REPLACE INTO cached_branches
(rfc_slug, branch_name, head_sha, state, last_commit_at)
VALUES (?, 'main', ?, 'open', datetime('now'))
""",
(slug, sha),
)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_graduate_check_validates_id_and_owners(app_with_fake_gitea):
"""Two-field dialog under meta-only: integer id + owners. No repo
name to validate (§13.2)."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="Open Human Model",
pitch=PITCH, owners=["ben"])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
# Happy: a fresh RFC-0001.
r = client.get("/api/rfcs/ohm/graduate/check", params={"id": "RFC-0001"})
assert r.status_code == 200, r.text
d = r.json()
assert d["id"]["ok"] is True
assert d["owners"]["ok"] is True
assert d["can_submit"] is True
# No repo field in the meta-only check response.
assert "repo" not in d
# ID format error — non-numeric tail.
r = client.get("/api/rfcs/ohm/graduate/check", params={"id": "RFC-abcd"})
d = r.json()
assert d["id"]["ok"] is False
assert d["can_submit"] is False
def test_graduate_check_refuses_when_no_owners(app_with_fake_gitea):
"""An unclaimed super-draft fails the owners precondition; can_submit
flips false even with a valid id."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="OHM", pitch=PITCH, owners=[])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
r = client.get("/api/rfcs/ohm/graduate/check", params={"id": "RFC-0001"})
d = r.json()
assert d["owners"]["ok"] is False
assert "No owners" in d["owners"]["error"]
assert d["can_submit"] is False
def test_graduate_happy_path_flips_in_place_keeping_body(app_with_fake_gitea):
"""The meta-only flip: open + merge a frontmatter PR. End state:
cached_rfcs.state='active', the meta entry's body is KEPT, `repo` is
null, NO per-RFC repo exists, and the audit log carries graduate_start
→ graduate_pr_open → graduate_pr_merge → graduate_complete."""
from fastapi.testclient import TestClient
from app import db, entry as entry_mod
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="Open Human Model",
pitch=PITCH, owners=["ben"], arbiters=["ben"],
tags=["identity", "schema"])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner", email="ben@test")
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0042", "owners": ["ben"]},
)
assert r.status_code == 200, r.text
d = r.json()
assert d["finished"] is True
assert d["succeeded"] is True
# No repo in the response, no per-RFC repo on Gitea.
assert "repo" not in d
assert ("wiggleverse", "rfc-0042-ohm") not in fake.repos
assert not any(
k[1].startswith("rfc-0042") for k in fake.repos
), f"a per-RFC repo was created: {fake.repos}"
# Meta entry on main: state flipped, body KEPT, repo null.
meta_text = fake.files[("wiggleverse", "meta", "main", "rfcs/ohm.md")]["content"]
graduated = entry_mod.parse(meta_text)
assert graduated.state == "active"
assert graduated.id == "RFC-0042"
assert graduated.repo is None
assert graduated.graduated_by == "ben"
assert graduated.graduated_at # non-empty ISO date
assert "Open Human Model is a framework" in graduated.body
# cached_rfcs flipped to active via the inline refresh; body intact.
cached = db.conn().execute(
"SELECT state, rfc_id, repo, body FROM cached_rfcs WHERE slug = 'ohm'"
).fetchone()
assert cached["state"] == "active"
assert cached["rfc_id"] == "RFC-0042"
assert cached["repo"] is None
assert "Open Human Model is a framework" in cached["body"]
kinds = [
row["action_kind"]
for row in db.conn().execute(
"SELECT action_kind FROM actions WHERE rfc_slug = 'ohm' ORDER BY id"
)
]
for needed in ("graduate_start", "graduate_pr_open",
"graduate_pr_merge", "graduate_complete"):
assert needed in kinds, f"missing audit row {needed}: {kinds}"
# The retired per-repo steps must NOT appear.
for gone in ("graduate_repo_create", "graduate_repo_seed",
"graduate_repo_delete", "graduate_rollback"):
assert gone not in kinds, f"retired audit row present: {gone}"
def test_graduate_coexists_with_open_body_edit_pr(app_with_fake_gitea):
"""§9.8 (meta-only): an open meta-repo body-edit PR no longer blocks
graduation — the body is kept, so they coexist. /check stays
submittable and the flip succeeds."""
from fastapi.testclient import TestClient
from app import db
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
provision_user_row(user_id=2, login="alice", role="contributor")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=["ben"])
grant_rfc_collaborator(user_id=2, rfc_slug="ohm", role_in_rfc="contributor")
sign_in_as(client, user_id=2, gitea_login="alice",
display_name="Alice", role="contributor")
# Cut an edit branch and open a body-edit PR.
branch = client.post("/api/rfcs/ohm/start-edit-branch", json={}).json()["branch_name"]
view = client.get(f"/api/rfcs/ohm/branches/{branch}").json()
thread_id = view["main_thread_id"]
cur = db.conn().execute(
"""
INSERT INTO changes
(rfc_slug, branch_name, thread_id, kind, state, original, proposed, reason)
VALUES ('ohm', ?, ?, 'ai', 'pending', ?, ?, 'tighten')
""",
(branch, thread_id,
"It defines consent, trait, and agency in compatible terms.",
"It defines consent, trait, harm, and agency in compatible terms."),
)
change_id = cur.lastrowid
client.post(
f"/api/rfcs/ohm/branches/{branch}/changes/{change_id}/accept",
json={"proposed": "It defines consent, trait, harm, and agency in compatible terms.",
"was_edited_before_accept": False},
)
pr_number = client.post(
f"/api/rfcs/ohm/branches/{branch}/open-pr",
json={"title": "Add harm", "description": "Adds harm dimension."},
).json()["pr_number"]
assert pr_number # PR is open
# /check stays submittable despite the open body-edit PR.
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
d = client.get("/api/rfcs/ohm/graduate/check", params={"id": "RFC-0001"}).json()
assert "blocking_prs" not in d
assert d["can_submit"] is True
# The flip succeeds — coexists with the open body-edit PR.
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0001", "owners": ["ben"]},
)
assert r.status_code == 200, r.text
assert r.json()["succeeded"] is True
cached = db.conn().execute(
"SELECT state FROM cached_rfcs WHERE slug = 'ohm'"
).fetchone()
assert cached["state"] == "active"
def test_graduate_open_pr_failure_leaves_super_draft(app_with_fake_gitea):
"""An open-PR failure creates nothing — the entry stays a super-draft
with its body intact and no graduation PR on the meta repo."""
from fastapi.testclient import TestClient
from app import db
from app.bot import Bot
from app.gitea import GiteaError
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=["ben"])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
orig_open_pr = Bot.open_graduation_pr
async def boom(self, *args, **kwargs):
raise GiteaError(502, "simulated PR-open failure")
Bot.open_graduation_pr = boom
try:
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0007", "owners": ["ben"]},
)
finally:
Bot.open_graduation_pr = orig_open_pr
assert r.status_code == 200, r.text
assert r.json()["succeeded"] is False
# Entry untouched: still super-draft, body intact on main.
cached = db.conn().execute(
"SELECT state, rfc_id FROM cached_rfcs WHERE slug = 'ohm'"
).fetchone()
assert cached["state"] == "super-draft"
assert cached["rfc_id"] is None
meta_text = fake.files[("wiggleverse", "meta", "main", "rfcs/ohm.md")]["content"]
assert "Open Human Model is a framework" in meta_text
kinds = [
row["action_kind"]
for row in db.conn().execute(
"SELECT action_kind FROM actions WHERE rfc_slug = 'ohm' ORDER BY id"
)
]
assert "graduate_start" in kinds
assert "graduate_failed" in kinds
assert "graduate_complete" not in kinds
def test_graduate_merge_failure_cleans_up_pr(app_with_fake_gitea):
"""A merge failure leaves the flip PR open on its dash-suffixed
branch; the orchestrator closes the PR and deletes the branch so
failed attempts don't accumulate. The entry stays a super-draft —
the flip PR's commit was on a branch, not on main."""
from fastapi.testclient import TestClient
from app import db
from app.bot import Bot
from app.gitea import GiteaError
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=["ben"])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
orig_merge = Bot.merge_graduation_pr
async def boom(self, *args, **kwargs):
raise GiteaError(502, "simulated merge failure")
Bot.merge_graduation_pr = boom
try:
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0009", "owners": ["ben"]},
)
finally:
Bot.merge_graduation_pr = orig_merge
assert r.status_code == 200, r.text
assert r.json()["succeeded"] is False
# Entry stays super-draft on main (the flip never merged).
cached = db.conn().execute(
"SELECT state FROM cached_rfcs WHERE slug = 'ohm'"
).fetchone()
assert cached["state"] == "super-draft"
meta_text = fake.files[("wiggleverse", "meta", "main", "rfcs/ohm.md")]["content"]
assert "state: super-draft" in meta_text
# The dash-suffixed graduation branch was cleaned up.
grad_branches = [
name for (o, repo), branches in fake.branches.items()
if (o, repo) == ("wiggleverse", "meta")
for name in branches
if name.startswith("graduate-ohm-")
]
assert grad_branches == [], f"leftover graduation branch: {grad_branches}"
kinds = [
row["action_kind"]
for row in db.conn().execute(
"SELECT action_kind FROM actions WHERE rfc_slug = 'ohm' ORDER BY id"
)
]
assert "graduate_pr_open" in kinds
assert "graduate_failed" in kinds
assert "graduate_complete" not in kinds
def test_graduate_refuses_concurrent_graduation(app_with_fake_gitea):
"""A second graduation request for a slug already in-flight is refused."""
from fastapi.testclient import TestClient
from app import api_graduation
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=["ben"])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
st = api_graduation._new_active(
"ohm", rfc_id="RFC-0001", owners=["ben"], arbiters=["ben"],
)
st.finished = False
try:
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0001", "owners": ["ben"]},
)
assert r.status_code == 409
finally:
api_graduation._active.pop("ohm", None)
def test_chat_threads_survive_graduation_without_data_movement(app_with_fake_gitea):
"""§13.4: chat threads on the entry's main view (`branch_name='main'`)
stay put across the flip — the rfc_slug is canonical per §2.3 — so the
same thread surfaces from /branches/main before and after graduation."""
from fastapi.testclient import TestClient
from app import db
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=["ben"])
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
cur = db.conn().execute(
"""
INSERT INTO threads (rfc_slug, branch_name, anchor_kind, thread_kind, created_by)
VALUES ('ohm', 'main', 'whole-doc', 'chat', 1)
"""
)
thread_id = cur.lastrowid
db.conn().execute(
"""
INSERT INTO thread_messages (thread_id, role, author_user_id, text)
VALUES (?, 'user', 1, 'pre-grad note on the canonical body')
""",
(thread_id,),
)
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0099", "owners": ["ben"]},
)
assert r.status_code == 200, r.text
row = db.conn().execute(
"SELECT id, branch_name FROM threads WHERE id = ?", (thread_id,),
).fetchone()
assert row["branch_name"] == "main"
r = client.get("/api/rfcs/ohm/branches/main")
assert r.status_code == 200, r.text
assert r.json()["main_thread_id"] == thread_id
def test_edit_branch_surfaces_normally_after_graduation(app_with_fake_gitea):
"""§13.4 (meta-only): after graduation an edit branch is a *current*
branch of the now-active RFC — it surfaces in the normal `branches`
list, and there is no separate `pre_graduation_history` set (that
affordance is legacy per-repo only)."""
from fastapi.testclient import TestClient
from app import db
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
provision_user_row(user_id=2, login="alice", role="contributor")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=["ben"])
grant_rfc_collaborator(user_id=2, rfc_slug="ohm", role_in_rfc="contributor")
sign_in_as(client, user_id=2, gitea_login="alice",
display_name="Alice", role="contributor")
branch = client.post("/api/rfcs/ohm/start-edit-branch", json={}).json()["branch_name"]
view = client.get(f"/api/rfcs/ohm/branches/{branch}").json()
thread_id = view["main_thread_id"]
db.conn().execute(
"""
INSERT INTO thread_messages (thread_id, role, author_user_id, text)
VALUES (?, 'user', 2, 'note on an edit branch')
""",
(thread_id,),
)
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner")
r = client.post(
"/api/rfcs/ohm/graduate?_sync=1",
json={"rfc_id": "RFC-0100", "owners": ["ben"]},
)
assert r.status_code == 200, r.text
d = client.get("/api/rfcs/ohm/main").json()
assert d["state"] == "active"
# The edit branch is a current branch; no pre-graduation hop.
assert d["pre_graduation_history"] == []
assert any(b["name"] == branch for b in d["branches"]), \
f"edit branch not in branches: {[b['name'] for b in d['branches']]}"
def test_claim_opens_meta_pr(app_with_fake_gitea):
"""§13.1: any signed-in contributor can claim ownership of an
unclaimed super-draft; the result is a meta-repo PR
(`pr_kind='meta_claim'`) adding their gitea_login to the entry's
owners list."""
from fastapi.testclient import TestClient
from app import db, entry as entry_mod
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=2, login="alice", role="contributor")
seed_owned_super_draft(fake, slug="ohm", title="OHM",
pitch=PITCH, owners=[])
sign_in_as(client, user_id=2, gitea_login="alice",
display_name="Alice", role="contributor")
r = client.post("/api/rfcs/ohm/claim")
assert r.status_code == 200, r.text
d = r.json()
assert d["branch_name"] == "claim/ohm"
text = fake.files[("wiggleverse", "meta", "claim/ohm", "rfcs/ohm.md")]["content"]
ent = entry_mod.parse(text)
assert "alice" in ent.owners
row = db.conn().execute(
"SELECT pr_kind FROM cached_prs WHERE pr_number = ?", (d["pr_number"],),
).fetchone()
assert row["pr_kind"] == "meta_claim"