Files
gh-christianlouis-docuelevate/tests/test_billing.py
T
copilot-swe-agent[bot] 52e3852129 feat(auth): add local user signup, email verification, and Stripe billing
- Add LocalUser model with bcrypt password hashing, email verification
  tokens, and password reset tokens
- Add ALLOW_LOCAL_SIGNUP config flag (requires SMTP to be configured)
- Add Stripe billing config fields (STRIPE_SECRET_KEY, etc.)
- Add stripe_customer_id to UserProfile and stripe_price_id_monthly/
  stripe_price_id_yearly to SubscriptionPlan
- Create migration 018_add_local_users_and_billing
- Add app/utils/local_auth.py: hash_password, verify_password,
  generate_token, is_token_expired, send_verification_email,
  send_password_reset_email, build_session_user
- Add app/api/local_auth.py: signup, email verification, password reset
  endpoints plus signup/verify-email-sent/reset-password page routes
- Add app/api/billing.py: Stripe Checkout, Customer Portal, and webhook
  endpoints; syncs subscription tier from webhook events
- Update auth() to check LocalUser table before admin credentials fallback
- Update login() to pass allow_signup context variable to template
- Add signup.html, verify_email_sent.html, password_reset_form.html,
  billing_success.html templates (Alpine.js, Tailwind, WCAG 2.1 AA)
- Update login.html to show 'Create account' link when signup enabled
- Update pricing.html CTA buttons to use Stripe Checkout for paid tiers
- Add docs/BillingSetup.md with setup guide, webhook config, compliance
- Add tests/test_local_auth.py (42 tests) and tests/test_billing.py
  (31 tests); all 106 tests in the modified test suite pass
- Add stripe>=7.0.0,<15.0.0 to requirements.txt

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-03-07 13:15:15 +00:00

487 lines
17 KiB
Python

"""Tests for the Stripe billing API endpoints.
Covers:
- POST /api/billing/create-checkout-session
- POST /api/billing/create-portal-session
- POST /api/billing/webhook (all event types)
- GET /api/billing/success
- Internal helpers: _handle_stripe_event, _on_checkout_completed,
_on_subscription_updated, _on_subscription_deleted
"""
import json
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.api.billing import (
_handle_stripe_event,
_on_checkout_completed,
_on_subscription_deleted,
_on_subscription_updated,
_resolve_plan_id_from_price,
_resolve_user_id_from_customer,
)
from app.database import Base, get_db
from app.models import SubscriptionPlan, UserProfile
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def bill_engine():
"""In-memory SQLite engine for billing tests."""
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(bind=engine)
yield engine
Base.metadata.drop_all(bind=engine)
@pytest.fixture()
def bill_session(bill_engine):
"""DB session for one test."""
Session = sessionmaker(bind=bill_engine)
session = Session()
yield session
session.close()
@pytest.fixture()
def bill_client(bill_engine):
"""TestClient with DB dependency overridden and a logged-in session."""
from app.main import app
Session = sessionmaker(bind=bill_engine)
def override_get_db():
db = Session()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
yield client
app.dependency_overrides.pop(get_db, None)
@pytest.fixture()
def starter_plan(bill_session):
"""A SubscriptionPlan with Stripe price IDs in the DB."""
plan = SubscriptionPlan(
plan_id="starter",
name="Starter",
price_monthly=9.0,
price_yearly=90.0,
trial_days=0,
stripe_price_id_monthly="price_monthly_starter",
stripe_price_id_yearly="price_yearly_starter",
)
bill_session.add(plan)
bill_session.commit()
return plan
@pytest.fixture()
def user_profile(bill_session):
"""A UserProfile for user1@example.com."""
profile = UserProfile(
user_id="user1@example.com",
display_name="Test User",
stripe_customer_id=None,
)
bill_session.add(profile)
bill_session.commit()
return profile
# ---------------------------------------------------------------------------
# Tests: _get_stripe returns None when not configured
# ---------------------------------------------------------------------------
@pytest.mark.unit
def test_get_stripe_returns_none_when_not_configured():
"""_get_stripe returns None when stripe_secret_key is not set."""
from app.api.billing import _get_stripe
with patch("app.api.billing.settings") as mock_settings:
mock_settings.stripe_secret_key = None
result = _get_stripe()
assert result is None
@pytest.mark.unit
def test_get_stripe_returns_client_when_configured():
"""_get_stripe returns a StripeClient when key is configured."""
from app.api.billing import _get_stripe
with patch("app.api.billing.settings") as mock_settings:
mock_settings.stripe_secret_key = "sk_test_fake"
result = _get_stripe()
assert result is not None
# ---------------------------------------------------------------------------
# Tests: create-checkout-session
# ---------------------------------------------------------------------------
@pytest.mark.integration
def test_create_checkout_session_stripe_not_configured(bill_client):
"""POST /api/billing/create-checkout-session returns 503 when Stripe not set."""
with patch("app.api.billing._get_stripe", return_value=None):
resp = bill_client.post(
"/api/billing/create-checkout-session",
json={"plan_id": "starter", "billing_cycle": "monthly"},
)
assert resp.status_code == 503
@pytest.mark.integration
def test_create_checkout_session_plan_not_found(bill_client):
"""POST /api/billing/create-checkout-session returns 404 for unknown plan."""
mock_client = MagicMock()
with patch("app.api.billing._get_stripe", return_value=mock_client):
resp = bill_client.post(
"/api/billing/create-checkout-session",
json={"plan_id": "nonexistent", "billing_cycle": "monthly"},
)
assert resp.status_code == 404
@pytest.mark.integration
def test_create_checkout_session_no_price_id(bill_client, bill_session):
"""POST /api/billing/create-checkout-session returns 404 when price ID not set."""
plan = SubscriptionPlan(
plan_id="noprice",
name="No Price",
price_monthly=5.0,
price_yearly=50.0,
trial_days=0,
stripe_price_id_monthly=None,
stripe_price_id_yearly=None,
)
bill_session.add(plan)
bill_session.commit()
mock_client = MagicMock()
with patch("app.api.billing._get_stripe", return_value=mock_client):
resp = bill_client.post(
"/api/billing/create-checkout-session",
json={"plan_id": "noprice", "billing_cycle": "monthly"},
)
assert resp.status_code == 404
@pytest.mark.integration
def test_create_checkout_session_success(bill_client, starter_plan, user_profile):
"""POST /api/billing/create-checkout-session returns checkout_url on success."""
mock_client = MagicMock()
mock_customer = MagicMock()
mock_customer.id = "cus_test123"
mock_session = MagicMock()
mock_session.id = "cs_test456"
mock_session.url = "https://checkout.stripe.com/test"
mock_client.customers.create.return_value = mock_customer
mock_client.checkout.sessions.create.return_value = mock_session
with (
patch("app.api.billing._get_stripe", return_value=mock_client),
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
):
resp = bill_client.post(
"/api/billing/create-checkout-session",
json={"plan_id": "starter", "billing_cycle": "monthly"},
)
assert resp.status_code == 200
data = resp.json()
assert "checkout_url" in data
assert data["checkout_url"] == "https://checkout.stripe.com/test"
@pytest.mark.integration
def test_create_checkout_session_yearly(bill_client, starter_plan, user_profile):
"""POST /api/billing/create-checkout-session uses yearly price ID for yearly cycle."""
mock_client = MagicMock()
mock_customer = MagicMock()
mock_customer.id = "cus_test123"
mock_session = MagicMock()
mock_session.id = "cs_test456"
mock_session.url = "https://checkout.stripe.com/yearly"
mock_client.customers.create.return_value = mock_customer
mock_client.checkout.sessions.create.return_value = mock_session
with (
patch("app.api.billing._get_stripe", return_value=mock_client),
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
):
resp = bill_client.post(
"/api/billing/create-checkout-session",
json={"plan_id": "starter", "billing_cycle": "yearly"},
)
assert resp.status_code == 200
# Verify yearly price ID was used
call_params = mock_client.checkout.sessions.create.call_args[1]["params"]
assert call_params["line_items"][0]["price"] == "price_yearly_starter"
# ---------------------------------------------------------------------------
# Tests: create-portal-session
# ---------------------------------------------------------------------------
@pytest.mark.integration
def test_create_portal_session_stripe_not_configured(bill_client):
"""POST /api/billing/create-portal-session returns 503 when not configured."""
with patch("app.api.billing._get_stripe", return_value=None):
resp = bill_client.post("/api/billing/create-portal-session", json={})
assert resp.status_code == 503
@pytest.mark.integration
def test_create_portal_session_no_customer(bill_client, user_profile):
"""POST /api/billing/create-portal-session returns 404 when no Stripe customer."""
mock_client = MagicMock()
with (
patch("app.api.billing._get_stripe", return_value=mock_client),
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
):
resp = bill_client.post("/api/billing/create-portal-session", json={})
assert resp.status_code == 404
@pytest.mark.integration
def test_create_portal_session_success(bill_client, bill_session, user_profile):
"""POST /api/billing/create-portal-session returns portal_url on success."""
user_profile.stripe_customer_id = "cus_existing"
bill_session.commit()
mock_client = MagicMock()
mock_portal = MagicMock()
mock_portal.url = "https://billing.stripe.com/portal/test"
mock_client.billing_portal.sessions.create.return_value = mock_portal
with (
patch("app.api.billing._get_stripe", return_value=mock_client),
patch("app.api.billing.get_current_owner_id", return_value="user1@example.com"),
):
resp = bill_client.post("/api/billing/create-portal-session", json={})
assert resp.status_code == 200
assert resp.json()["portal_url"] == "https://billing.stripe.com/portal/test"
# ---------------------------------------------------------------------------
# Tests: webhook
# ---------------------------------------------------------------------------
@pytest.mark.integration
def test_webhook_not_configured(bill_client):
"""POST /api/billing/webhook returns 503 when billing not configured."""
with patch("app.api.billing.settings") as mock_settings:
mock_settings.stripe_secret_key = None
mock_settings.stripe_webhook_secret = None
resp = bill_client.post(
"/api/billing/webhook",
content=b"{}",
headers={"content-type": "application/json"},
)
assert resp.status_code == 503
@pytest.mark.integration
def test_webhook_invalid_signature(bill_client):
"""POST /api/billing/webhook returns 400 on invalid Stripe signature."""
import stripe
with (
patch("app.api.billing.settings") as mock_settings,
patch("stripe.Webhook.construct_event", side_effect=stripe.SignatureVerificationError("bad", "sig")),
):
mock_settings.stripe_secret_key = "sk_test_fake"
mock_settings.stripe_webhook_secret = "whsec_test"
resp = bill_client.post(
"/api/billing/webhook",
content=b'{"type":"test"}',
headers={"stripe-signature": "bad_sig", "content-type": "application/json"},
)
assert resp.status_code == 400
@pytest.mark.integration
def test_webhook_checkout_completed(bill_client, bill_session, starter_plan, user_profile):
"""POST /api/billing/webhook activates plan on checkout.session.completed."""
payload = json.dumps(
{
"type": "checkout.session.completed",
"data": {
"object": {
"customer": "cus_new",
"metadata": {
"docuelevate_user_id": "user1@example.com",
"plan_id": "starter",
"billing_cycle": "monthly",
},
}
},
}
).encode()
with (
patch("app.api.billing.settings") as mock_settings,
patch("stripe.Event.construct_from", return_value=json.loads(payload)),
):
mock_settings.stripe_secret_key = "sk_test_fake"
mock_settings.stripe_webhook_secret = None
resp = bill_client.post(
"/api/billing/webhook",
content=payload,
headers={"content-type": "application/json"},
)
assert resp.status_code == 200
bill_session.expire_all()
profile = bill_session.query(UserProfile).filter(UserProfile.user_id == "user1@example.com").first()
assert profile.subscription_tier == "starter"
@pytest.mark.integration
def test_webhook_subscription_deleted(bill_client, bill_session, user_profile):
"""POST /api/billing/webhook downgrades to free on subscription deleted."""
user_profile.stripe_customer_id = "cus_del"
user_profile.subscription_tier = "starter"
bill_session.commit()
payload = json.dumps(
{
"type": "customer.subscription.deleted",
"data": {"object": {"customer": "cus_del"}},
}
).encode()
with (
patch("app.api.billing.settings") as mock_settings,
patch("stripe.Event.construct_from", return_value=json.loads(payload)),
):
mock_settings.stripe_secret_key = "sk_test_fake"
mock_settings.stripe_webhook_secret = None
resp = bill_client.post(
"/api/billing/webhook",
content=payload,
headers={"content-type": "application/json"},
)
assert resp.status_code == 200
bill_session.expire_all()
profile = bill_session.query(UserProfile).filter(UserProfile.user_id == "user1@example.com").first()
assert profile.subscription_tier == "free"
# ---------------------------------------------------------------------------
# Unit tests: internal helpers
# ---------------------------------------------------------------------------
@pytest.mark.unit
def test_resolve_user_id_from_customer(bill_session, user_profile):
"""_resolve_user_id_from_customer returns user_id for known Stripe customer."""
user_profile.stripe_customer_id = "cus_known"
bill_session.commit()
result = _resolve_user_id_from_customer(bill_session, "cus_known")
assert result == "user1@example.com"
@pytest.mark.unit
def test_resolve_user_id_from_customer_unknown(bill_session):
"""_resolve_user_id_from_customer returns None for unknown customer."""
result = _resolve_user_id_from_customer(bill_session, "cus_unknown")
assert result is None
@pytest.mark.unit
def test_resolve_plan_id_from_price(bill_session, starter_plan):
"""_resolve_plan_id_from_price finds plan by monthly price ID."""
result = _resolve_plan_id_from_price(bill_session, "price_monthly_starter")
assert result == "starter"
@pytest.mark.unit
def test_resolve_plan_id_from_price_yearly(bill_session, starter_plan):
"""_resolve_plan_id_from_price finds plan by yearly price ID."""
result = _resolve_plan_id_from_price(bill_session, "price_yearly_starter")
assert result == "starter"
@pytest.mark.unit
def test_resolve_plan_id_from_price_unknown(bill_session):
"""_resolve_plan_id_from_price returns None for unknown price."""
result = _resolve_plan_id_from_price(bill_session, "price_unknown")
assert result is None
@pytest.mark.unit
def test_on_checkout_completed_missing_user_id(bill_session):
"""_on_checkout_completed does nothing when user_id is absent."""
data = {"metadata": {}, "customer": "cus_test"}
_on_checkout_completed(bill_session, data) # Should not raise
@pytest.mark.unit
def test_on_subscription_updated_no_items(bill_session, user_profile):
"""_on_subscription_updated does nothing when items list is empty."""
user_profile.stripe_customer_id = "cus_upd"
bill_session.commit()
data = {"customer": "cus_upd", "items": {"data": []}}
_on_subscription_updated(bill_session, data) # Should not raise
@pytest.mark.unit
def test_on_subscription_deleted_unknown_customer(bill_session):
"""_on_subscription_deleted does nothing for unknown customer."""
data = {"customer": "cus_nobody"}
_on_subscription_deleted(bill_session, data) # Should not raise
@pytest.mark.unit
def test_handle_stripe_event_unhandled_type(bill_session):
"""_handle_stripe_event logs but does not raise for unknown event types."""
event = {"type": "unknown.event.type", "data": {"object": {}}}
_handle_stripe_event(bill_session, event) # Should not raise
@pytest.mark.unit
def test_handle_stripe_event_payment_failed(bill_session):
"""_handle_stripe_event handles invoice.payment_failed without raising."""
event = {
"type": "invoice.payment_failed",
"data": {"object": {"customer": "cus_fail"}},
}
_handle_stripe_event(bill_session, event) # Should not raise
# ---------------------------------------------------------------------------
# Tests: billing success page
# ---------------------------------------------------------------------------
@pytest.mark.integration
def test_billing_success_page(bill_client):
"""GET /api/billing/success returns 200 for logged-in user."""
resp = bill_client.get("/api/billing/success")
assert resp.status_code == 200
assert b"subscription" in resp.content.lower() or b"success" in resp.content.lower()