Phase 1 backend implementation: Add invite management endpoints
Implements core invite management features for band admins:
- GET /bands/{band_id}/invites - List all invites for a band (admin only)
- DELETE /invites/{invite_id} - Revoke pending invite (admin only)
- GET /invites/{token}/info - Get invite details (public)
Backend changes:
- Add invites router with 3 endpoints
- Update BandRepository with get_invites_for_band and get_invite_by_id methods
- Add new schemas for invite listing and info
- Register invites router in main.py
Tests:
- Integration tests for all 3 endpoints
- Permission tests (admin vs non-admin)
- Edge cases (not found, expired, etc.)
This addresses the core requirements:
- Admins can see pending invites
- Admins can revoke pending invites
- Users can view invite details before accepting
Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
Mistral Vibe
209
api/tests/integration/test_api_invites.py
Normal file
209
api/tests/integration/test_api_invites.py
Normal file
@@ -0,0 +1,209 @@
|
||||
"""Integration tests for invite endpoints."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.factories import create_band, create_member
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def band_with_admin(db_session):
|
||||
"""Create a band with an admin member."""
|
||||
admin = create_member(db_session, email="admin@test.com")
|
||||
band = create_band(db_session, creator_id=admin.id)
|
||||
db_session.commit()
|
||||
return {"band": band, "admin": admin}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def band_with_members(db_session):
|
||||
"""Create a band with admin and regular member."""
|
||||
admin = create_member(db_session, email="admin@test.com")
|
||||
member = create_member(db_session, email="member@test.com")
|
||||
band = create_band(db_session, creator_id=admin.id)
|
||||
# Add member to band
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
member_role = await repo.add_member(band.id, member.id)
|
||||
db_session.commit()
|
||||
return {"band": band, "admin": admin, "member": member}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_list_invites_admin_can_see(client, db_session, auth_headers_for, band_with_admin):
|
||||
"""Test that admin can list invites for their band."""
|
||||
headers = await auth_headers_for(band_with_admin["admin"])
|
||||
band = band_with_admin["band"]
|
||||
|
||||
# Create some invites
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
invite1 = await repo.create_invite(band.id, band_with_admin["admin"].id)
|
||||
invite2 = await repo.create_invite(band.id, band_with_admin["admin"].id)
|
||||
db_session.commit()
|
||||
|
||||
resp = await client.get(f"/api/v1/bands/{band.id}/invites", headers=headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert "invites" in data
|
||||
assert "total" in data
|
||||
assert "pending" in data
|
||||
assert data["total"] >= 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_list_invites_non_admin_returns_403(client, db_session, auth_headers_for, band_with_members):
|
||||
"""Test that non-admin cannot list invites."""
|
||||
headers = await auth_headers_for(band_with_members["member"])
|
||||
band = band_with_members["band"]
|
||||
|
||||
resp = await client.get(f"/api/v1/bands/{band.id}/invites", headers=headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_list_invites_no_invites_returns_empty(client, db_session, auth_headers_for, band_with_admin):
|
||||
"""Test listing invites when none exist."""
|
||||
headers = await auth_headers_for(band_with_admin["admin"])
|
||||
band = band_with_admin["band"]
|
||||
|
||||
resp = await client.get(f"/api/v1/bands/{band.id}/invites", headers=headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["invites"] == []
|
||||
assert data["total"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_list_invites_includes_pending_and_used(client, db_session, auth_headers_for, band_with_admin):
|
||||
"""Test that list includes both pending and used invites."""
|
||||
headers = await auth_headers_for(band_with_admin["admin"])
|
||||
band = band_with_admin["band"]
|
||||
|
||||
# Create invites with different statuses
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
|
||||
# Create pending invite
|
||||
pending_invite = await repo.create_invite(band.id, band_with_admin["admin"].id)
|
||||
|
||||
# Create used invite (simulate by setting used_at)
|
||||
used_invite = await repo.create_invite(band.id, band_with_admin["admin"].id)
|
||||
used_invite.used_at = datetime.now(timezone.utc)
|
||||
|
||||
db_session.commit()
|
||||
|
||||
resp = await client.get(f"/api/v1/bands/{band.id}/invites", headers=headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
|
||||
# Check we have both invites
|
||||
assert data["total"] >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_revoke_invite_admin_can_revoke(client, db_session, auth_headers_for, band_with_admin):
|
||||
"""Test that admin can revoke an invite."""
|
||||
headers = await auth_headers_for(band_with_admin["admin"])
|
||||
band = band_with_admin["band"]
|
||||
|
||||
# Create an invite
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
invite = await repo.create_invite(band.id, band_with_admin["admin"].id)
|
||||
invite_id = invite.id
|
||||
db_session.commit()
|
||||
|
||||
resp = await client.delete(f"/api/v1/invites/{invite_id}", headers=headers)
|
||||
assert resp.status_code == 204, resp.text
|
||||
|
||||
# Verify invite was revoked
|
||||
resp = await client.delete(f"/api/v1/invites/{invite_id}", headers=headers)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_revoke_invite_non_admin_returns_403(client, db_session, auth_headers_for, band_with_members):
|
||||
"""Test that non-admin cannot revoke invites."""
|
||||
headers = await auth_headers_for(band_with_members["member"])
|
||||
band = band_with_members["band"]
|
||||
|
||||
# Create an invite
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
invite = await repo.create_invite(band.id, band_with_members["admin"].id)
|
||||
invite_id = invite.id
|
||||
db_session.commit()
|
||||
|
||||
resp = await client.delete(f"/api/v1/invites/{invite_id}", headers=headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_revoke_invite_not_found_returns_404(client, db_session, auth_headers_for, band_with_admin):
|
||||
"""Test revoking a non-existent invite."""
|
||||
headers = await auth_headers_for(band_with_admin["admin"])
|
||||
|
||||
resp = await client.delete("/api/v1/invites/00000000-0000-0000-0000-000000000000", headers=headers)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_get_invite_info_valid_token(client, db_session):
|
||||
"""Test getting invite info with valid token."""
|
||||
admin = create_member(db_session, email="admin@test.com")
|
||||
band = create_band(db_session, creator_id=admin.id)
|
||||
|
||||
# Create an invite
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
invite = await repo.create_invite(band.id, admin.id)
|
||||
token = invite.token
|
||||
db_session.commit()
|
||||
|
||||
resp = await client.get(f"/api/v1/invites/{token}/info")
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["band_id"] == str(band.id)
|
||||
assert data["role"] == "member"
|
||||
assert data["is_used"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_get_invite_info_invalid_token(client):
|
||||
"""Test getting invite info with invalid token."""
|
||||
resp = await client.get("/api/v1/invites/invalid-token/info")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_get_invite_info_expired_invite(client, db_session):
|
||||
"""Test getting invite info for expired invite."""
|
||||
admin = create_member(db_session, email="admin@test.com")
|
||||
band = create_band(db_session, creator_id=admin.id)
|
||||
|
||||
# Create an invite with very short expiry
|
||||
from rehearsalhub.repositories.band import BandRepository
|
||||
repo = BandRepository(db_session)
|
||||
invite = await repo.create_invite(band.id, admin.id, ttl_hours=0)
|
||||
token = invite.token
|
||||
db_session.commit()
|
||||
|
||||
# Wait a bit for expiry
|
||||
import asyncio
|
||||
await asyncio.sleep(1)
|
||||
|
||||
resp = await client.get(f"/api/v1/invites/{token}/info")
|
||||
assert resp.status_code == 400
|
||||
Reference in New Issue
Block a user