counterweight
280c1e5687
- Add get_available_slots() and _expand_availability_to_slots() to ExchangeService - Update routes/exchange.py to use ExchangeService.get_available_slots() - Remove all business logic from get_available_slots endpoint - Add AvailabilityRepository to ExchangeService dependencies - Add Availability and BookableSlot imports to ExchangeService - Fix import path for validate_date_in_range (use date_validation module) - Remove unused user_repo variable and import from routes/invites.py - Fix mypy error in ValidationError by adding proper type annotation
1164 lines
40 KiB
Python
1164 lines
40 KiB
Python
"""Tests for invite functionality."""
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from invite_utils import (
|
|
BIP39_WORDS,
|
|
generate_invite_identifier,
|
|
is_valid_identifier_format,
|
|
normalize_identifier,
|
|
)
|
|
from models import ROLE_REGULAR, Invite, InviteStatus, User
|
|
from tests.conftest import create_user_with_roles
|
|
from tests.helpers import unique_email
|
|
|
|
# ============================================================================
|
|
# Invite Utils Tests
|
|
# ============================================================================
|
|
|
|
|
|
def test_bip39_words_loaded():
|
|
"""BIP39 word list should have exactly 2048 words."""
|
|
assert len(BIP39_WORDS) == 2048
|
|
|
|
|
|
def test_generate_invite_identifier_format():
|
|
"""Generated identifier should have word-word-NN format."""
|
|
identifier = generate_invite_identifier()
|
|
assert is_valid_identifier_format(identifier)
|
|
|
|
parts = identifier.split("-")
|
|
assert len(parts) == 3
|
|
assert parts[0] in BIP39_WORDS
|
|
assert parts[1] in BIP39_WORDS
|
|
assert len(parts[2]) == 2
|
|
assert parts[2].isdigit()
|
|
|
|
|
|
def test_generate_invite_identifier_lowercase():
|
|
"""Generated identifier should be lowercase."""
|
|
for _ in range(10):
|
|
identifier = generate_invite_identifier()
|
|
assert identifier == identifier.lower()
|
|
|
|
|
|
def test_generate_invite_identifier_randomness():
|
|
"""Generated identifiers should be different (with high probability)."""
|
|
identifiers = {generate_invite_identifier() for _ in range(100)}
|
|
# With ~419M possibilities, 100 samples should all be unique
|
|
assert len(identifiers) == 100
|
|
|
|
|
|
def test_normalize_identifier_lowercase():
|
|
"""normalize_identifier should convert to lowercase."""
|
|
assert normalize_identifier("APPLE-BANANA-42") == "apple-banana-42"
|
|
assert normalize_identifier("Apple-Banana-42") == "apple-banana-42"
|
|
|
|
|
|
def test_normalize_identifier_strips_whitespace():
|
|
"""normalize_identifier should strip whitespace."""
|
|
assert normalize_identifier(" apple-banana-42 ") == "apple-banana-42"
|
|
assert normalize_identifier("\tapple-banana-42\n") == "apple-banana-42"
|
|
|
|
|
|
def test_is_valid_identifier_format_valid():
|
|
"""Valid formats should pass."""
|
|
assert is_valid_identifier_format("apple-banana-42") is True
|
|
assert is_valid_identifier_format("zoo-abandon-00") is True
|
|
assert is_valid_identifier_format("word-another-99") is True
|
|
|
|
|
|
def test_is_valid_identifier_format_invalid():
|
|
"""Invalid formats should fail."""
|
|
# Wrong number of parts
|
|
assert is_valid_identifier_format("apple-banana") is False
|
|
assert is_valid_identifier_format("apple-banana-42-extra") is False
|
|
assert is_valid_identifier_format("applebanan42") is False
|
|
|
|
# Empty parts
|
|
assert is_valid_identifier_format("-banana-42") is False
|
|
assert is_valid_identifier_format("apple--42") is False
|
|
|
|
# Invalid number format
|
|
assert is_valid_identifier_format("apple-banana-4") is False # Single digit
|
|
assert is_valid_identifier_format("apple-banana-420") is False # Three digits
|
|
assert is_valid_identifier_format("apple-banana-ab") is False # Not digits
|
|
|
|
|
|
# ============================================================================
|
|
# Invite Model Tests
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invite(client_factory):
|
|
"""Can create an invite with godfather."""
|
|
async with client_factory.get_db_session() as db:
|
|
# Create godfather user
|
|
godfather = await create_user_with_roles(
|
|
db, unique_email("godfather"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
# Create invite
|
|
invite = Invite(
|
|
identifier="test-invite-01",
|
|
godfather_id=godfather.id,
|
|
status=InviteStatus.READY,
|
|
)
|
|
db.add(invite)
|
|
await db.commit()
|
|
await db.refresh(invite)
|
|
|
|
assert invite.id is not None
|
|
assert invite.identifier == "test-invite-01"
|
|
assert invite.godfather_id == godfather.id
|
|
assert invite.status == InviteStatus.READY
|
|
assert invite.used_by_id is None
|
|
assert invite.created_at is not None
|
|
assert invite.spent_at is None
|
|
assert invite.revoked_at is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invite_godfather_relationship(client_factory):
|
|
"""Invite should have godfather relationship loaded."""
|
|
async with client_factory.get_db_session() as db:
|
|
godfather = await create_user_with_roles(
|
|
db, unique_email("godfather"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
invite = Invite(
|
|
identifier="rel-test-01",
|
|
godfather_id=godfather.id,
|
|
)
|
|
db.add(invite)
|
|
await db.commit()
|
|
|
|
# Query invite fresh
|
|
result = await db.execute(
|
|
select(Invite).where(Invite.identifier == "rel-test-01")
|
|
)
|
|
loaded_invite = result.scalar_one()
|
|
|
|
assert loaded_invite.godfather is not None
|
|
assert loaded_invite.godfather.email == godfather.email
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invite_unique_identifier(client_factory):
|
|
"""Invite identifier must be unique."""
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
godfather = await create_user_with_roles(
|
|
db, unique_email("godfather"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
invite1 = Invite(
|
|
identifier="unique-test-01",
|
|
godfather_id=godfather.id,
|
|
)
|
|
db.add(invite1)
|
|
await db.commit()
|
|
|
|
invite2 = Invite(
|
|
identifier="unique-test-01", # Same identifier
|
|
godfather_id=godfather.id,
|
|
)
|
|
db.add(invite2)
|
|
|
|
with pytest.raises(IntegrityError):
|
|
await db.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invite_status_transitions(client_factory):
|
|
"""Invite status can be changed."""
|
|
from datetime import UTC, datetime
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
godfather = await create_user_with_roles(
|
|
db, unique_email("godfather"), "password123", [ROLE_REGULAR]
|
|
)
|
|
user = await create_user_with_roles(
|
|
db, unique_email("invitee"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
invite = Invite(
|
|
identifier="status-test-01",
|
|
godfather_id=godfather.id,
|
|
status=InviteStatus.READY,
|
|
)
|
|
db.add(invite)
|
|
await db.commit()
|
|
|
|
# Transition to SPENT
|
|
invite.status = InviteStatus.SPENT
|
|
invite.used_by_id = user.id
|
|
invite.spent_at = datetime.now(UTC)
|
|
await db.commit()
|
|
await db.refresh(invite)
|
|
|
|
assert invite.status == InviteStatus.SPENT
|
|
assert invite.used_by_id == user.id
|
|
assert invite.spent_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invite_revoke(client_factory):
|
|
"""Invite can be revoked."""
|
|
from datetime import UTC, datetime
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
godfather = await create_user_with_roles(
|
|
db, unique_email("godfather"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
invite = Invite(
|
|
identifier="revoke-test-01",
|
|
godfather_id=godfather.id,
|
|
status=InviteStatus.READY,
|
|
)
|
|
db.add(invite)
|
|
await db.commit()
|
|
|
|
# Revoke
|
|
invite.status = InviteStatus.REVOKED
|
|
invite.revoked_at = datetime.now(UTC)
|
|
await db.commit()
|
|
await db.refresh(invite)
|
|
|
|
assert invite.status == InviteStatus.REVOKED
|
|
assert invite.revoked_at is not None
|
|
assert invite.used_by_id is None # Not used
|
|
|
|
|
|
# ============================================================================
|
|
# User Godfather Tests
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_godfather_relationship(client_factory):
|
|
"""User can have a godfather."""
|
|
async with client_factory.get_db_session() as db:
|
|
godfather = await create_user_with_roles(
|
|
db, unique_email("godfather"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
# Create user with godfather
|
|
user = User(
|
|
email=unique_email("godchild"),
|
|
hashed_password="hashed",
|
|
godfather_id=godfather.id,
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
|
|
# Query user fresh
|
|
result = await db.execute(select(User).where(User.id == user.id))
|
|
loaded_user = result.scalar_one()
|
|
|
|
assert loaded_user.godfather_id == godfather.id
|
|
assert loaded_user.godfather is not None
|
|
assert loaded_user.godfather.email == godfather.email
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_without_godfather(client_factory):
|
|
"""User can exist without godfather (for seeded/admin users)."""
|
|
async with client_factory.get_db_session() as db:
|
|
user = await create_user_with_roles(
|
|
db, unique_email("noparent"), "password123", [ROLE_REGULAR]
|
|
)
|
|
|
|
assert user.godfather_id is None
|
|
assert user.godfather is None
|
|
|
|
|
|
# ============================================================================
|
|
# Admin Create Invite API Tests (Phase 2)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_create_invite(client_factory, admin_user, regular_user):
|
|
"""Admin can create an invite for a regular user."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
# Get regular user ID
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
response = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["godfather_id"] == godfather.id
|
|
assert data["godfather_email"] == regular_user["email"]
|
|
assert data["status"] == "ready"
|
|
assert data["used_by_id"] is None
|
|
assert data["used_by_email"] is None
|
|
assert data["spent_at"] is None
|
|
assert data["revoked_at"] is None
|
|
assert "-" in data["identifier"] # word-word-NN format
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_create_invite_for_self(client_factory, admin_user):
|
|
"""Admin can create an invite for themselves."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
# Get admin user ID
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == admin_user["email"])
|
|
)
|
|
admin = result.scalar_one()
|
|
|
|
response = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": admin.id},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["godfather_id"] == admin.id
|
|
assert data["godfather_email"] == admin_user["email"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_regular_user_cannot_create_invite(client_factory, regular_user):
|
|
"""Regular user cannot create invites (403)."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": 1},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthenticated_cannot_create_invite(client_factory):
|
|
"""Unauthenticated user cannot create invites (401)."""
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": 1},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invite_invalid_godfather(client_factory, admin_user):
|
|
"""Creating invite with non-existent godfather returns 400."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
response = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": 99999},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "not found" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_created_invite_persisted_in_db(client_factory, admin_user, regular_user):
|
|
"""Created invite is persisted in database."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
response = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
|
|
data = response.json()
|
|
invite_id = data["id"]
|
|
|
|
# Query from DB
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(Invite).where(Invite.id == invite_id))
|
|
invite = result.scalar_one()
|
|
|
|
assert invite.identifier == data["identifier"]
|
|
assert invite.godfather_id == godfather.id
|
|
assert invite.status == InviteStatus.READY
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invite_retries_on_collision(
|
|
client_factory, admin_user, regular_user
|
|
):
|
|
"""Create invite retries with new identifier on collision."""
|
|
from unittest.mock import patch
|
|
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
# Create first invite normally
|
|
response1 = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
assert response1.status_code == 200
|
|
identifier1 = response1.json()["identifier"]
|
|
|
|
# Mock generator to first return the same identifier (collision), then a new one
|
|
call_count = 0
|
|
|
|
def mock_generator():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return identifier1 # Will collide
|
|
return f"unique-word-{call_count:02d}" # Won't collide
|
|
|
|
with patch(
|
|
"services.invite.generate_invite_identifier", side_effect=mock_generator
|
|
):
|
|
response2 = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
|
|
assert response2.status_code == 200
|
|
# Should have retried and gotten a new identifier
|
|
assert response2.json()["identifier"] != identifier1
|
|
assert call_count >= 2 # At least one retry
|
|
|
|
|
|
# ============================================================================
|
|
# Invite Check API Tests (Phase 3)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_invite_valid(client_factory, admin_user, regular_user):
|
|
"""Check endpoint returns valid=True for READY invite."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Check invite (no auth needed)
|
|
async with client_factory.create() as client:
|
|
response = await client.get(f"/api/invites/{identifier}/check")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is True
|
|
assert data["status"] == "ready"
|
|
assert data["error"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_invite_not_found(client_factory):
|
|
"""Check endpoint returns valid=False for unknown invite."""
|
|
async with client_factory.create() as client:
|
|
response = await client.get("/api/invites/fake-invite-99/check")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
assert "not found" in data["error"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_invite_invalid_format(client_factory):
|
|
"""Check endpoint returns error for invalid format without querying DB."""
|
|
async with client_factory.create() as client:
|
|
# Missing number part
|
|
response = await client.get("/api/invites/word-word/check")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
assert "format" in data["error"].lower()
|
|
|
|
# Single digit number
|
|
response = await client.get("/api/invites/word-word-1/check")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
assert "format" in data["error"].lower()
|
|
|
|
# Too many parts
|
|
response = await client.get("/api/invites/word-word-word-00/check")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
assert "format" in data["error"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_invite_spent_returns_not_found(
|
|
client_factory, admin_user, regular_user
|
|
):
|
|
"""Check endpoint returns same error for spent invite as for non-existent (no info leakage)."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Use the invite
|
|
async with client_factory.create() as client:
|
|
await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("spentcheck"),
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
# Check spent invite - should return same error as non-existent
|
|
async with client_factory.create() as client:
|
|
response = await client.get(f"/api/invites/{identifier}/check")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
assert "not found" in data["error"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_invite_revoked_returns_not_found(
|
|
client_factory, admin_user, regular_user
|
|
):
|
|
"""Check endpoint returns same error for revoked invite as for non-existent (no info leakage)."""
|
|
from datetime import UTC, datetime
|
|
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
invite_id = create_resp.json()["id"]
|
|
|
|
# Revoke the invite
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(Invite).where(Invite.id == invite_id))
|
|
invite = result.scalar_one()
|
|
invite.status = InviteStatus.REVOKED
|
|
invite.revoked_at = datetime.now(UTC)
|
|
await db.commit()
|
|
|
|
# Check revoked invite - should return same error as non-existent
|
|
async with client_factory.create() as client:
|
|
response = await client.get(f"/api/invites/{identifier}/check")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
assert "not found" in data["error"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_invite_case_insensitive(client_factory, admin_user, regular_user):
|
|
"""Check endpoint handles case-insensitive identifiers."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Check with uppercase
|
|
async with client_factory.create() as client:
|
|
response = await client.get(f"/api/invites/{identifier.upper()}/check")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["valid"] is True
|
|
|
|
|
|
# ============================================================================
|
|
# Register with Invite Tests (Phase 3)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_valid_invite(client_factory, admin_user, regular_user):
|
|
"""Can register with valid invite code."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
godfather_id = godfather.id
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather_id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Register with invite
|
|
new_email = unique_email("newuser")
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": new_email,
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["email"] == new_email
|
|
assert "regular" in data["roles"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_marks_invite_spent(client_factory, admin_user, regular_user):
|
|
"""Registering marks the invite as SPENT."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
invite_data = create_resp.json()
|
|
identifier = invite_data["identifier"]
|
|
invite_id = invite_data["id"]
|
|
|
|
# Register
|
|
async with client_factory.create() as client:
|
|
await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("spenttest"),
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
# Check invite status
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(Invite).where(Invite.id == invite_id))
|
|
invite = result.scalar_one()
|
|
|
|
assert invite.status == InviteStatus.SPENT
|
|
assert invite.used_by_id is not None
|
|
assert invite.spent_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_sets_godfather(client_factory, admin_user, regular_user):
|
|
"""New user has correct godfather_id."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
godfather_id = godfather.id
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather_id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Register
|
|
new_email = unique_email("godchildtest")
|
|
async with client_factory.create() as client:
|
|
await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": new_email,
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
# Check user's godfather
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(User).where(User.email == new_email))
|
|
new_user = result.scalar_one()
|
|
|
|
assert new_user.godfather_id == godfather_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_invalid_invite(client_factory):
|
|
"""Cannot register with non-existent invite."""
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("invalid"),
|
|
"password": "password123",
|
|
"invite_identifier": "fake-invite-99",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "invalid" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_spent_invite(client_factory, admin_user, regular_user):
|
|
"""Cannot register with already-spent invite."""
|
|
# Create and use invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# First registration
|
|
async with client_factory.create() as client:
|
|
await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("first"),
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
# Second registration with same invite
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("second"),
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "invalid invite code" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_revoked_invite(client_factory, admin_user, regular_user):
|
|
"""Cannot register with revoked invite."""
|
|
from datetime import UTC, datetime
|
|
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
invite_data = create_resp.json()
|
|
identifier = invite_data["identifier"]
|
|
invite_id = invite_data["id"]
|
|
|
|
# Revoke invite directly in DB
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(Invite).where(Invite.id == invite_id))
|
|
invite = result.scalar_one()
|
|
invite.status = InviteStatus.REVOKED
|
|
invite.revoked_at = datetime.now(UTC)
|
|
await db.commit()
|
|
|
|
# Try to register
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("revoked"),
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "invalid invite code" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_duplicate_email(client_factory, admin_user, regular_user):
|
|
"""Cannot register with already-used email."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Try to register with existing email
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": regular_user["email"], # Already exists
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "already registered" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_sets_auth_cookie(client_factory, admin_user, regular_user):
|
|
"""Registration sets auth cookie."""
|
|
# Create invite
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Register
|
|
async with client_factory.create() as client:
|
|
response = await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("cookietest"),
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
assert "auth_token" in response.cookies
|
|
|
|
|
|
# ============================================================================
|
|
# User Invites API Tests (Phase 4)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_regular_user_can_list_invites(client_factory, admin_user, regular_user):
|
|
"""Regular user can list their own invites."""
|
|
# Create invites for the regular user
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
await client.post("/api/admin/invites", json={"godfather_id": godfather.id})
|
|
await client.post("/api/admin/invites", json={"godfather_id": godfather.id})
|
|
|
|
# List invites as regular user
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.get("/api/invites")
|
|
|
|
assert response.status_code == 200
|
|
invites = response.json()
|
|
assert len(invites) == 2
|
|
for invite in invites:
|
|
assert "identifier" in invite
|
|
assert invite["status"] == "ready"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_with_no_invites_gets_empty_list(client_factory, regular_user):
|
|
"""User with no invites gets empty list."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.get("/api/invites")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spent_invite_shows_used_by_email(
|
|
client_factory, admin_user, regular_user
|
|
):
|
|
"""Spent invite shows who used it."""
|
|
# Create invite for regular user
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
identifier = create_resp.json()["identifier"]
|
|
|
|
# Use the invite
|
|
invitee_email = unique_email("invitee")
|
|
async with client_factory.create() as client:
|
|
await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": invitee_email,
|
|
"password": "password123",
|
|
"invite_identifier": identifier,
|
|
},
|
|
)
|
|
|
|
# Check that regular user sees the invitee email
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.get("/api/invites")
|
|
|
|
assert response.status_code == 200
|
|
invites = response.json()
|
|
assert len(invites) == 1
|
|
assert invites[0]["status"] == "spent"
|
|
assert invites[0]["used_by_email"] == invitee_email
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_cannot_list_own_invites(client_factory, admin_user):
|
|
"""Admin without VIEW_OWN_INVITES permission gets 403."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
response = await client.get("/api/invites")
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthenticated_cannot_list_invites(client_factory):
|
|
"""Unauthenticated user gets 401."""
|
|
async with client_factory.create() as client:
|
|
response = await client.get("/api/invites")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
# ============================================================================
|
|
# Admin Invite Management Tests (Phase 5)
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_list_all_invites(client_factory, admin_user, regular_user):
|
|
"""Admin can list all invites."""
|
|
# Create some invites
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
await client.post("/api/admin/invites", json={"godfather_id": godfather.id})
|
|
await client.post("/api/admin/invites", json={"godfather_id": godfather.id})
|
|
|
|
# List all
|
|
response = await client.get("/api/admin/invites")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 2
|
|
assert len(data["records"]) >= 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_list_pagination(client_factory, admin_user, regular_user):
|
|
"""Admin can paginate invite list."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
# Create 5 invites
|
|
for _ in range(5):
|
|
await client.post("/api/admin/invites", json={"godfather_id": godfather.id})
|
|
|
|
# Get page 1 with 2 per page
|
|
response = await client.get("/api/admin/invites?page=1&per_page=2")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["records"]) == 2
|
|
assert data["page"] == 1
|
|
assert data["per_page"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_filter_by_status(client_factory, admin_user, regular_user):
|
|
"""Admin can filter invites by status."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
# Create an invite
|
|
await client.post("/api/admin/invites", json={"godfather_id": godfather.id})
|
|
|
|
# Filter by ready
|
|
response = await client.get("/api/admin/invites?status=ready")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for record in data["records"]:
|
|
assert record["status"] == "ready"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_revoke_invite(client_factory, admin_user, regular_user):
|
|
"""Admin can revoke a READY invite."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
# Create invite
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
invite_id = create_resp.json()["id"]
|
|
|
|
# Revoke it
|
|
response = await client.post(f"/api/admin/invites/{invite_id}/revoke")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "revoked"
|
|
assert data["revoked_at"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cannot_revoke_spent_invite(client_factory, admin_user, regular_user):
|
|
"""Cannot revoke an already-spent invite."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
godfather = result.scalar_one()
|
|
|
|
# Create invite
|
|
create_resp = await client.post(
|
|
"/api/admin/invites",
|
|
json={"godfather_id": godfather.id},
|
|
)
|
|
invite_data = create_resp.json()
|
|
|
|
# Use the invite
|
|
async with client_factory.create() as client:
|
|
await client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": unique_email("spent"),
|
|
"password": "password123",
|
|
"invite_identifier": invite_data["identifier"],
|
|
},
|
|
)
|
|
|
|
# Try to revoke
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
response = await client.post(f"/api/admin/invites/{invite_data['id']}/revoke")
|
|
|
|
assert response.status_code == 400
|
|
assert "only ready" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revoke_nonexistent_invite(client_factory, admin_user):
|
|
"""Revoking non-existent invite returns 404."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
response = await client.post("/api/admin/invites/99999/revoke")
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_regular_user_cannot_access_admin_invites(client_factory, regular_user):
|
|
"""Regular user cannot access admin invite endpoints."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
# List
|
|
response = await client.get("/api/admin/invites")
|
|
assert response.status_code == 403
|
|
|
|
# Revoke
|
|
response = await client.post("/api/admin/invites/1/revoke")
|
|
assert response.status_code == 403
|