#!/usr/bin/env python3
"""Billing route tests (Phase 2 — 2b)
Tests for billing routes:
- GET /billing/upgrade — authenticated, unauthenticated
- POST /billing/create-checkout — tier validation
- POST /billing/stripe/webhook — Stripe event handling:
- checkout.session.completed — activates subscription
- customer.subscription.deleted — downgrades to free
- invoice.payment_succeeded — clears overdue
- Signature validation (Stripe-Signature header)
- Rejected invalid signatures (400)
- Unconfigured webhook secret
Uses shared flask_app from app.app (same pattern as test_auth/test_hardening).
"""
import json
import os
import random
import sqlite3
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Use shared test DB (singleton pattern — only initializes once)
import tests.test_shared_db # noqa: F401
# Set required env vars BEFORE importing app modules
os.environ.setdefault("AGENTFORMS_SECRET_KEY", "billing-test-secret")
os.environ.setdefault("STRIPE_WEBHOOK_SECRET", "whsec_test_secret")
os.environ.setdefault("ENCRYPTION_KEY", "5c321ae59453d7bae2e05a9e19e97347411a033d9858a0456d5cc63acfcb0369")
# Import shared Flask app — same instance used by all test files
from app.app import app as flask_app
# Ensure test mode
flask_app.config["TESTING"] = True
flask_app.config["SESSION_COOKIE_SECURE"] = False
class TestUpgradePage:
"""Test GET /billing/upgrade."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_upgrade_requires_login(self, client):
resp = client.get("/billing/upgrade", follow_redirects=False)
assert resp.status_code in (302, 403)
def test_upgrade_accessible_when_logged_in(self, client):
email = f"upgrade-billing{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Upgrade User",
})
resp = client.get("/billing/upgrade")
assert resp.status_code == 200
class TestCreateCheckout:
"""Test POST /billing/create-checkout."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_checkout_requires_login(self, client):
resp = client.post("/billing/create-checkout", data={
"tier": "starter",
}, follow_redirects=False)
assert resp.status_code in (302, 403)
def test_checkout_invalid_tier(self, client):
email = f"checkout-billing{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Checkout User",
})
resp = client.post("/billing/create-checkout", data={
"tier": "nonexistent_tier",
}, follow_redirects=True)
# Should redirect back to upgrade page with error
assert resp.status_code == 200
class TestStripeWebhook:
"""Test POST /billing/stripe/webhook."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def _random_email(self):
return f"webhook-billing{random.randint(10000, 99999)}@example.com"
def test_webhook_no_signature(self, client):
"""Test webhook without Stripe-Signature header."""
resp = client.post("/billing/stripe/webhook", json={})
assert resp.status_code == 400
def test_webhook_invalid_signature(self, client):
"""Test webhook with invalid signature returns 400."""
resp = client.post(
"/billing/stripe/webhook",
data=json.dumps({"type": "test"}),
content_type="application/json",
headers={"Stripe-Signature": "invalid_signature_here"},
)
assert resp.status_code == 400
def test_webhook_no_body(self, client):
"""Test webhook with empty body."""
resp = client.post(
"/billing/stripe/webhook",
data=b"",
headers={"Stripe-Signature": "no_body_sig"},
)
assert resp.status_code == 400
def test_webhook_csrf_exempt(self, client):
"""Test that webhook is CSRF exempt (no session required)."""
# Should not return 403 CSRF error
resp = client.post(
"/billing/stripe/webhook",
data=b"{}",
content_type="application/json",
)
assert resp.status_code != 403
class TestWebhookIdempotency:
"""Test that duplicate Stripe events are handled idempotently."""
def test_dedup_table_created(self):
"""Test that stripe_webhook_events table can be created."""
import app.models
# The table is created lazily by the webhook handler. Create it directly.
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS stripe_webhook_events (
event_id TEXT PRIMARY KEY,
processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
tables = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='stripe_webhook_events'"
).fetchall()
conn.close()
assert len(tables) == 1
def test_record_webhook_event_idempotent(self):
"""Test _record_webhook_event returns False for duplicates."""
from app.routes.billing import _record_webhook_event
event_id = "evt_test_idempotent_" + str(random.randint(1000, 9999))
with flask_app.app_context():
first = _record_webhook_event(event_id)
assert first is True
second = _record_webhook_event(event_id)
assert second is False
class TestTierHandling:
"""Test tier update helpers."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_user_starts_as_free(self, client):
email = f"tier-billing{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Free User",
})
import app.models
from app.models import hash_value
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
row = conn.execute(
"SELECT tier FROM users WHERE email_hash = ?",
(hash_value(email.lower()),)
).fetchone()
conn.close()
assert row[0] == "free" # tier column
def test_subscription_deleted_downgrades_to_free(self):
"""Test _handle_subscription_deleted downgrades user."""
email = f"subdel-billing{random.randint(10000, 99999)}@example.com"
from app.models import register_user, get_db, hash_value, update_user_tier
with flask_app.app_context():
user = register_user(email, "SecurePass123", "Sub User")
update_user_tier(user["id"], "starter", "sub_test_123")
db = get_db()
row = db.execute(
"SELECT tier FROM users WHERE email_hash = ?",
(hash_value(email.lower()),)
).fetchone()
db.close()
assert row[0] == "starter" # tier column
# Simulate subscription deleted handler
update_user_tier(user["id"], "free")
db = get_db()
row = db.execute(
"SELECT tier FROM users WHERE email_hash = ?",
(hash_value(email.lower()),)
).fetchone()
db.close()
assert row[0] == "free" # tier column
def _random_email(self):
return f"tier-billing{random.randint(10000, 99999)}@example.com"
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])