50cb0fc7f0
- Migration 030: add email, assigned_plan, trial_duration_days, email_sent_at
to invite_codes with CHECK constraints
- Resend email integration (graceful degradation when API key not set)
- Invite codes now support plan assignment (free/pro/team) and trial duration (1-90 days)
- Registration applies invite code plan/trial to new subscription
- Auto-downgrade expired trials on authenticated access
- Enriched GET /admin/users/{id} with account, subscription, sessions, audit logs
- New endpoints: PUT /admin/users/{id}/subscription/plan and extend-trial
- Frontend: enhanced invite codes page with email, plan, trial fields
- Frontend: new user detail page at /admin/users/:userId
- Fixed API path drift: /invite-codes -> /invites
- 11 new backend tests, 416 total passing
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
228 lines
7.7 KiB
Python
228 lines
7.7 KiB
Python
"""Tests for enhanced invite codes with plan assignment and trial durations."""
|
|
|
|
import pytest
|
|
from datetime import datetime, timezone, timedelta
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.models.invite_code import InviteCode
|
|
from app.models.subscription import Subscription
|
|
from app.models.user import User
|
|
|
|
|
|
class TestInviteCodeCreation:
|
|
"""Test invite code creation with plan/trial fields."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invite_with_plan(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "pro", "note": "Beta tester"},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["assigned_plan"] == "pro"
|
|
assert data["has_trial"] is False
|
|
assert data["trial_duration_days"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invite_with_trial(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "pro", "trial_duration_days": 14},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["assigned_plan"] == "pro"
|
|
assert data["trial_duration_days"] == 14
|
|
assert data["has_trial"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_invite_with_email(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "team", "email": "beta@example.com"},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["email"] == "beta@example.com"
|
|
# Email not sent because RESEND_API_KEY not configured
|
|
assert data["email_sent"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_free_plan_rejects_trial(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "free", "trial_duration_days": 14},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trial_duration_bounds(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
# Too low
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "pro", "trial_duration_days": 0},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
# Too high
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "pro", "trial_duration_days": 91},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_plan_is_free(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
response = await client.post(
|
|
"/api/v1/invites",
|
|
json={},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 201
|
|
assert response.json()["assigned_plan"] == "free"
|
|
|
|
|
|
class TestRegistrationWithInvitePlan:
|
|
"""Test that registration applies invite code plan/trial to subscription."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_pro_trial_invite(
|
|
self, client: AsyncClient, admin_auth_headers: dict, test_db: AsyncSession
|
|
):
|
|
# Create a pro trial invite
|
|
resp = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "pro", "trial_duration_days": 14},
|
|
headers=admin_auth_headers,
|
|
)
|
|
code = resp.json()["code"]
|
|
|
|
# Register with the invite code
|
|
reg_resp = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "trial_user@example.com",
|
|
"password": "SecurePass1",
|
|
"name": "Trial User",
|
|
"invite_code": code,
|
|
},
|
|
)
|
|
assert reg_resp.status_code == 201
|
|
user_id = reg_resp.json()["id"]
|
|
|
|
# Check subscription
|
|
user = (await test_db.execute(
|
|
select(User).where(User.id == user_id)
|
|
)).scalar_one()
|
|
sub = (await test_db.execute(
|
|
select(Subscription).where(Subscription.account_id == user.account_id)
|
|
)).scalar_one()
|
|
assert sub.plan == "pro"
|
|
assert sub.status == "trialing"
|
|
assert sub.current_period_end is not None
|
|
assert sub.current_period_end > datetime.now(timezone.utc)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_with_team_no_trial(
|
|
self, client: AsyncClient, admin_auth_headers: dict, test_db: AsyncSession
|
|
):
|
|
# Create team invite without trial
|
|
resp = await client.post(
|
|
"/api/v1/invites",
|
|
json={"assigned_plan": "team"},
|
|
headers=admin_auth_headers,
|
|
)
|
|
code = resp.json()["code"]
|
|
|
|
reg_resp = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "team_user@example.com",
|
|
"password": "SecurePass1",
|
|
"name": "Team User",
|
|
"invite_code": code,
|
|
},
|
|
)
|
|
assert reg_resp.status_code == 201
|
|
user_id = reg_resp.json()["id"]
|
|
|
|
user = (await test_db.execute(
|
|
select(User).where(User.id == user_id)
|
|
)).scalar_one()
|
|
sub = (await test_db.execute(
|
|
select(Subscription).where(Subscription.account_id == user.account_id)
|
|
)).scalar_one()
|
|
assert sub.plan == "team"
|
|
assert sub.status == "active"
|
|
|
|
|
|
class TestAdminSubscriptionManagement:
|
|
"""Test admin subscription plan change and trial extension endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_change_user_plan(
|
|
self, client: AsyncClient, admin_auth_headers: dict, test_user: dict
|
|
):
|
|
user_id = test_user["user_data"]["id"]
|
|
response = await client.put(
|
|
f"/api/v1/admin/users/{user_id}/subscription/plan",
|
|
json={"plan": "pro"},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["plan"] == "pro"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extend_trial(
|
|
self, client: AsyncClient, admin_auth_headers: dict, test_user: dict
|
|
):
|
|
user_id = test_user["user_data"]["id"]
|
|
response = await client.put(
|
|
f"/api/v1/admin/users/{user_id}/subscription/extend-trial",
|
|
json={"days": 14},
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "trialing"
|
|
assert data["current_period_end"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enriched_user_detail(
|
|
self, client: AsyncClient, admin_auth_headers: dict, test_user: dict
|
|
):
|
|
user_id = test_user["user_data"]["id"]
|
|
response = await client.get(
|
|
f"/api/v1/admin/users/{user_id}",
|
|
headers=admin_auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# Should have enriched fields
|
|
assert "subscription" in data
|
|
assert "account" in data
|
|
assert "recent_sessions" in data
|
|
assert "total_sessions" in data
|
|
assert "recent_audit_logs" in data
|
|
assert "total_audit_logs" in data
|