ee4925b6ac
Wave 5 / Track B. Roadmap item #12. Folds in #21 Part C Amplitude
wiring inline (operator ask: "best practices from the very get-go").
RFC owners can invite specific users to one of two per-RFC roles:
contributor (open PRs + join discussion) or discussant (discussion
only). Non-invited users keep the v0.6.0 anonymous-read contract.
The per-RFC write gate layers on top of the existing
require_contributor gate; a super-draft with no owners yet falls
through to the platform-granted contract, preserving the
v0.6.0/v0.7.0/v0.8.0 contracts in their domains.
Backend: migration 018_rfc_invitations.sql (auto-applied — two
tables: rfc_invitations + rfc_collaborators); api_invitations.py
with five endpoints + transactional email; auth.py helpers
(is_rfc_owner / is_rfc_collaborator / can_discuss_rfc /
can_contribute_to_rfc / can_invite_to_rfc); api_discussion +
api_branches + api_prs gate composition; api_admin.py additive
rfc_invitations[] per user. 237 backend tests pass (18 new in
test_rfc_invitations_vertical.py).
Frontend: InvitationsModal.jsx (owner surface), AcceptInvitation.jsx
(/invitations/accept route), api.js helpers, RFCView.jsx
Invitations button, App.jsx route registration.
Amplitude wiring (inline, #21 Part C):
- INVITATION_SENT from InvitationsModal { rfc_slug, role_in_rfc }
- INVITATION_ACCEPTED from AcceptInvitation { rfc_slug, role_in_rfc }
- identify() BEFORE the accept event with properties: invited_at
(setOnce), last_invited_to_rfc, last_invite_role_in_rfc,
claim_method: 'rfc-invite'
- EVENTS taxonomy extended with INVITATION_SENT + INVITATION_ACCEPTED
No new secrets, no new overlay keys, no operator gesture beyond the
v0.15.0 overlay-set + restart. Frontend build verified green.
Subagent ν shipped the feature on feature/v0.16.0-owner-invite
(a51beec). Driver-side integration squash-merged into main,
hand-resolved VERSION + package.json + CHANGELOG (strict-descending
0.16.0 → 0.15.0), and added the inline Amplitude wiring.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
659 lines
26 KiB
Python
659 lines
26 KiB
Python
"""End-to-end integration tests for v0.16.0's owner-only invite for
|
|
per-RFC PR or PR-less discussion (roadmap item #12, §6 / §10).
|
|
|
|
The release lands a per-RFC membership layer:
|
|
|
|
* `rfc_invitations` — issued by the RFC's owner, addressed to an
|
|
email, granting one of two roles ('contributor' or 'discussant').
|
|
* `rfc_collaborators` — the accepted-invitation substrate; the
|
|
table the per-RFC write gate consults.
|
|
|
|
The tests prove:
|
|
|
|
* Only the RFC's owner (or a platform admin/owner) can invite —
|
|
a platform-granted but non-owner user gets 403.
|
|
* Creating an invitation lands a row, mints a token, and queues
|
|
an envelope on the SMTP buffer.
|
|
* Re-inviting the same (email, role) on the same RFC returns 409.
|
|
* The accept endpoint requires the accepting user's email to match
|
|
the invitee_email (case-insensitive).
|
|
* Acceptance lands a rfc_collaborators row and flips the
|
|
invitation to 'accepted'.
|
|
* Re-accepting the same invitation is idempotent (200, changed=false).
|
|
* An expired invitation refuses 409 even if the row's column status
|
|
is still 'pending'.
|
|
* A revoked invitation refuses 409.
|
|
* The owner's listing carries pending + accepted in one response.
|
|
* The per-RFC discussion-write gate refuses a non-invited
|
|
platform-granted user 403 (was previously 200 before v0.16.0).
|
|
* The same gate admits a user who holds an accepted 'discussant'
|
|
invitation.
|
|
* The same gate admits a user who holds an accepted 'contributor'
|
|
invitation (contributor strictly includes discussion).
|
|
* The platform admin/owner is admitted regardless of per-RFC
|
|
membership (the platform-level capability path).
|
|
* The /api/admin/users listing carries `rfc_invitations` per-user
|
|
after an acceptance — the §17 admin surface hook.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# Reuse fixtures and helpers from the propose / RFC-view harnesses.
|
|
from test_propose_vertical import ( # noqa: F401 — fixtures land via import
|
|
FakeGitea,
|
|
app_with_fake_gitea,
|
|
provision_user_row,
|
|
sign_in_as,
|
|
tmp_env,
|
|
)
|
|
from test_rfc_view_vertical import seed_active_rfc, SEED_BODY
|
|
|
|
|
|
def _reset_outbound():
|
|
from app import email as email_mod
|
|
email_mod.reset_sent_envelopes()
|
|
|
|
|
|
def _invitation_envelopes(to_address: str | None = None) -> list[dict]:
|
|
"""Pluck v0.16.0 invitation envelopes out of the shared _SENT buffer.
|
|
Same access pattern as the OTC tests use for `kind='otc'`."""
|
|
from app import email as email_mod
|
|
out = []
|
|
for env in email_mod.sent_envelopes():
|
|
if env.get("kind") != "rfc_invitation":
|
|
continue
|
|
if to_address is not None and env["to"] != to_address:
|
|
continue
|
|
out.append(env)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create / list / revoke (owner-side)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_owner_can_invite_creates_row_and_sends_email(app_with_fake_gitea):
|
|
"""The end-to-end create gesture: RFC owner posts an invitation,
|
|
a row lands, the token comes back in the response, and an
|
|
`rfc_invitation`-kind envelope hits the SMTP buffer."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
# The frontmatter owner of the seeded RFC is "alice" (per
|
|
# seed_active_rfc's default), so we sign in as that user.
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(
|
|
client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor",
|
|
)
|
|
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newperson@example.com", "role_in_rfc": "contributor"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["rfc_slug"] == "ohm"
|
|
assert body["invitee_email"] == "newperson@example.com"
|
|
assert body["role_in_rfc"] == "contributor"
|
|
assert body["status"] == "pending"
|
|
assert body["token"] and len(body["token"]) > 16
|
|
|
|
# Row landed.
|
|
row = db.conn().execute(
|
|
"SELECT * FROM rfc_invitations WHERE id = ?", (body["id"],),
|
|
).fetchone()
|
|
assert row["rfc_slug"] == "ohm"
|
|
assert row["invitee_email"] == "newperson@example.com"
|
|
assert row["inviter_user_id"] == 1
|
|
assert row["status"] == "pending"
|
|
|
|
# Email envelope went out.
|
|
envs = _invitation_envelopes("newperson@example.com")
|
|
assert len(envs) == 1
|
|
assert "OHM" in envs[0]["subject"]
|
|
assert body["token"] in envs[0]["body"]
|
|
|
|
|
|
def test_non_owner_cannot_invite(app_with_fake_gitea):
|
|
"""A platform-granted user who isn't in the RFC's frontmatter
|
|
owners list cannot invite — 403. Distinct from the
|
|
require_contributor gate (which would be 401 for anonymous)."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
# alice is the RFC owner per the seed; bob is a regular
|
|
# platform-granted contributor with no per-RFC role.
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="bob", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(
|
|
client, user_id=2, gitea_login="bob",
|
|
display_name="Bob", role="contributor",
|
|
)
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "ignored@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_platform_admin_can_invite_to_any_rfc(app_with_fake_gitea):
|
|
"""Per §6.1 the platform admin/owner role carries the maximal
|
|
per-RFC capability, so admins can invite on any RFC even if
|
|
they're not in its owners list."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=99, login="adminzero", role="admin")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(
|
|
client, user_id=99, gitea_login="adminzero",
|
|
display_name="Admin Zero", role="admin",
|
|
)
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "another@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
def test_anonymous_cannot_invite(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "x@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_re_invite_same_email_and_role_returns_409(app_with_fake_gitea):
|
|
"""Refuse a duplicate pending invitation for the same (email, role)
|
|
on the same RFC. A different role on the same email is allowed
|
|
(the owner may want to upgrade discussant → contributor)."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(
|
|
client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor",
|
|
)
|
|
|
|
r1 = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "dup@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
assert r1.status_code == 200
|
|
r2 = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "dup@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
assert r2.status_code == 409
|
|
|
|
# Same email, different role is allowed.
|
|
r3 = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "dup@example.com", "role_in_rfc": "contributor"},
|
|
)
|
|
assert r3.status_code == 200
|
|
|
|
|
|
def test_owner_can_list_invitations(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(
|
|
client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor",
|
|
)
|
|
|
|
client.post("/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "a@example.com", "role_in_rfc": "discussant"})
|
|
client.post("/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "b@example.com", "role_in_rfc": "contributor"})
|
|
|
|
r = client.get("/api/rfcs/ohm/invitations")
|
|
assert r.status_code == 200, r.text
|
|
items = r.json()["items"]
|
|
emails = sorted(i["invitee_email"] for i in items)
|
|
assert emails == ["a@example.com", "b@example.com"]
|
|
assert all(i["status"] == "pending" for i in items)
|
|
# The inviter is named.
|
|
assert all(i["inviter_login"] == "alice" for i in items)
|
|
|
|
|
|
def test_revoke_pending_invitation_works_already_accepted_refuses(app_with_fake_gitea):
|
|
"""Revoke flips a pending invitation to 'revoked'. An already-
|
|
accepted invitation refuses 409 — accepted membership is removed
|
|
via a different (future) surface; the v0.16.0 revoke only lifts
|
|
the pending link."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(
|
|
client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor",
|
|
)
|
|
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "revokee@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
invitation_id = r.json()["id"]
|
|
|
|
r = client.post(f"/api/rfcs/ohm/invitations/{invitation_id}/revoke")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "revoked"
|
|
|
|
# Re-revoke refuses 409.
|
|
r2 = client.post(f"/api/rfcs/ohm/invitations/{invitation_id}/revoke")
|
|
assert r2.status_code == 409
|
|
|
|
row = db.conn().execute(
|
|
"SELECT status FROM rfc_invitations WHERE id = ?", (invitation_id,),
|
|
).fetchone()
|
|
assert row["status"] == "revoked"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Accept (invitee-side)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_accept_invitation_lands_collaborator_row(app_with_fake_gitea):
|
|
"""The end-to-end accept gesture: the invitee signs in, posts the
|
|
token, and an rfc_collaborators row lands at the issued role."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
# provision_user_row sets the email to "<login>@test", so the
|
|
# invitee row we'll create needs the same email shape.
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
# alice (owner) invites newbie@test.
|
|
sign_in_as(
|
|
client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor",
|
|
)
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "contributor"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
token = r.json()["token"]
|
|
|
|
# Switch to newbie, accept.
|
|
sign_in_as(
|
|
client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test",
|
|
)
|
|
r = client.post("/api/invitations/accept", json={"token": token})
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["changed"] is True
|
|
assert body["rfc_slug"] == "ohm"
|
|
assert body["role_in_rfc"] == "contributor"
|
|
|
|
# Collaborator row landed; invitation flipped.
|
|
collab = db.conn().execute(
|
|
"SELECT role_in_rfc FROM rfc_collaborators WHERE rfc_slug = 'ohm' AND user_id = 2",
|
|
).fetchone()
|
|
assert collab is not None
|
|
assert collab["role_in_rfc"] == "contributor"
|
|
|
|
inv = db.conn().execute(
|
|
"SELECT status, accepted_by_user_id FROM rfc_invitations WHERE token = ?",
|
|
(token,),
|
|
).fetchone()
|
|
assert inv["status"] == "accepted"
|
|
assert inv["accepted_by_user_id"] == 2
|
|
|
|
|
|
def test_accept_refuses_when_email_does_not_match(app_with_fake_gitea):
|
|
"""The accepting user's email must match the invitation's
|
|
invitee_email (case-insensitive)."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="mallory", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "intended@example.com", "role_in_rfc": "discussant"},
|
|
)
|
|
token = r.json()["token"]
|
|
|
|
# mallory's email is "mallory@test", not "intended@example.com".
|
|
sign_in_as(client, user_id=2, gitea_login="mallory",
|
|
display_name="Mallory", role="contributor",
|
|
email="mallory@test")
|
|
r = client.post("/api/invitations/accept", json={"token": token})
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_accept_refuses_revoked_invitation(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "discussant"},
|
|
)
|
|
invitation_id = r.json()["id"]
|
|
token = r.json()["token"]
|
|
|
|
client.post(f"/api/rfcs/ohm/invitations/{invitation_id}/revoke")
|
|
|
|
sign_in_as(client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test")
|
|
r = client.post("/api/invitations/accept", json={"token": token})
|
|
assert r.status_code == 409
|
|
|
|
|
|
def test_accept_refuses_expired_invitation(app_with_fake_gitea):
|
|
"""An invitation past its `expires_at` is refused 409 even if
|
|
the row's column status is still 'pending'. We backdate the
|
|
expires_at directly to model the elapsed-window state."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "discussant"},
|
|
)
|
|
token = r.json()["token"]
|
|
invitation_id = r.json()["id"]
|
|
|
|
# Backdate.
|
|
db.conn().execute(
|
|
"UPDATE rfc_invitations SET expires_at = datetime('now', '-1 day') WHERE id = ?",
|
|
(invitation_id,),
|
|
)
|
|
|
|
sign_in_as(client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test")
|
|
r = client.post("/api/invitations/accept", json={"token": token})
|
|
assert r.status_code == 409
|
|
|
|
|
|
def test_accept_is_idempotent_on_re_accept(app_with_fake_gitea):
|
|
"""Re-accepting the same already-accepted invitation reads as a
|
|
200 no-op with `changed=false`. The collaborator row is unchanged."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "discussant"},
|
|
)
|
|
token = r.json()["token"]
|
|
|
|
sign_in_as(client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test")
|
|
r1 = client.post("/api/invitations/accept", json={"token": token})
|
|
assert r1.status_code == 200
|
|
assert r1.json()["changed"] is True
|
|
|
|
r2 = client.post("/api/invitations/accept", json={"token": token})
|
|
assert r2.status_code == 200
|
|
assert r2.json()["changed"] is False
|
|
|
|
# Still exactly one collaborator row.
|
|
rows = db.conn().execute(
|
|
"SELECT COUNT(*) AS n FROM rfc_collaborators WHERE rfc_slug = 'ohm' AND user_id = 2"
|
|
).fetchone()
|
|
assert rows["n"] == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Discussion-write gate enforcement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_non_invited_user_cannot_post_to_discussion(app_with_fake_gitea):
|
|
"""v0.16.0 narrows the discussion-write gate: a platform-granted
|
|
user with no per-RFC role gets 403 when posting to the
|
|
discussion. (v0.6.0 left the gate at require_contributor only;
|
|
item #12 layers can_discuss_rfc on top.)"""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="bob", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
# bob is platform-granted but not in OHM's owners list and has
|
|
# no invitation. The thread-create surface refuses 403.
|
|
sign_in_as(client, user_id=2, gitea_login="bob",
|
|
display_name="Bob", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/discussion/threads",
|
|
json={"label": "Question", "message": "Should I be allowed?"},
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_invited_discussant_can_post_to_discussion(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
# alice invites newbie as a discussant.
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "discussant"},
|
|
)
|
|
token = r.json()["token"]
|
|
|
|
# newbie accepts.
|
|
sign_in_as(client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test")
|
|
client.post("/api/invitations/accept", json={"token": token})
|
|
|
|
# newbie can now post to the discussion.
|
|
r = client.post(
|
|
"/api/rfcs/ohm/discussion/threads",
|
|
json={"label": "Question", "message": "Now I can speak."},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
def test_contributor_role_includes_discussion(app_with_fake_gitea):
|
|
"""A 'contributor' per-RFC role strictly includes discussion
|
|
permission — accepting a contributor invitation admits the user
|
|
to the discussion endpoint too."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "contributor"},
|
|
)
|
|
token = r.json()["token"]
|
|
|
|
sign_in_as(client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test")
|
|
client.post("/api/invitations/accept", json={"token": token})
|
|
|
|
r = client.post(
|
|
"/api/rfcs/ohm/discussion/threads",
|
|
json={"label": "Q", "message": "Hello."},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_platform_admin_can_post_to_discussion_without_invitation(app_with_fake_gitea):
|
|
"""Per §6.1 / item #12's permission shape: platform admins/owners
|
|
can write to any RFC's discussion regardless of per-RFC
|
|
membership."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=99, login="adminzero", role="admin")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
sign_in_as(client, user_id=99, gitea_login="adminzero",
|
|
display_name="Admin Zero", role="admin")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/discussion/threads",
|
|
json={"label": "Admin chime", "message": "Drive-by from admin."},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_rfc_owner_can_post_to_discussion(app_with_fake_gitea):
|
|
"""The frontmatter RFC owner is admitted by virtue of being on
|
|
the owners list — they don't need to invite themselves."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/discussion/threads",
|
|
json={"label": "Owner thought", "message": "Kicking off the conversation."},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin-page hook (additive on /api/admin/users)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_admin_users_listing_surfaces_per_rfc_invitations(app_with_fake_gitea):
|
|
"""v0.16.0 hook into the v0.9.0 admin user-management surface:
|
|
each user row carries an `rfc_invitations` array listing the
|
|
per-RFC roles they hold. Empty array for users without any."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
provision_user_row(user_id=2, login="newbie", role="contributor")
|
|
provision_user_row(user_id=99, login="adminzero", role="admin")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
# alice invites newbie; newbie accepts.
|
|
sign_in_as(client, user_id=1, gitea_login="alice",
|
|
display_name="Alice", role="contributor")
|
|
r = client.post(
|
|
"/api/rfcs/ohm/invitations",
|
|
json={"invitee_email": "newbie@test", "role_in_rfc": "contributor"},
|
|
)
|
|
token = r.json()["token"]
|
|
|
|
sign_in_as(client, user_id=2, gitea_login="newbie",
|
|
display_name="Newbie", role="contributor",
|
|
email="newbie@test")
|
|
client.post("/api/invitations/accept", json={"token": token})
|
|
|
|
# Admin lists.
|
|
sign_in_as(client, user_id=99, gitea_login="adminzero",
|
|
display_name="Admin Zero", role="admin")
|
|
r = client.get("/api/admin/users")
|
|
assert r.status_code == 200
|
|
items = r.json()["items"]
|
|
newbie_row = next(i for i in items if i["gitea_login"] == "newbie")
|
|
assert isinstance(newbie_row["rfc_invitations"], list)
|
|
assert len(newbie_row["rfc_invitations"]) == 1
|
|
invite = newbie_row["rfc_invitations"][0]
|
|
assert invite["rfc_slug"] == "ohm"
|
|
assert invite["role_in_rfc"] == "contributor"
|
|
assert invite["inviter_login"] == "alice"
|
|
|
|
# Users with no invitations carry an empty array, not null.
|
|
alice_row = next(i for i in items if i["gitea_login"] == "alice")
|
|
assert alice_row["rfc_invitations"] == []
|