feat(billing): B-2.8 Stripe webhook handler (subscription lifecycle + idempotency)

Endpoint: POST /checkout/webhooks/stripe (CSRF-exempt; signature-verified)

Handles 5 Stripe events:
  - checkout.session.completed     -> create Subscription, activate user
  - customer.subscription.updated  -> sync status + current_period_end
  - customer.subscription.deleted  -> mark canceled
  - invoice.payment_succeeded      -> recover from past_due if applicable
  - invoice.payment_failed         -> mark past_due

Idempotency via WebhookEvent table (Stripe ID dedup) and Subscription
unique constraint on stripe_subscription_id (defends against duplicate
deliveries with distinct event IDs).

User resolution prefers stripe_customer_id (server-set, anti-tamper)
over event metadata.dictia_user_id over customer_email (per B-2.7
review note).

New tables created via db.create_all():
  - subscription (FK user.id ondelete=SET NULL for Loi 25 art. 28.1)
  - webhook_event (idempotency ledger)

CSRF exemption wired via src/billing/exempt_webhook_csrf(csrf) called
from src/app.py after billing_bp registration.

Tests: 17/17 pass via tests/_run_stripe_webhook_windows.py.
Existing 25 B-2.7 + 21 TOTP + 22 WebAuthn + 21 OAuth + 16 email tests
unaffected.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Allison
2026-04-28 08:41:03 -04:00
parent f1a5ad565f
commit 64738bfd1f
10 changed files with 1306 additions and 10 deletions

View File

@@ -0,0 +1,74 @@
"""Windows manual driver for tests/test_stripe_webhook.py.
src/init_db.py imports `fcntl`, which is POSIX-only. On Windows we stub it
before src.app gets imported, then run each test_* function and report.
Run from the repo root:
py -3 tests/_run_stripe_webhook_windows.py
"""
import os
import sys
import types
import traceback
# 1) Stub fcntl BEFORE any import of src.* happens.
if 'fcntl' not in sys.modules:
fcntl_stub = types.ModuleType('fcntl')
fcntl_stub.LOCK_EX = 2
fcntl_stub.LOCK_NB = 4
fcntl_stub.LOCK_UN = 8
fcntl_stub.LOCK_SH = 1
fcntl_stub.flock = lambda *_args, **_kw: None
fcntl_stub.fcntl = lambda *_args, **_kw: 0
sys.modules['fcntl'] = fcntl_stub
# 2) Make repo root importable
HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(HERE)
sys.path.insert(0, REPO)
# 3) Test-friendly env defaults
os.environ.setdefault('SQLALCHEMY_DATABASE_URI', 'sqlite:///:memory:')
os.environ.setdefault('SECRET_KEY', 'test-secret-key-webhook')
os.environ.setdefault('ENABLE_EMAIL_VERIFICATION', 'false')
os.environ.setdefault('REQUIRE_EMAIL_VERIFICATION', 'false')
os.environ.setdefault('TRANSCRIPTION_BASE_URL', 'http://test-stub')
os.environ.setdefault('TRANSCRIPTION_API_KEY', 'test-stub')
os.environ.setdefault('RATELIMIT_ENABLED', 'false')
try:
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
sys.stderr.reconfigure(encoding='utf-8', errors='replace')
except Exception:
pass
# 4) Import the test module and run every test_* function
import importlib.util # noqa: E402
spec = importlib.util.spec_from_file_location(
'test_stripe_webhook',
os.path.join(HERE, 'test_stripe_webhook.py'),
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
tests = [(name, fn) for name, fn in vars(mod).items()
if name.startswith('test_') and callable(fn)]
passed = 0
failed = []
for name, fn in tests:
try:
fn()
print(f' PASS {name}')
passed += 1
except Exception as e: # noqa: BLE001
print(f' FAIL {name}: {type(e).__name__}: {e}')
failed.append((name, traceback.format_exc()))
total = len(tests)
print()
print(f'Result: {passed}/{total} passed, {len(failed)} failed')
if failed:
print('\n--- Failures ---\n')
for name, tb in failed:
print(f'### {name}\n{tb}\n')
sys.exit(0 if not failed else 1)