rfc-app/backend/tests/test_pr_flow_vertical.py

"""End-to-end integration tests for the Slice 3 vertical (§10 in full).

Reuses the FakeGitea + session helpers from test_propose_vertical.py
and the active-RFC seed from test_rfc_view_vertical.py. Walks the §10
vertical end-to-end against an in-process fake Gitea:

  * open-pr from a non-main branch with a pending §11.3 visibility flip
  * the AI-drafted title/description from `pr-draft`
  * the §10.3 review payload: three-column data, threads, seen cursor
  * §10.4 review-kind thread posting
  * §10.5 no-fast-forward merge by an arbiter
  * §10.5 merge by a non-arbiter is refused
  * §10.8 withdraw by the contributor and by an arbiter
  * §10.9 conflict-replay: original PR auto-closes when the resolution
    PR merges
"""
from __future__ import annotations

import pytest

from test_propose_vertical import (  # noqa: F401
    FakeGitea,
    app_with_fake_gitea,
    provision_user_row,
    sign_in_as,
    tmp_env,
)
from test_rfc_view_vertical import SEED_BODY, seed_active_rfc


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _cut_branch_and_accept_change(client, fake, *, slug: str, original: str, proposed: str):
    """Cut a branch and produce one accepted AI change on it.

    Returns the branch name and the change row id.
    """
    from app import db

    r = client.post(f"/api/rfcs/{slug}/branches/main/promote-to-branch", json={})
    assert r.status_code == 200, r.text
    branch = r.json()["branch_name"]
    view = client.get(f"/api/rfcs/{slug}/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 (?, ?, ?, 'ai', 'pending', ?, ?, 'test')
        """,
        (slug, branch, thread_id, original, proposed),
    )
    change_id = cur.lastrowid
    r = client.post(
        f"/api/rfcs/{slug}/branches/{branch}/changes/{change_id}/accept",
        json={"proposed": proposed, "was_edited_before_accept": False},
    )
    assert r.status_code == 200, r.text
    return branch, change_id


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------


def test_open_pr_creates_pr_and_flips_branch_public(app_with_fake_gitea):
    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=2, login="alice", role="contributor")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")

        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )

        # Flip branch private to exercise the §11.3 universal-public flip.
        r = client.post(f"/api/rfcs/ohm/branches/{branch}/visibility", json={"read_public": False})
        assert r.status_code == 200, r.text

        r = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten the opening", "description": "Scope to systems."},
        )
        assert r.status_code == 200, r.text
        pr_number = r.json()["pr_number"]
        assert pr_number > 0

        # Branch is now public.
        vis = db.conn().execute(
            "SELECT read_public FROM branch_visibility WHERE rfc_slug = 'ohm' AND branch_name = ?",
            (branch,),
        ).fetchone()
        assert vis["read_public"] == 1

        # Second open-pr on the same branch fails per §10.9.
        r = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Again", "description": "x"},
        )
        assert r.status_code == 409


def test_pr_draft_returns_title_and_description(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=2, login="alice", role="contributor")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="It defines consent, trait, and agency in compatible terms.",
            proposed="It defines consent, trait, harm, and agency in compatible terms.",
        )
        r = client.post(f"/api/rfcs/ohm/branches/{branch}/pr-draft")
        assert r.status_code == 200, r.text
        data = r.json()
        # The stub draft is sufficient when no provider is configured.
        assert "title" in data and data["title"]
        assert "description" in data and data["description"]


def test_get_pr_returns_three_column_payload(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=2, login="alice", role="contributor")
        provision_user_row(user_id=3, login="bob", role="contributor")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        # Bob is the non-arbiter contributor — alice is seeded as an RFC owner.
        sign_in_as(client, user_id=3, gitea_login="bob", display_name="Bob", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )
        r = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten", "description": "Scope."},
        )
        pr_number = r.json()["pr_number"]

        r = client.get(f"/api/rfcs/ohm/prs/{pr_number}")
        assert r.status_code == 200, r.text
        d = r.json()
        assert d["pr_number"] == pr_number
        assert d["title"] == "Tighten"
        assert d["head_branch"] == branch
        assert d["state"] == "open"
        assert "main_body" in d and "branch_body" in d
        # The branch body has the accepted edit; main does not.
        assert "across systems" in d["branch_body"]
        assert "across systems" not in d["main_body"]
        # Mergeable when nothing else has moved on main.
        assert d["mergeable"] is True
        # Aggregate counts are present.
        assert "counts" in d
        assert d["counts"]["open_review_threads"] == 0
        # Capabilities surface — bob is not arbiter/admin/owner, can't merge.
        assert d["capabilities"]["can_merge"] is False
        # Bob opened the PR; he can withdraw.
        assert d["capabilities"]["can_withdraw"] is True


def test_pr_seen_cursor_advances(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=2, login="alice", role="contributor")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )
        pr_number = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten", "description": "Scope."},
        ).json()["pr_number"]

        # First read: no cursor.
        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        assert d["seen"] is None

        # Drop two messages into the branch chat so the seen cursor has
        # FK-valid ids to point at.
        from app import db
        view = client.get(f"/api/rfcs/ohm/branches/{branch}").json()
        thread_id = view["main_thread_id"]
        cur = db.conn().execute(
            "INSERT INTO thread_messages (thread_id, role, text) VALUES (?, 'system', 'first')",
            (thread_id,),
        )
        first_msg = cur.lastrowid
        cur = db.conn().execute(
            "INSERT INTO thread_messages (thread_id, role, text) VALUES (?, 'system', 'second')",
            (thread_id,),
        )
        second_msg = cur.lastrowid

        # Advance to the second message.
        r = client.post(
            f"/api/rfcs/ohm/prs/{pr_number}/seen",
            json={"last_seen_commit_sha": "sha9999", "last_seen_message_id": second_msg},
        )
        assert r.status_code == 200

        # Re-read: cursor reflects the advance.
        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        assert d["seen"]["last_seen_commit_sha"] == "sha9999"
        assert d["seen"]["last_seen_message_id"] == second_msg

        # A stale advance can't roll the cursor backward.
        client.post(
            f"/api/rfcs/ohm/prs/{pr_number}/seen",
            json={"last_seen_message_id": first_msg},
        )
        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        assert d["seen"]["last_seen_message_id"] == second_msg


def test_review_thread_lands_as_review_kind(app_with_fake_gitea):
    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=2, login="alice", role="contributor")
        provision_user_row(user_id=1, login="ben", role="owner")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )
        pr_number = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten", "description": "Scope."},
        ).json()["pr_number"]

        # Ben (the arbiter) leaves a review comment.
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        r = client.post(
            f"/api/rfcs/ohm/prs/{pr_number}/review",
            json={
                "text": "Should we say 'in software systems' instead?",
                "anchor_payload": {"from": 10, "to": 50},
                "quote": "across systems",
            },
        )
        assert r.status_code == 200, r.text
        thread_id = r.json()["thread_id"]

        # The thread persists as thread_kind='review', anchor_kind='range'.
        row = db.conn().execute(
            "SELECT thread_kind, anchor_kind, branch_name FROM threads WHERE id = ?",
            (thread_id,),
        ).fetchone()
        assert row["thread_kind"] == "review"
        assert row["anchor_kind"] == "range"
        assert row["branch_name"] == branch

        # The PR payload surfaces the review thread inline.
        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        review_threads = [t for t in d["threads"] if t["thread_kind"] == "review"]
        assert len(review_threads) == 1
        assert d["counts"]["open_review_threads"] == 1


def test_merge_by_arbiter_advances_main_and_marks_pr_merged(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=2, login="alice", role="contributor")
        provision_user_row(user_id=3, login="bob", role="contributor")
        provision_user_row(user_id=1, login="ben", role="owner")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        # Bob is neither owner nor arbiter — the non-merge baseline.
        sign_in_as(client, user_id=3, gitea_login="bob", display_name="Bob", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )
        pr_number = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten", "description": "Scope."},
        ).json()["pr_number"]

        # Bob is a plain contributor — refused per §6.3.
        r = client.post(f"/api/rfcs/ohm/prs/{pr_number}/merge")
        assert r.status_code == 403

        # Ben is the app owner and the RFC arbiter — can merge.
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        r = client.post(f"/api/rfcs/ohm/prs/{pr_number}/merge")
        assert r.status_code == 200, r.text

        # main now carries the accepted text.
        body = fake.files[("wiggleverse", "rfc-0001-ohm", "main", "RFC.md")]["content"]
        assert "across systems" in body

        # PR is reported as merged + post-merge fields surface.
        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        assert d["state"] == "merged"
        assert d["capabilities"]["can_merge"] is False  # already merged
        # The merge produced a fresh sha on main per §10.5's no-ff.
        assert d["merge_commit_sha"]


def test_withdraw_by_contributor_marks_state_withdrawn(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=2, login="alice", role="contributor")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )
        pr_number = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten", "description": "Scope."},
        ).json()["pr_number"]

        r = client.post(f"/api/rfcs/ohm/prs/{pr_number}/withdraw")
        assert r.status_code == 200, r.text

        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        assert d["state"] == "withdrawn"
        # Withdrawn PRs are read-only; no merge button.
        assert d["capabilities"]["can_merge"] is False
        assert d["capabilities"]["can_withdraw"] is False


def test_resolution_branch_replays_clean_and_supersedes_on_merge(app_with_fake_gitea):
    """Per §10.9: a conflicting PR can be resolved by cutting a fresh
    branch off main's tip and replaying. The original auto-closes when
    the resolution PR merges."""
    from fastapi.testclient import TestClient

    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=2, login="alice", role="contributor")
        provision_user_row(user_id=3, login="bob", role="contributor")
        provision_user_row(user_id=1, login="ben", role="owner")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)

        # Alice cuts a branch and accepts a change on it.
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        alice_branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="It defines consent, trait, and agency in compatible terms.",
            proposed="It defines consent, trait, harm, and agency in compatible terms.",
        )
        alice_pr_number = client.post(
            f"/api/rfcs/ohm/branches/{alice_branch}/open-pr",
            json={"title": "Add harm", "description": "Adds harm to the dimension list."},
        ).json()["pr_number"]

        # Bob lands a different change to the same paragraph on main
        # by opening + merging his own PR. That moves main and makes
        # Alice's PR unmergeable.
        sign_in_as(client, user_id=3, gitea_login="bob", display_name="Bob", role="contributor")
        bob_branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="It defines consent, trait, and agency in compatible terms.",
            proposed="It defines consent, agency, and trait in compatible terms.",
        )
        bob_pr_number = client.post(
            f"/api/rfcs/ohm/branches/{bob_branch}/open-pr",
            json={"title": "Reorder", "description": "Reorders the dimensions."},
        ).json()["pr_number"]
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        r = client.post(f"/api/rfcs/ohm/prs/{bob_pr_number}/merge")
        assert r.status_code == 200, r.text

        # Alice's PR is now unmergeable.
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        d = client.get(f"/api/rfcs/ohm/prs/{alice_pr_number}").json()
        assert d["mergeable"] is False

        # Direct merge attempt is refused with 409.
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        r = client.post(f"/api/rfcs/ohm/prs/{alice_pr_number}/merge")
        assert r.status_code == 409

        # Alice starts a resolution branch.
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        r = client.post(f"/api/rfcs/ohm/prs/{alice_pr_number}/resolution-branch")
        assert r.status_code == 200, r.text
        resolution_branch = r.json()["resolution_branch"]
        # Alice's `<original>` text no longer matches main; the replay
        # surfaces as ambiguous so she can re-anchor.
        assert r.json()["replayed_ambiguous"] >= 1

        # Resolve the ambiguous change manually by opening a fresh
        # accept on the resolution branch — substituting the now-correct
        # `original` from main.
        from app import db
        ambiguous_change = db.conn().execute(
            """
            SELECT id FROM changes WHERE rfc_slug = 'ohm' AND branch_name = ?
              AND state = 'pending' AND stale_since IS NOT NULL
             ORDER BY id LIMIT 1
            """,
            (resolution_branch,),
        ).fetchone()
        assert ambiguous_change is not None

        # Re-anchor by inserting a fresh AI-pending row anchored to
        # main's text, then accepting it.
        view = client.get(f"/api/rfcs/ohm/branches/{resolution_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', ?, ?, 'add harm')
            """,
            (resolution_branch, thread_id,
             "It defines consent, agency, and trait in compatible terms.",
             "It defines consent, agency, trait, and harm in compatible terms."),
        )
        replay_change_id = cur.lastrowid
        r = client.post(
            f"/api/rfcs/ohm/branches/{resolution_branch}/changes/{replay_change_id}/accept",
            json={
                "proposed": "It defines consent, agency, trait, and harm in compatible terms.",
                "was_edited_before_accept": False,
            },
        )
        assert r.status_code == 200, r.text

        # Open the resolution PR.
        r = client.post(
            f"/api/rfcs/ohm/branches/{resolution_branch}/open-pr",
            json={"title": "Add harm (rebased)", "description": "Rebased on bob's reorder."},
        )
        assert r.status_code == 200, r.text
        resolution_pr_number = r.json()["pr_number"]

        # The supersession is recorded — both directions visible.
        d = client.get(f"/api/rfcs/ohm/prs/{resolution_pr_number}").json()
        assert d["supersedes_pr_number"] == alice_pr_number

        # Arbiter merges the resolution PR.
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        r = client.post(f"/api/rfcs/ohm/prs/{resolution_pr_number}/merge")
        assert r.status_code == 200, r.text

        # Original PR auto-closes via the Supersedes: trailer.
        d_orig = client.get(f"/api/rfcs/ohm/prs/{alice_pr_number}").json()
        assert d_orig["state"] == "closed"
        assert d_orig["superseded_by_pr_number"] == resolution_pr_number


def test_anonymous_can_read_pr_but_not_post(app_with_fake_gitea):
    """§11.3: PRs are always public; anonymous viewers read but cannot
    advance the seen cursor or post review comments."""
    from fastapi.testclient import TestClient

    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=2, login="alice", role="contributor")
        seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
        sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
        branch, _ = _cut_branch_and_accept_change(
            client, fake, slug="ohm",
            original="Open Human Model is a framework for representing humans.",
            proposed="Open Human Model is a framework for representing humans across systems.",
        )
        pr_number = client.post(
            f"/api/rfcs/ohm/branches/{branch}/open-pr",
            json={"title": "Tighten", "description": "Scope."},
        ).json()["pr_number"]

        # Drop the session.
        client.cookies.clear()
        d = client.get(f"/api/rfcs/ohm/prs/{pr_number}").json()
        assert d["state"] == "open"
        assert d["capabilities"]["is_anonymous"] is True

        # Anonymous post is 401.
        r = client.post(
            f"/api/rfcs/ohm/prs/{pr_number}/review",
            json={"text": "x", "anchor_payload": {}, "quote": None},
        )
        assert r.status_code == 401