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>
210 lines
7.4 KiB
Python
210 lines
7.4 KiB
Python
"""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
|