86 lines
3.4 KiB
Python
86 lines
3.4 KiB
Python
"""Smoke test for the complimentary backfill: assertions about the post-state.
|
|
The actual migration runs at deploy time; tests use create_all so the
|
|
migration body isn't executed automatically. We invoke the SQL inline to
|
|
exercise the same effect."""
|
|
import uuid
|
|
import pytest
|
|
from sqlalchemy import select, text, delete
|
|
from app.models.account import Account
|
|
from app.models.subscription import Subscription
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complimentary_backfill_sets_status_and_inserts_missing_rows(test_db):
|
|
"""Inline-run the backfill SQL and assert post-state."""
|
|
# Seed a fresh account with no subscription
|
|
no_sub_account = Account(name="NoSub", display_code="NOSUB001")
|
|
test_db.add(no_sub_account)
|
|
await test_db.flush()
|
|
|
|
# Seed an account with a trialing subscription (should become complimentary)
|
|
trial_account = Account(name="Trial", display_code="TRIAL001")
|
|
test_db.add(trial_account)
|
|
await test_db.flush()
|
|
test_db.add(Subscription(
|
|
account_id=trial_account.id, plan="free", status="trialing",
|
|
))
|
|
|
|
# Seed an account with a canceled subscription (should be preserved)
|
|
canceled_account = Account(name="Cancel", display_code="CANCL001")
|
|
test_db.add(canceled_account)
|
|
await test_db.flush()
|
|
test_db.add(Subscription(
|
|
account_id=canceled_account.id, plan="pro", status="canceled",
|
|
))
|
|
await test_db.commit()
|
|
|
|
# Run the same SQL the migration runs
|
|
await test_db.execute(text("""
|
|
UPDATE subscriptions
|
|
SET status = 'complimentary', plan = 'pro',
|
|
current_period_end = NULL, current_period_start = NULL,
|
|
updated_at = now()
|
|
WHERE status NOT IN ('canceled', 'past_due')
|
|
"""))
|
|
await test_db.execute(text("""
|
|
INSERT INTO subscriptions (id, account_id, plan, status, cancel_at_period_end, created_at, updated_at)
|
|
SELECT gen_random_uuid(), a.id, 'pro', 'complimentary', false, now(), now()
|
|
FROM accounts a
|
|
WHERE NOT EXISTS (SELECT 1 FROM subscriptions s WHERE s.account_id = a.id)
|
|
"""))
|
|
await test_db.commit()
|
|
|
|
# All three accounts now have a Subscription
|
|
no_sub_row = (await test_db.execute(
|
|
select(Subscription).where(Subscription.account_id == no_sub_account.id)
|
|
)).scalar_one()
|
|
assert no_sub_row.status == "complimentary"
|
|
assert no_sub_row.plan == "pro"
|
|
|
|
trial_row = (await test_db.execute(
|
|
select(Subscription).where(Subscription.account_id == trial_account.id)
|
|
)).scalar_one()
|
|
assert trial_row.status == "complimentary"
|
|
assert trial_row.plan == "pro"
|
|
|
|
canceled_row = (await test_db.execute(
|
|
select(Subscription).where(Subscription.account_id == canceled_account.id)
|
|
)).scalar_one()
|
|
# Canceled is preserved
|
|
assert canceled_row.status == "canceled"
|
|
assert canceled_row.plan == "pro"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complimentary_subscription_passes_active_subscription_guard(
|
|
client, test_db, test_user, auth_headers
|
|
):
|
|
"""The require_active_subscription guard accepts complimentary status."""
|
|
account_id = uuid.UUID(test_user["user_data"]["account_id"])
|
|
await test_db.execute(delete(Subscription).where(Subscription.account_id == account_id))
|
|
test_db.add(Subscription(account_id=account_id, plan="pro", status="complimentary"))
|
|
await test_db.commit()
|
|
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert response.status_code != 402
|