80c2f111d2
- Fix unused imports in stripe.py (hashlib, hmac, time) - Update test_billing_routes.py: insert into payment_products table, fix mock paths for extracted paddle.py, add Stripe webhook 404 test - Update CHANGELOG.md with Stripe provider feature - Update PROJECT.md Done section Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""
|
|
Route integration tests for billing endpoints.
|
|
Tests work with the default provider (Paddle) via dispatch layer.
|
|
Checkout returns JSON (Paddle overlay or Stripe redirect URL), manage/cancel use provider SDK.
|
|
"""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
CHECKOUT_PLAN = "starter"
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# Public routes (pricing, success)
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
class TestPricingPage:
|
|
async def test_accessible_without_auth(self, client, db):
|
|
response = await client.get("/billing/pricing")
|
|
assert response.status_code == 200
|
|
|
|
async def test_accessible_with_auth(self, auth_client, db, test_user):
|
|
response = await auth_client.get("/billing/pricing")
|
|
assert response.status_code == 200
|
|
|
|
async def test_with_subscription(self, auth_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], plan="pro", status="active")
|
|
response = await auth_client.get("/billing/pricing")
|
|
assert response.status_code == 200
|
|
|
|
|
|
class TestSuccessPage:
|
|
async def test_requires_auth(self, client, db):
|
|
response = await client.get("/billing/success", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_accessible_with_auth(self, auth_client, db, test_user):
|
|
response = await auth_client.get("/billing/success")
|
|
assert response.status_code == 200
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# Helper: insert a product into both payment_products and paddle_products
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
async def _insert_test_product(db, key="starter", price_id="pri_starter_123"):
|
|
"""Insert a test product into payment_products (used by get_price_id) and paddle_products (legacy fallback)."""
|
|
await db.execute(
|
|
"INSERT INTO payment_products (provider, key, provider_product_id, provider_price_id, name, price_cents, currency, billing_type) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
("paddle", key, "pro_test", price_id, "Starter", 1900, "EUR", "subscription"),
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO paddle_products (key, paddle_product_id, paddle_price_id, name, price_cents, currency, billing_type) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(key, "pro_test", price_id, "Starter", 1900, "EUR", "subscription"),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# Checkout (returns JSON — Paddle overlay or Stripe redirect)
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
class TestCheckoutRoute:
|
|
async def test_requires_auth(self, client, db):
|
|
response = await client.post(f"/billing/checkout/{CHECKOUT_PLAN}", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_returns_checkout_json(self, auth_client, db, test_user):
|
|
await _insert_test_product(db)
|
|
|
|
response = await auth_client.post(f"/billing/checkout/{CHECKOUT_PLAN}")
|
|
assert response.status_code == 200
|
|
data = await response.get_json()
|
|
assert "items" in data
|
|
assert data["items"][0]["priceId"] == "pri_starter_123"
|
|
|
|
async def test_invalid_plan_rejected(self, auth_client, db, test_user):
|
|
response = await auth_client.post("/billing/checkout/nonexistent_plan")
|
|
assert response.status_code == 400
|
|
data = await response.get_json()
|
|
assert "error" in data
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# Manage subscription / Portal
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
class TestManageRoute:
|
|
async def test_requires_auth(self, client, db):
|
|
response = await client.post("/billing/manage", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_requires_subscription(self, auth_client, db, test_user):
|
|
response = await auth_client.post("/billing/manage", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_redirects_to_portal(self, auth_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], provider_subscription_id="sub_test")
|
|
|
|
mock_sub = MagicMock()
|
|
mock_sub.management_urls.update_payment_method = "https://paddle.com/manage/test_123"
|
|
mock_client = MagicMock()
|
|
mock_client.subscriptions.get.return_value = mock_sub
|
|
|
|
with patch("padelnomics.billing.paddle._paddle_client", return_value=mock_client):
|
|
response = await auth_client.post("/billing/manage", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# Cancel subscription
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
class TestCancelRoute:
|
|
async def test_requires_auth(self, client, db):
|
|
response = await client.post("/billing/cancel", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_no_error_without_subscription(self, auth_client, db, test_user):
|
|
response = await auth_client.post("/billing/cancel", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_cancels_subscription(self, auth_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], provider_subscription_id="sub_test")
|
|
|
|
mock_client = MagicMock()
|
|
with patch("padelnomics.billing.paddle._paddle_client", return_value=mock_client):
|
|
response = await auth_client.post("/billing/cancel", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
mock_client.subscriptions.cancel.assert_called_once()
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# Stripe webhook returns 404 when not configured
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
class TestStripeWebhookEndpoint:
|
|
async def test_returns_404_when_not_configured(self, client, db):
|
|
"""Stripe webhook returns 404 when STRIPE_WEBHOOK_SECRET is empty."""
|
|
response = await client.post(
|
|
"/billing/webhook/stripe",
|
|
data=b'{}',
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# subscription_required decorator
|
|
# ════════════════════════════════════════════════════════════
|
|
|
|
from padelnomics.auth.routes import subscription_required # noqa: E402
|
|
from quart import Blueprint # noqa: E402
|
|
|
|
test_bp = Blueprint("test", __name__)
|
|
|
|
|
|
@test_bp.route("/protected")
|
|
@subscription_required()
|
|
async def protected_route():
|
|
return "success", 200
|
|
|
|
|
|
@test_bp.route("/custom_allowed")
|
|
@subscription_required(allowed=("active", "past_due"))
|
|
async def custom_allowed_route():
|
|
return "success", 200
|
|
|
|
|
|
class TestSubscriptionRequiredDecorator:
|
|
@pytest.fixture
|
|
async def test_app(self, app):
|
|
app.register_blueprint(test_bp)
|
|
return app
|
|
|
|
@pytest.fixture
|
|
async def test_client(self, test_app):
|
|
async with test_app.test_client() as c:
|
|
yield c
|
|
|
|
async def test_redirects_unauthenticated(self, test_client, db):
|
|
response = await test_client.get("/protected", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_redirects_without_subscription(self, test_client, db, test_user):
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/protected", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_allows_active_subscription(self, test_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], plan="pro", status="active")
|
|
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/protected")
|
|
assert response.status_code == 200
|
|
|
|
async def test_allows_on_trial(self, test_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], plan="pro", status="on_trial")
|
|
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/protected")
|
|
assert response.status_code == 200
|
|
|
|
async def test_allows_cancelled(self, test_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], plan="pro", status="cancelled")
|
|
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/protected")
|
|
assert response.status_code == 200
|
|
|
|
async def test_rejects_expired(self, test_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], plan="pro", status="expired")
|
|
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/protected", follow_redirects=False)
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
@pytest.mark.parametrize("status", ["free", "active", "on_trial", "cancelled", "past_due", "paused", "expired"])
|
|
async def test_default_allowed_tuple(self, test_client, db, test_user, create_subscription, status):
|
|
if status != "free":
|
|
await create_subscription(test_user["id"], plan="pro", status=status)
|
|
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/protected", follow_redirects=False)
|
|
|
|
if status in ("active", "on_trial", "cancelled"):
|
|
assert response.status_code == 200
|
|
else:
|
|
assert response.status_code in (302, 303, 307)
|
|
|
|
async def test_custom_allowed_tuple(self, test_client, db, test_user, create_subscription):
|
|
await create_subscription(test_user["id"], plan="pro", status="past_due")
|
|
|
|
async with test_client.session_transaction() as sess:
|
|
sess["user_id"] = test_user["id"]
|
|
|
|
response = await test_client.get("/custom_allowed")
|
|
assert response.status_code == 200
|