Add Groups API router: create/join/leave, roles, transfer, group page
Implements docs/planning/social.md US-5's API layer on top of the
already-merged Group/GroupMember models (app/models/groups.py):
- POST/GET /api/groups, GET/PATCH/DELETE /api/groups/{id}
- POST .../join (idempotent), .../leave (owner must transfer first)
- POST .../members/{user_id}/role (owner-only, admin/member),
.../transfer (owner-only, demotes prior owner to admin)
Group page member list is a dedicated schema (user_id/username/role/
joined_at) so no private User fields leak. 27 new tests cover
authorization boundaries, idempotency, and exactly-one-owner invariants;
full backend suite (220 tests) passes with zero regressions.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QFSPt6rX1gwZmcr2HEfvzg
This commit is contained in:
+2
-1
@@ -6,7 +6,7 @@ from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.config import settings
|
||||
from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence
|
||||
from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence, groups
|
||||
|
||||
|
||||
from app.services.game_servers import supervisor
|
||||
@@ -45,6 +45,7 @@ app.include_router(assets.router, prefix="/api/assets", tags=["assets"])
|
||||
app.include_router(avatar.router, prefix="/api/avatar", tags=["avatar"])
|
||||
app.include_router(internal.router, prefix="/api/internal", tags=["internal"])
|
||||
app.include_router(presence.router, prefix="/api/presence", tags=["presence"])
|
||||
app.include_router(groups.router, prefix="/api/groups", tags=["groups"])
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
|
||||
@@ -0,0 +1,465 @@
|
||||
"""Groups API router.
|
||||
|
||||
Endpoints:
|
||||
- POST /api/groups -- create (creator becomes owner)
|
||||
- GET /api/groups -- list groups (paginated), optional mine=true
|
||||
- GET /api/groups/{id} -- group page (name/description/members)
|
||||
- PATCH /api/groups/{id} -- update name/description (owner/admin)
|
||||
- DELETE /api/groups/{id} -- delete (owner only)
|
||||
- POST /api/groups/{id}/join -- join (idempotent)
|
||||
- POST /api/groups/{id}/leave -- leave (owner must transfer first)
|
||||
- POST /api/groups/{id}/members/{user_id}/role -- set role admin/member (owner only)
|
||||
- POST /api/groups/{id}/transfer -- transfer ownership (owner only)
|
||||
|
||||
See app/models/groups.py for the schema and delete-behavior rationale
|
||||
(docs/planning/social.md US-5). This module owns create/join/leave, role
|
||||
authorization, and the group page aggregate.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.database import get_db
|
||||
from app.dependencies import CurrentUser
|
||||
from app.models.groups import Group, GroupMember, GroupRole
|
||||
from app.models.user import User
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
DbDep = Annotated[AsyncSession, Depends(get_db)]
|
||||
|
||||
_NAME_MAX_LEN = 100
|
||||
_DESCRIPTION_MAX_LEN = 2000
|
||||
|
||||
|
||||
# ──────────────────────────── Pydantic schemas ────────────────────────────────
|
||||
|
||||
|
||||
class GroupCreate(BaseModel):
|
||||
name: str = Field(..., min_length=1, max_length=_NAME_MAX_LEN)
|
||||
description: str | None = Field(default=None, max_length=_DESCRIPTION_MAX_LEN)
|
||||
|
||||
@field_validator("name")
|
||||
@classmethod
|
||||
def strip_name(cls, v: str) -> str:
|
||||
v = v.strip()
|
||||
if not v:
|
||||
raise ValueError("name must not be blank")
|
||||
return v
|
||||
|
||||
@field_validator("description")
|
||||
@classmethod
|
||||
def strip_description(cls, v: str | None) -> str | None:
|
||||
if v is not None:
|
||||
v = v.strip() or None
|
||||
return v
|
||||
|
||||
|
||||
class GroupUpdate(BaseModel):
|
||||
name: str | None = Field(default=None, min_length=1, max_length=_NAME_MAX_LEN)
|
||||
description: str | None = Field(default=None, max_length=_DESCRIPTION_MAX_LEN)
|
||||
|
||||
@field_validator("name")
|
||||
@classmethod
|
||||
def strip_name(cls, v: str | None) -> str | None:
|
||||
if v is not None:
|
||||
v = v.strip()
|
||||
if not v:
|
||||
raise ValueError("name must not be blank")
|
||||
return v
|
||||
|
||||
@field_validator("description")
|
||||
@classmethod
|
||||
def strip_description(cls, v: str | None) -> str | None:
|
||||
if v is not None:
|
||||
v = v.strip() or None
|
||||
return v
|
||||
|
||||
|
||||
class GroupSummary(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
description: str | None
|
||||
owner_id: int | None
|
||||
member_count: int
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class GroupListResponse(BaseModel):
|
||||
groups: list[GroupSummary]
|
||||
total: int
|
||||
|
||||
|
||||
class GroupMemberOut(BaseModel):
|
||||
"""Public-safe member entry -- no email or other private User fields."""
|
||||
|
||||
user_id: int
|
||||
username: str
|
||||
role: GroupRole
|
||||
joined_at: datetime
|
||||
|
||||
|
||||
class GroupDetail(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
description: str | None
|
||||
owner_id: int | None
|
||||
member_count: int
|
||||
members: list[GroupMemberOut]
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class RoleUpdateRequest(BaseModel):
|
||||
role: GroupRole
|
||||
|
||||
@field_validator("role")
|
||||
@classmethod
|
||||
def only_admin_or_member(cls, v: GroupRole) -> GroupRole:
|
||||
if v == GroupRole.OWNER:
|
||||
raise ValueError(
|
||||
"cannot set role to owner directly -- use POST /transfer"
|
||||
)
|
||||
return v
|
||||
|
||||
|
||||
class TransferRequest(BaseModel):
|
||||
new_owner_id: int
|
||||
|
||||
|
||||
class MembershipOut(BaseModel):
|
||||
group_id: int
|
||||
user_id: int
|
||||
role: GroupRole
|
||||
joined_at: datetime
|
||||
|
||||
|
||||
# ──────────────────────────────── Helpers ─────────────────────────────────────
|
||||
|
||||
|
||||
async def _get_group_or_404(db: AsyncSession, group_id: int) -> Group:
|
||||
group = await db.get(Group, group_id)
|
||||
if group is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
|
||||
return group
|
||||
|
||||
|
||||
async def _get_membership(db: AsyncSession, group_id: int, user_id: int) -> GroupMember | None:
|
||||
result = await db.execute(
|
||||
select(GroupMember).where(
|
||||
GroupMember.group_id == group_id, GroupMember.user_id == user_id
|
||||
)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
async def _member_count(db: AsyncSession, group_id: int) -> int:
|
||||
result = await db.execute(
|
||||
select(func.count()).select_from(GroupMember).where(GroupMember.group_id == group_id)
|
||||
)
|
||||
return int(result.scalar_one())
|
||||
|
||||
|
||||
def _require_role(membership: GroupMember | None, *allowed: GroupRole) -> None:
|
||||
if membership is None or membership.role not in allowed:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="You do not have permission to perform this action",
|
||||
)
|
||||
|
||||
|
||||
# ──────────────────────────────── Endpoints ───────────────────────────────────
|
||||
|
||||
|
||||
@router.post("", response_model=GroupDetail, status_code=status.HTTP_201_CREATED)
|
||||
@router.post("/", response_model=GroupDetail, status_code=status.HTTP_201_CREATED, include_in_schema=False)
|
||||
async def create_group(
|
||||
body: GroupCreate,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Create a group. The creator becomes its owner."""
|
||||
group = Group(name=body.name, description=body.description, owner_id=current_user.id)
|
||||
db.add(group)
|
||||
await db.flush()
|
||||
|
||||
db.add(GroupMember(group_id=group.id, user_id=current_user.id, role=GroupRole.OWNER))
|
||||
await db.commit()
|
||||
await db.refresh(group)
|
||||
|
||||
return GroupDetail(
|
||||
id=group.id,
|
||||
name=group.name,
|
||||
description=group.description,
|
||||
owner_id=group.owner_id,
|
||||
member_count=1,
|
||||
members=[
|
||||
GroupMemberOut(
|
||||
user_id=current_user.id,
|
||||
username=current_user.username,
|
||||
role=GroupRole.OWNER,
|
||||
joined_at=group.created_at,
|
||||
)
|
||||
],
|
||||
created_at=group.created_at,
|
||||
)
|
||||
|
||||
|
||||
@router.get("", response_model=GroupListResponse)
|
||||
@router.get("/", response_model=GroupListResponse, include_in_schema=False)
|
||||
async def list_groups(
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
skip: int = Query(0, ge=0),
|
||||
limit: int = Query(50, ge=1, le=100),
|
||||
mine: bool = Query(False, description="If true, only groups the caller is a member of"),
|
||||
):
|
||||
"""List groups, paginated. With ``mine=true``, restrict to the caller's own groups."""
|
||||
base_query = select(Group)
|
||||
count_query = select(func.count()).select_from(Group)
|
||||
|
||||
if mine:
|
||||
base_query = base_query.join(
|
||||
GroupMember, GroupMember.group_id == Group.id
|
||||
).where(GroupMember.user_id == current_user.id)
|
||||
count_query = count_query.join(
|
||||
GroupMember, GroupMember.group_id == Group.id
|
||||
).where(GroupMember.user_id == current_user.id)
|
||||
|
||||
total = (await db.execute(count_query)).scalar_one()
|
||||
|
||||
result = await db.execute(
|
||||
base_query.order_by(Group.created_at.desc()).offset(skip).limit(limit)
|
||||
)
|
||||
groups = result.scalars().all()
|
||||
|
||||
summaries = []
|
||||
for group in groups:
|
||||
summaries.append(
|
||||
GroupSummary(
|
||||
id=group.id,
|
||||
name=group.name,
|
||||
description=group.description,
|
||||
owner_id=group.owner_id,
|
||||
member_count=await _member_count(db, group.id),
|
||||
created_at=group.created_at,
|
||||
)
|
||||
)
|
||||
|
||||
return GroupListResponse(groups=summaries, total=total)
|
||||
|
||||
|
||||
@router.get("/{group_id}", response_model=GroupDetail)
|
||||
async def get_group(
|
||||
group_id: int,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Group page: name, description, member count, and member list with roles."""
|
||||
group = await _get_group_or_404(db, group_id)
|
||||
|
||||
result = await db.execute(
|
||||
select(GroupMember, User.username)
|
||||
.join(User, User.id == GroupMember.user_id)
|
||||
.where(GroupMember.group_id == group_id)
|
||||
.order_by(GroupMember.joined_at.asc())
|
||||
)
|
||||
rows = result.all()
|
||||
|
||||
members = [
|
||||
GroupMemberOut(
|
||||
user_id=gm.user_id,
|
||||
username=username,
|
||||
role=gm.role,
|
||||
joined_at=gm.joined_at,
|
||||
)
|
||||
for gm, username in rows
|
||||
]
|
||||
|
||||
return GroupDetail(
|
||||
id=group.id,
|
||||
name=group.name,
|
||||
description=group.description,
|
||||
owner_id=group.owner_id,
|
||||
member_count=len(members),
|
||||
members=members,
|
||||
created_at=group.created_at,
|
||||
)
|
||||
|
||||
|
||||
@router.patch("/{group_id}", response_model=GroupDetail)
|
||||
async def update_group(
|
||||
group_id: int,
|
||||
body: GroupUpdate,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Update name/description. Owner/admin only."""
|
||||
group = await _get_group_or_404(db, group_id)
|
||||
membership = await _get_membership(db, group_id, current_user.id)
|
||||
_require_role(membership, GroupRole.OWNER, GroupRole.ADMIN)
|
||||
|
||||
if body.name is not None:
|
||||
group.name = body.name
|
||||
if body.description is not None:
|
||||
group.description = body.description if body.description else None
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(group)
|
||||
|
||||
return await get_group(group_id, current_user, db)
|
||||
|
||||
|
||||
@router.delete("/{group_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_group(
|
||||
group_id: int,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Delete a group. Owner only."""
|
||||
group = await _get_group_or_404(db, group_id)
|
||||
if group.owner_id != current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only the owner can delete this group",
|
||||
)
|
||||
|
||||
await db.delete(group)
|
||||
await db.commit()
|
||||
return None
|
||||
|
||||
|
||||
@router.post("/{group_id}/join", response_model=MembershipOut)
|
||||
async def join_group(
|
||||
group_id: int,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Join a group. Idempotent -- joining an already-joined group is a no-op."""
|
||||
await _get_group_or_404(db, group_id)
|
||||
|
||||
existing = await _get_membership(db, group_id, current_user.id)
|
||||
if existing is not None:
|
||||
return MembershipOut(
|
||||
group_id=group_id,
|
||||
user_id=current_user.id,
|
||||
role=existing.role,
|
||||
joined_at=existing.joined_at,
|
||||
)
|
||||
|
||||
membership = GroupMember(group_id=group_id, user_id=current_user.id, role=GroupRole.MEMBER)
|
||||
db.add(membership)
|
||||
await db.commit()
|
||||
await db.refresh(membership)
|
||||
|
||||
return MembershipOut(
|
||||
group_id=group_id,
|
||||
user_id=current_user.id,
|
||||
role=membership.role,
|
||||
joined_at=membership.joined_at,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{group_id}/leave", response_model=dict)
|
||||
async def leave_group(
|
||||
group_id: int,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Leave a group. The owner cannot leave without transferring ownership first."""
|
||||
await _get_group_or_404(db, group_id)
|
||||
membership = await _get_membership(db, group_id, current_user.id)
|
||||
if membership is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="You are not a member of this group",
|
||||
)
|
||||
|
||||
if membership.role == GroupRole.OWNER:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail="Transfer ownership before leaving this group",
|
||||
)
|
||||
|
||||
await db.delete(membership)
|
||||
await db.commit()
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@router.post("/{group_id}/members/{user_id}/role", response_model=MembershipOut)
|
||||
async def set_member_role(
|
||||
group_id: int,
|
||||
user_id: int,
|
||||
body: RoleUpdateRequest,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Set a member's role (admin/member). Owner only."""
|
||||
group = await _get_group_or_404(db, group_id)
|
||||
if group.owner_id != current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only the owner can change member roles",
|
||||
)
|
||||
|
||||
target = await _get_membership(db, group_id, user_id)
|
||||
if target is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Member not found")
|
||||
|
||||
if target.role == GroupRole.OWNER:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Use POST /transfer to change ownership",
|
||||
)
|
||||
|
||||
target.role = body.role
|
||||
await db.commit()
|
||||
await db.refresh(target)
|
||||
|
||||
return MembershipOut(
|
||||
group_id=group_id, user_id=user_id, role=target.role, joined_at=target.joined_at
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{group_id}/transfer", response_model=GroupDetail)
|
||||
async def transfer_ownership(
|
||||
group_id: int,
|
||||
body: TransferRequest,
|
||||
current_user: CurrentUser,
|
||||
db: DbDep,
|
||||
):
|
||||
"""Transfer group ownership to another existing member. Owner only."""
|
||||
group = await _get_group_or_404(db, group_id)
|
||||
if group.owner_id != current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only the owner can transfer ownership",
|
||||
)
|
||||
|
||||
if body.new_owner_id == current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="You already own this group",
|
||||
)
|
||||
|
||||
new_owner_membership = await _get_membership(db, group_id, body.new_owner_id)
|
||||
if new_owner_membership is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="New owner must already be a member of this group",
|
||||
)
|
||||
|
||||
current_owner_membership = await _get_membership(db, group_id, current_user.id)
|
||||
|
||||
new_owner_membership.role = GroupRole.OWNER
|
||||
if current_owner_membership is not None:
|
||||
current_owner_membership.role = GroupRole.ADMIN
|
||||
group.owner_id = body.new_owner_id
|
||||
|
||||
await db.commit()
|
||||
|
||||
return await get_group(group_id, current_user, db)
|
||||
@@ -0,0 +1,437 @@
|
||||
"""Integration tests for the Groups API router (app/routers/groups.py).
|
||||
|
||||
app/models/groups.py owns the schema (see test_group_models.py); this
|
||||
covers create/join/leave, role authorization, ownership transfer, and the
|
||||
group page's member-list privacy.
|
||||
"""
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
|
||||
|
||||
def _register(client: TestClient, username: str) -> dict:
|
||||
resp = client.post(
|
||||
"/api/auth/register",
|
||||
json={
|
||||
"username": username,
|
||||
"email": f"{username}@example.com",
|
||||
"password": "secret123",
|
||||
"date_of_birth": "1990-01-01",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
token = resp.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
me = client.get("/api/users/me", headers=headers)
|
||||
assert me.status_code == 200, me.text
|
||||
return {"headers": headers, "id": me.json()["id"], "username": username}
|
||||
|
||||
|
||||
def _create_group(client: TestClient, headers: dict, name: str, description: str | None = None) -> dict:
|
||||
resp = client.post(
|
||||
"/api/groups",
|
||||
json={"name": name, "description": description},
|
||||
headers=headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()
|
||||
|
||||
|
||||
class TestCreateGroup:
|
||||
def test_create_makes_creator_owner(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_create_owner")
|
||||
|
||||
group = _create_group(client, owner["headers"], "Explorers", "We explore worlds")
|
||||
assert group["name"] == "Explorers"
|
||||
assert group["description"] == "We explore worlds"
|
||||
assert group["owner_id"] == owner["id"]
|
||||
assert group["member_count"] == 1
|
||||
assert len(group["members"]) == 1
|
||||
assert group["members"][0]["user_id"] == owner["id"]
|
||||
assert group["members"][0]["role"] == "owner"
|
||||
|
||||
def test_requires_auth(self):
|
||||
client = TestClient(app)
|
||||
resp = client.post("/api/groups", json={"name": "No Auth"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_blank_name_rejected(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_blank_name")
|
||||
resp = client.post("/api/groups", json={"name": " "}, headers=owner["headers"])
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_name_over_max_length_rejected(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_long_name")
|
||||
resp = client.post(
|
||||
"/api/groups", json={"name": "x" * 101}, headers=owner["headers"]
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
class TestGroupPage:
|
||||
def test_get_group_page_shows_members_and_roles_no_private_fields(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_page_owner")
|
||||
member = _register(client, "grp_api_page_member")
|
||||
|
||||
group = _create_group(client, owner["headers"], "Page Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.get(f"/api/groups/{group['id']}", headers=member["headers"])
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["name"] == "Page Group"
|
||||
assert body["member_count"] == 2
|
||||
by_user = {m["user_id"]: m for m in body["members"]}
|
||||
assert by_user[owner["id"]]["role"] == "owner"
|
||||
assert by_user[member["id"]]["role"] == "member"
|
||||
assert by_user[member["id"]]["username"] == member["username"]
|
||||
|
||||
# No private User fields (email, hashed_password, etc.) leak.
|
||||
for m in body["members"]:
|
||||
assert set(m.keys()) == {"user_id", "username", "role", "joined_at"}
|
||||
|
||||
def test_get_nonexistent_group_is_404(self):
|
||||
client = TestClient(app)
|
||||
user = _register(client, "grp_api_404")
|
||||
resp = client.get("/api/groups/999999999", headers=user["headers"])
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
class TestJoinLeave:
|
||||
def test_join_is_idempotent(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_join_owner")
|
||||
joiner = _register(client, "grp_api_join_joiner")
|
||||
group = _create_group(client, owner["headers"], "Join Group")
|
||||
|
||||
r1 = client.post(f"/api/groups/{group['id']}/join", headers=joiner["headers"])
|
||||
assert r1.status_code == 200, r1.text
|
||||
r2 = client.post(f"/api/groups/{group['id']}/join", headers=joiner["headers"])
|
||||
assert r2.status_code == 200, r2.text
|
||||
assert r1.json()["role"] == r2.json()["role"] == "member"
|
||||
|
||||
page = client.get(f"/api/groups/{group['id']}", headers=owner["headers"]).json()
|
||||
assert page["member_count"] == 2 # not 3 -- duplicate join didn't add a row
|
||||
|
||||
def test_leave_removes_membership(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_leave_owner")
|
||||
member = _register(client, "grp_api_leave_member")
|
||||
group = _create_group(client, owner["headers"], "Leave Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.post(f"/api/groups/{group['id']}/leave", headers=member["headers"])
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
page = client.get(f"/api/groups/{group['id']}", headers=owner["headers"]).json()
|
||||
assert page["member_count"] == 1
|
||||
assert all(m["user_id"] != member["id"] for m in page["members"])
|
||||
|
||||
def test_owner_cannot_leave_without_transfer(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_owner_leave")
|
||||
group = _create_group(client, owner["headers"], "Owner Leave Group")
|
||||
|
||||
resp = client.post(f"/api/groups/{group['id']}/leave", headers=owner["headers"])
|
||||
assert resp.status_code == 409, resp.text
|
||||
|
||||
def test_leave_when_not_a_member_is_404(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_notmember_owner")
|
||||
stranger = _register(client, "grp_api_notmember_stranger")
|
||||
group = _create_group(client, owner["headers"], "Not A Member Group")
|
||||
|
||||
resp = client.post(f"/api/groups/{group['id']}/leave", headers=stranger["headers"])
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
class TestRoleAuthorization:
|
||||
def test_owner_can_promote_and_demote(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_role_owner")
|
||||
member = _register(client, "grp_api_role_member")
|
||||
group = _create_group(client, owner["headers"], "Role Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
promote = client.post(
|
||||
f"/api/groups/{group['id']}/members/{member['id']}/role",
|
||||
json={"role": "admin"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert promote.status_code == 200, promote.text
|
||||
assert promote.json()["role"] == "admin"
|
||||
|
||||
demote = client.post(
|
||||
f"/api/groups/{group['id']}/members/{member['id']}/role",
|
||||
json={"role": "member"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert demote.status_code == 200, demote.text
|
||||
assert demote.json()["role"] == "member"
|
||||
|
||||
def test_plain_member_cannot_promote(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_noauth_owner")
|
||||
member = _register(client, "grp_api_noauth_member")
|
||||
other = _register(client, "grp_api_noauth_other")
|
||||
group = _create_group(client, owner["headers"], "No Auth Promote Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=other["headers"])
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/members/{other['id']}/role",
|
||||
json={"role": "admin"},
|
||||
headers=member["headers"],
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_admin_cannot_promote_or_demote(self):
|
||||
"""Only the owner may change roles -- admins cannot, per spec."""
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_adminauth_owner")
|
||||
admin = _register(client, "grp_api_adminauth_admin")
|
||||
other = _register(client, "grp_api_adminauth_other")
|
||||
group = _create_group(client, owner["headers"], "Admin Cannot Promote Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=admin["headers"])
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=other["headers"])
|
||||
client.post(
|
||||
f"/api/groups/{group['id']}/members/{admin['id']}/role",
|
||||
json={"role": "admin"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/members/{other['id']}/role",
|
||||
json={"role": "admin"},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_plain_member_cannot_delete_group(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_nodelete_owner")
|
||||
member = _register(client, "grp_api_nodelete_member")
|
||||
group = _create_group(client, owner["headers"], "No Delete Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.delete(f"/api/groups/{group['id']}", headers=member["headers"])
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_admin_cannot_delete_group(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_admindelete_owner")
|
||||
admin = _register(client, "grp_api_admindelete_admin")
|
||||
group = _create_group(client, owner["headers"], "Admin No Delete Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=admin["headers"])
|
||||
client.post(
|
||||
f"/api/groups/{group['id']}/members/{admin['id']}/role",
|
||||
json={"role": "admin"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
|
||||
resp = client.delete(f"/api/groups/{group['id']}", headers=admin["headers"])
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_plain_member_cannot_transfer_ownership(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_notransfer_owner")
|
||||
member = _register(client, "grp_api_notransfer_member")
|
||||
group = _create_group(client, owner["headers"], "No Transfer Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/transfer",
|
||||
json={"new_owner_id": member["id"]},
|
||||
headers=member["headers"],
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_cannot_set_role_to_owner_via_role_endpoint(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_noownerrole_owner")
|
||||
member = _register(client, "grp_api_noownerrole_member")
|
||||
group = _create_group(client, owner["headers"], "No Owner Role Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/members/{member['id']}/role",
|
||||
json={"role": "owner"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
class TestOwnershipTransfer:
|
||||
def test_transfer_moves_ownership_and_demotes_old_owner(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_xfer_owner")
|
||||
member = _register(client, "grp_api_xfer_member")
|
||||
group = _create_group(client, owner["headers"], "Transfer Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/transfer",
|
||||
json={"new_owner_id": member["id"]},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["owner_id"] == member["id"]
|
||||
by_user = {m["user_id"]: m for m in body["members"]}
|
||||
assert by_user[member["id"]]["role"] == "owner"
|
||||
assert by_user[owner["id"]]["role"] == "admin"
|
||||
|
||||
# Exactly one owner remains.
|
||||
owners = [m for m in body["members"] if m["role"] == "owner"]
|
||||
assert len(owners) == 1
|
||||
|
||||
# Old owner can now leave (no longer the owner).
|
||||
leave = client.post(f"/api/groups/{group['id']}/leave", headers=owner["headers"])
|
||||
assert leave.status_code == 200, leave.text
|
||||
|
||||
# New owner cannot leave without transferring again.
|
||||
leave2 = client.post(f"/api/groups/{group['id']}/leave", headers=member["headers"])
|
||||
assert leave2.status_code == 409
|
||||
|
||||
def test_transfer_target_must_be_a_member(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_xfernonmember_owner")
|
||||
stranger = _register(client, "grp_api_xfernonmember_stranger")
|
||||
group = _create_group(client, owner["headers"], "Transfer Nonmember Group")
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/transfer",
|
||||
json={"new_owner_id": stranger["id"]},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_transfer_to_self_rejected(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_xferself_owner")
|
||||
group = _create_group(client, owner["headers"], "Transfer Self Group")
|
||||
|
||||
resp = client.post(
|
||||
f"/api/groups/{group['id']}/transfer",
|
||||
json={"new_owner_id": owner["id"]},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
class TestUpdateGroup:
|
||||
def test_owner_can_update_name_and_description(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_update_owner")
|
||||
group = _create_group(client, owner["headers"], "Old Name", "Old desc")
|
||||
|
||||
resp = client.patch(
|
||||
f"/api/groups/{group['id']}",
|
||||
json={"name": "New Name", "description": "New desc"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["name"] == "New Name"
|
||||
assert resp.json()["description"] == "New desc"
|
||||
|
||||
def test_admin_can_update(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_adminupdate_owner")
|
||||
admin = _register(client, "grp_api_adminupdate_admin")
|
||||
group = _create_group(client, owner["headers"], "Admin Update Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=admin["headers"])
|
||||
client.post(
|
||||
f"/api/groups/{group['id']}/members/{admin['id']}/role",
|
||||
json={"role": "admin"},
|
||||
headers=owner["headers"],
|
||||
)
|
||||
|
||||
resp = client.patch(
|
||||
f"/api/groups/{group['id']}",
|
||||
json={"name": "Updated By Admin"},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["name"] == "Updated By Admin"
|
||||
|
||||
def test_plain_member_cannot_update(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_noupdate_owner")
|
||||
member = _register(client, "grp_api_noupdate_member")
|
||||
group = _create_group(client, owner["headers"], "No Update Group")
|
||||
client.post(f"/api/groups/{group['id']}/join", headers=member["headers"])
|
||||
|
||||
resp = client.patch(
|
||||
f"/api/groups/{group['id']}",
|
||||
json={"name": "Hacked Name"},
|
||||
headers=member["headers"],
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_non_member_cannot_update(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_strangernoupdate_owner")
|
||||
stranger = _register(client, "grp_api_strangernoupdate_stranger")
|
||||
group = _create_group(client, owner["headers"], "Stranger No Update Group")
|
||||
|
||||
resp = client.patch(
|
||||
f"/api/groups/{group['id']}",
|
||||
json={"name": "Hacked Name"},
|
||||
headers=stranger["headers"],
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
class TestDeleteGroup:
|
||||
def test_owner_can_delete(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_delete_owner")
|
||||
group = _create_group(client, owner["headers"], "Delete Group")
|
||||
|
||||
resp = client.delete(f"/api/groups/{group['id']}", headers=owner["headers"])
|
||||
assert resp.status_code == 204, resp.text
|
||||
|
||||
after = client.get(f"/api/groups/{group['id']}", headers=owner["headers"])
|
||||
assert after.status_code == 404
|
||||
|
||||
|
||||
class TestListGroups:
|
||||
def test_list_all_groups_paginated(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_list_owner")
|
||||
_create_group(client, owner["headers"], "List Group A")
|
||||
_create_group(client, owner["headers"], "List Group B")
|
||||
|
||||
resp = client.get("/api/groups", params={"limit": 1}, headers=owner["headers"])
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert len(body["groups"]) == 1
|
||||
assert body["total"] >= 2
|
||||
|
||||
def test_list_mine_only_returns_own_groups(self):
|
||||
client = TestClient(app)
|
||||
owner = _register(client, "grp_api_listmine_owner")
|
||||
other_owner = _register(client, "grp_api_listmine_other")
|
||||
mine = _create_group(client, owner["headers"], "Mine Group")
|
||||
_create_group(client, other_owner["headers"], "Not Mine Group")
|
||||
|
||||
resp = client.get("/api/groups", params={"mine": True}, headers=owner["headers"])
|
||||
assert resp.status_code == 200, resp.text
|
||||
ids = [g["id"] for g in resp.json()["groups"]]
|
||||
assert mine["id"] in ids
|
||||
assert all(
|
||||
g["id"] != mine["id"] or g["owner_id"] == owner["id"]
|
||||
for g in resp.json()["groups"]
|
||||
)
|
||||
# The other owner's group must not appear.
|
||||
other_group_ids = [
|
||||
g["id"] for g in client.get(
|
||||
"/api/groups", params={"mine": True}, headers=other_owner["headers"]
|
||||
).json()["groups"]
|
||||
]
|
||||
assert mine["id"] not in other_group_ids
|
||||
Reference in New Issue
Block a user