Files
resolutionflow/backend/tests/test_plans_public.py
Michael Chihlas ba36c47075 feat(billing): reconcile plan taxonomy and add Stripe sync script
The marketing surface (PricingPage, Stripe products) was wired for
"Starter / Pro / Enterprise" while the backend was on "free / pro / team",
leaving plan_billing unseeded and BillingPlan accepting a literal that
violated the FK to plan_limits.

This change:

- Migration 4ce3e594cb87: defensive UPDATE of any subscriptions on
  plan='team' to 'enterprise' (dev has zero), renames the plan_limits
  row team -> enterprise, inserts a starter row with caps interpolated
  between free and pro (max_trees=10, sessions=75, ai=15/mo).
- Renames the plan tier across schemas (invite_code, billing, admin,
  subscription comment), is_paid/has_pro_entitlement checks in the
  Subscription model, admin/admin_dashboard plan validators, and the
  frontend useSubscription isPaidPlan check. Resource visibility uses
  the same string 'team' in a separate domain (Tree/StepLibrary
  visibility) and is intentionally untouched.
- New backend/scripts/sync_stripe_plan_ids.py: idempotent upsert of
  plan_billing rows from Stripe products by exact name match. Picks
  the active monthly recurring price for tiers that have one; leaves
  annual fields NULL by design. Works against test or live keys.
- Test fixture updates: conftest seeds the new taxonomy, the public
  plans helper is a true upsert so tests can override max_users, and
  team -> enterprise across test_admin_plan_limits and test_invite_plan.

Verified: 86/86 passing across the subscription/billing/plan/invite/
admin sweep; sync script run against test mode populates plan_billing
correctly for all three tiers.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-07 15:59:42 -04:00

140 lines
5.0 KiB
Python

"""Integration tests for the public plans endpoint.
Covers GET /api/v1/plans/public — the marketing /pricing page data source.
"""
from __future__ import annotations
import pytest
from httpx import AsyncClient
from sqlalchemy import delete
from app.models.plan_billing import PlanBilling
from app.models.plan_limits import PlanLimits
async def _seed_plan_limits(test_db, plan: str, max_users: int | None) -> None:
"""Ensure a plan_limits row exists with the given max_users.
Upserts: conftest seeds the canonical plans (free/starter/pro/enterprise)
so this helper has to overwrite max_users when a test wants different
values for fixture-driven assertions.
"""
existing = await test_db.get(PlanLimits, plan)
if existing is None:
test_db.add(
PlanLimits(
plan=plan,
max_trees=None,
max_sessions_per_month=None,
max_users=max_users,
custom_branding=False,
priority_support=False,
export_formats=["markdown", "text"],
)
)
else:
existing.max_users = max_users
await test_db.commit()
class TestGetPlansPublic:
"""GET /api/v1/plans/public — anonymous, no auth."""
@pytest.mark.asyncio
async def test_get_plans_public_returns_only_is_public_rows(
self, client: AsyncClient, test_db
):
"""Rows with is_public=False or is_archived=True must NOT appear."""
# Wipe any existing billing rows so this test owns the fixture state.
await test_db.execute(delete(PlanBilling))
await test_db.commit()
await _seed_plan_limits(test_db, "starter", 3)
await _seed_plan_limits(test_db, "pro", 10)
await _seed_plan_limits(test_db, "internal", None)
await _seed_plan_limits(test_db, "legacy", 5)
test_db.add_all(
[
PlanBilling(
plan="starter",
display_name="Starter",
monthly_price_cents=1900,
is_public=True,
is_archived=False,
sort_order=10,
),
PlanBilling(
plan="pro",
display_name="Pro",
monthly_price_cents=4900,
is_public=True,
is_archived=False,
sort_order=20,
),
PlanBilling(
plan="internal",
display_name="Internal",
is_public=False, # hidden
is_archived=False,
sort_order=30,
),
PlanBilling(
plan="legacy",
display_name="Legacy",
is_public=True,
is_archived=True, # archived
sort_order=40,
),
]
)
await test_db.commit()
response = await client.get("/api/v1/plans/public")
assert response.status_code == 200
plans = response.json()
plan_names = {p["plan"] for p in plans}
assert "starter" in plan_names
assert "pro" in plan_names
assert "internal" not in plan_names
assert "legacy" not in plan_names
# Schema sanity check
starter = next(p for p in plans if p["plan"] == "starter")
assert starter["display_name"] == "Starter"
assert starter["monthly_price_cents"] == 1900
assert starter["max_seats"] == 3
assert starter["is_public"] is True
@pytest.mark.asyncio
async def test_get_plans_public_orders_by_sort_order_then_plan(
self, client: AsyncClient, test_db
):
"""Result must be ordered by sort_order ASC, then plan name ASC."""
await test_db.execute(delete(PlanBilling))
await test_db.commit()
# plan_limits rows for FK satisfaction
for name in ("alpha", "bravo", "charlie", "delta"):
await _seed_plan_limits(test_db, name, None)
# Two with sort_order=10 (charlie should come before delta by plan ASC),
# one with sort_order=5 (alpha first overall),
# one with sort_order=20 (bravo last).
test_db.add_all(
[
PlanBilling(plan="charlie", display_name="C", sort_order=10, is_public=True, is_archived=False),
PlanBilling(plan="delta", display_name="D", sort_order=10, is_public=True, is_archived=False),
PlanBilling(plan="alpha", display_name="A", sort_order=5, is_public=True, is_archived=False),
PlanBilling(plan="bravo", display_name="B", sort_order=20, is_public=True, is_archived=False),
]
)
await test_db.commit()
response = await client.get("/api/v1/plans/public")
assert response.status_code == 200
ordered = [p["plan"] for p in response.json()]
assert ordered == ["alpha", "charlie", "delta", "bravo"]