46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
import uuid
|
|
import pytest
|
|
from datetime import datetime, timezone, timedelta
|
|
from sqlalchemy import select
|
|
from app.models.subscription import Subscription
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expired_trial_is_not_mutated_by_get_current_active_user(
|
|
test_db, client, test_user, auth_headers
|
|
):
|
|
"""The previous deps.py:109 logic mutated trialing→active+free on expiry.
|
|
That's gone. An expired-trial Subscription should retain status='trialing'
|
|
and current_period_end after any authenticated request."""
|
|
account_id = uuid.UUID(test_user["user_data"]["account_id"])
|
|
|
|
# If a Subscription already exists for this account (e.g. created by
|
|
# the register handler), update it; otherwise insert a new one.
|
|
existing = await test_db.execute(
|
|
select(Subscription).where(Subscription.account_id == account_id)
|
|
)
|
|
sub = existing.scalar_one_or_none()
|
|
expired_end = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
if sub is None:
|
|
sub = Subscription(
|
|
account_id=account_id,
|
|
plan="pro",
|
|
status="trialing",
|
|
current_period_end=expired_end,
|
|
)
|
|
test_db.add(sub)
|
|
else:
|
|
sub.plan = "pro"
|
|
sub.status = "trialing"
|
|
sub.current_period_end = expired_end
|
|
await test_db.commit()
|
|
|
|
# Call any authenticated endpoint that goes through get_current_active_user.
|
|
response = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
await test_db.refresh(sub)
|
|
assert sub.status == "trialing"
|
|
assert sub.plan == "pro"
|
|
assert sub.current_period_end is not None
|