feat(deps): add require_active_subscription guard with allowlist

Mounts on Pro routers (trees, sessions, scripts, FlowPilot, etc.) and
returns 402 with structured detail when an account's subscription is
missing or locked. Allowlist bypasses billing/account/auth flows so
users can recover from a lapsed subscription.

Conftest now seeds a default Pro/active Subscription on test_user and
test_admin (delete-then-insert because the register endpoint already
creates a free/active sub by default). Two existing tests adapted to
the new seeded plan; tenant-isolation tests seed Subscription rows for
the accounts they create directly.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-06 14:35:59 -04:00
parent cfe0e6cae6
commit 9ec208f6e7
7 changed files with 245 additions and 29 deletions

View File

@@ -248,13 +248,23 @@ async def client(test_db: AsyncSession):
@pytest.fixture
async def test_user(client):
async def test_user(client, test_db):
"""
Create a test user and return their credentials.
Also seeds a default active Pro Subscription so Pro-guarded routes work
in tests. Phase 1 Task 11 added require_active_subscription; without
this seed every existing test that hits a Pro router would 402. The
register endpoint creates a default `free`/`active` Subscription, so
we delete-then-insert to avoid the unique account_id constraint.
Returns:
dict with email, password, and user_data
"""
import uuid
from sqlalchemy import delete
from app.models.subscription import Subscription
user_data = {
"email": "test@example.com",
"password": "TestPassword123!",
@@ -264,6 +274,13 @@ async def test_user(client):
response = await client.post("/api/v1/auth/register", json=user_data)
assert response.status_code == 200 or response.status_code == 201
account_id = uuid.UUID(response.json()["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="active"))
await test_db.commit()
return {
"email": user_data["email"],
"password": user_data["password"],
@@ -346,11 +363,14 @@ async def test_admin(client, test_db):
Create a test super-admin user.
Registers as engineer (the only role available at registration),
then promotes to super_admin directly via the DB session.
then promotes to super_admin directly via the DB session. Also
seeds a default active Pro Subscription (see test_user docstring).
"""
import uuid
from uuid import UUID as PyUUID
from sqlalchemy import select
from sqlalchemy import select, delete
from app.models.user import User
from app.models.subscription import Subscription
admin_data = {
"email": "admin@example.com",
@@ -365,6 +385,12 @@ async def test_admin(client, test_db):
result = await test_db.execute(select(User).where(User.id == user_id))
user = result.scalar_one()
user.is_super_admin = True
account_id = uuid.UUID(response.json()["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="active"))
await test_db.commit()
return {