88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
import uuid
|
|
import pytest
|
|
from datetime import datetime, timezone, timedelta
|
|
from sqlalchemy import select
|
|
from app.models.user import User
|
|
|
|
|
|
async def _set_user_email_state(test_db, user_id, *, verified_at=None, created_at=None):
|
|
user = (await test_db.execute(select(User).where(User.id == user_id))).scalar_one()
|
|
user.email_verified_at = verified_at
|
|
if created_at is not None:
|
|
user.created_at = created_at
|
|
await test_db.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verified_user_passes(client, test_db, test_user, auth_headers):
|
|
user_id = uuid.UUID(test_user["user_data"]["id"])
|
|
await _set_user_email_state(test_db, user_id, verified_at=datetime.now(timezone.utc))
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert response.status_code != 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unverified_in_grace_passes(client, test_db, test_user, auth_headers):
|
|
user_id = uuid.UUID(test_user["user_data"]["id"])
|
|
await _set_user_email_state(
|
|
test_db, user_id,
|
|
verified_at=None,
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=2),
|
|
)
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert response.status_code != 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unverified_past_grace_blocks(client, test_db, test_user, auth_headers):
|
|
user_id = uuid.UUID(test_user["user_data"]["id"])
|
|
await _set_user_email_state(
|
|
test_db, user_id,
|
|
verified_at=None,
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=10),
|
|
)
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert response.status_code == 403
|
|
body = response.json()
|
|
assert body["detail"]["error"] == "email_not_verified"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unverified_past_grace_allowlisted_still_passes(client, test_db, test_user, auth_headers):
|
|
user_id = uuid.UUID(test_user["user_data"]["id"])
|
|
await _set_user_email_state(
|
|
test_db, user_id,
|
|
verified_at=None,
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=10),
|
|
)
|
|
response = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_combined_guards_unverified_expired_trial(client, test_db, test_user, auth_headers):
|
|
"""A user who is BOTH past grace AND on an expired trial should get blocked
|
|
by one of the two guards. Either error is acceptable; we just verify a
|
|
refusal."""
|
|
from app.models.subscription import Subscription
|
|
from sqlalchemy import delete
|
|
|
|
user_id = uuid.UUID(test_user["user_data"]["id"])
|
|
account_id = uuid.UUID(test_user["user_data"]["account_id"])
|
|
await _set_user_email_state(
|
|
test_db, user_id,
|
|
verified_at=None,
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=10),
|
|
)
|
|
|
|
# Replace the seeded active sub with an expired trial
|
|
await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
|
|
test_db.add(Subscription(
|
|
account_id=account_id, plan="pro", status="trialing",
|
|
current_period_end=datetime.now(timezone.utc) - timedelta(hours=1),
|
|
))
|
|
await test_db.commit()
|
|
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert response.status_code in (402, 403)
|