Innova-AI/dictia-public
3
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Files
0b91294c4569a260ab48e2516e4aef9149c5c172
dictia-public/tests/test_stripe_webhook.py
Allison 64738bfd1f 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>
2026-04-28 08:41:03 -04:00

785 lines
31 KiB
Python
Raw Blame History

"""Tests for B-2.8 — Stripe Webhook (subscription lifecycle + idempotency).
Covers:
- Signature verification (rejects missing/invalid; honors STRIPE_WEBHOOK_SECRET)
- Idempotency via WebhookEvent table
- Each event handler:
checkout.session.completed
customer.subscription.updated
customer.subscription.deleted
invoice.payment_succeeded
invoice.payment_failed
- User resolution order (stripe_customer_id → metadata → email)
- 500 retry path on handler exceptions
Mocks `_verify_event` (since real signatures need a real secret) and
`stripe.Subscription.retrieve` to avoid real Stripe API calls.
Note: pytest cannot collect this file on Windows native because src/init_db.py
imports `fcntl` (POSIX-only). Use tests/_run_stripe_webhook_windows.py.
"""
import json
import os
import sys
from unittest.mock import patch, MagicMock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
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('RATELIMIT_ENABLED', 'false')
from src.app import app, db, bcrypt # noqa: E402
from src.models.user import User # noqa: E402
from src.models.subscription import Subscription # noqa: E402
from src.models.webhook_event import WebhookEvent # noqa: E402
_WEBHOOK_ENV_VARS = (
'STRIPE_SECRET_KEY',
'STRIPE_WEBHOOK_SECRET',
)
def _clear_webhook_env():
for k in _WEBHOOK_ENV_VARS:
os.environ.pop(k, None)
import stripe
stripe.api_key = None
def _disable_csrf():
app.config['WTF_CSRF_ENABLED'] = False
def _make_user(email='hookuser@example.qc.ca', password='Password!123',
username=None, name='Hook User',
stripe_customer_id=None,
subscription_status=None):
hashed = bcrypt.generate_password_hash(password).decode('utf-8')
u = User(
username=username or email.split('@', 1)[0][:20],
email=email,
password=hashed,
email_verified=True,
name=name,
stripe_customer_id=stripe_customer_id,
subscription_status=subscription_status,
)
db.session.add(u)
db.session.commit()
return u
def _make_event(event_type, obj_data, event_id='evt_test_123'):
"""Build a fake Stripe event for testing webhook handlers."""
return MagicMock(
id=event_id,
type=event_type,
data=MagicMock(object=obj_data),
)
def _make_checkout_session(customer='cus_test', subscription='sub_test',
plan_slug='dictia-cloud', period='monthly',
email=None, user_id='1'):
return {
'id': 'cs_test_abc',
'customer': customer,
'subscription': subscription,
'customer_email': email,
'metadata': {
'dictia_user_id': user_id,
'dictia_plan_slug': plan_slug,
'dictia_period': period,
},
}
def _make_subscription_obj(sub_id='sub_test', customer='cus_test',
status='active', period_end=1730000000,
plan_slug='dictia-cloud', period='monthly'):
return {
'id': sub_id,
'customer': customer,
'status': status,
'current_period_end': period_end,
'metadata': {'dictia_plan_slug': plan_slug, 'dictia_period': period},
}
def _make_invoice(sub_id='sub_test', customer='cus_test'):
return {
'id': 'in_test_123',
'customer': customer,
'subscription': sub_id,
}
def _post_webhook(client, payload=b'{}', signature='t=1,v1=fake'):
return client.post(
'/checkout/webhooks/stripe',
data=payload,
headers={'Stripe-Signature': signature, 'Content-Type': 'application/json'},
)
# ----------------------------------------------------------------------
# 1-3. Signature verification
# ----------------------------------------------------------------------
def test_webhook_rejects_missing_signature():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
with app.test_client() as client:
# No Stripe-Signature header
resp = client.post(
'/checkout/webhooks/stripe',
data=b'{}',
headers={'Content-Type': 'application/json'},
)
assert resp.status_code == 400
body = resp.get_json()
assert body.get('error') == 'invalid_signature'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
def test_webhook_rejects_invalid_signature():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
with app.test_client() as client:
# Real construct_event will fail on bogus signature
resp = _post_webhook(client, payload=b'{"id":"evt_x"}',
signature='t=1,v1=bogus')
assert resp.status_code == 400
assert resp.get_json().get('error') == 'invalid_signature'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
def test_webhook_rejects_when_secret_not_configured():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
# NO STRIPE_WEBHOOK_SECRET
db.create_all()
try:
with app.test_client() as client:
resp = _post_webhook(client, payload=b'{"id":"evt_x"}')
assert resp.status_code == 400
assert resp.get_json().get('error') == 'invalid_signature'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 4. Idempotency
# ----------------------------------------------------------------------
def test_webhook_idempotent_on_duplicate_event_id():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='dup@example.qc.ca',
stripe_customer_id='cus_dup')
event = _make_event(
'checkout.session.completed',
_make_checkout_session(customer='cus_dup',
subscription='sub_dup',
user_id=str(user.id)),
event_id='evt_dup_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_verify.return_value = event
mock_retr.return_value = {
'current_period_end': 1730000000,
}
with app.test_client() as client:
# First call processes
resp1 = _post_webhook(client)
assert resp1.status_code == 200
assert resp1.get_json().get('received') is True
# Second call dedup'd
resp2 = _post_webhook(client)
assert resp2.status_code == 200
body2 = resp2.get_json()
assert body2.get('duplicate') is True
# Only one Subscription row + one WebhookEvent row
assert Subscription.query.filter_by(stripe_subscription_id='sub_dup').count() == 1
assert WebhookEvent.query.filter_by(stripe_event_id='evt_dup_1').count() == 1
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 5. Unhandled event types
# ----------------------------------------------------------------------
def test_webhook_acks_unhandled_event_type():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
event = _make_event('customer.created', {'id': 'cus_new'},
event_id='evt_unhandled_1')
with patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
body = resp.get_json()
assert body.get('received') is True
assert body.get('handled') is False
# Recorded so Stripe stops retrying
assert WebhookEvent.query.filter_by(stripe_event_id='evt_unhandled_1').count() == 1
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 6. checkout.session.completed — happy path
# ----------------------------------------------------------------------
def test_checkout_session_completed_creates_subscription_and_sets_user_status():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='create@example.qc.ca',
stripe_customer_id='cus_test')
event = _make_event(
'checkout.session.completed',
_make_checkout_session(customer='cus_test',
subscription='sub_test',
user_id=str(user.id)),
event_id='evt_create_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_verify.return_value = event
mock_retr.return_value = {'current_period_end': 1730000000}
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
sub = Subscription.query.filter_by(stripe_subscription_id='sub_test').first()
assert sub is not None
assert sub.user_id == user.id
assert sub.plan_slug == 'dictia-cloud'
assert sub.period == 'monthly'
assert sub.status == 'active'
assert sub.current_period_end is not None
db.session.refresh(user)
assert user.subscription_status == 'active'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 7. User resolution by stripe_customer_id
# ----------------------------------------------------------------------
def test_checkout_session_completed_resolves_user_by_stripe_customer_id():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='resolve@example.qc.ca',
stripe_customer_id='cus_resolve')
session_obj = _make_checkout_session(
customer='cus_resolve',
subscription='sub_resolve',
user_id='',
)
# No metadata.dictia_user_id
session_obj['metadata'].pop('dictia_user_id', None)
event = _make_event('checkout.session.completed', session_obj,
event_id='evt_resolve_1')
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_verify.return_value = event
mock_retr.return_value = {'current_period_end': 1730000000}
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
sub = Subscription.query.filter_by(stripe_subscription_id='sub_resolve').first()
assert sub is not None
assert sub.user_id == user.id
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 8. User resolution by metadata when customer unknown
# ----------------------------------------------------------------------
def test_checkout_session_completed_resolves_user_by_metadata_when_customer_unknown():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='meta@example.qc.ca',
stripe_customer_id=None)
event = _make_event(
'checkout.session.completed',
_make_checkout_session(customer='cus_new',
subscription='sub_meta',
user_id=str(user.id)),
event_id='evt_meta_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_verify.return_value = event
mock_retr.return_value = {'current_period_end': 1730000000}
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
sub = Subscription.query.filter_by(stripe_subscription_id='sub_meta').first()
assert sub is not None
assert sub.user_id == user.id
db.session.refresh(user)
# stripe_customer_id is bound from the event
assert user.stripe_customer_id == 'cus_new'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 9. Idempotency via subscription_id uniqueness (different event IDs, same sub)
# ----------------------------------------------------------------------
def test_checkout_session_completed_idempotent_via_subscription_id_uniqueness():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='subdup@example.qc.ca',
stripe_customer_id='cus_subdup')
session_obj = _make_checkout_session(
customer='cus_subdup',
subscription='sub_subdup',
user_id=str(user.id),
)
event_a = _make_event('checkout.session.completed', session_obj,
event_id='evt_subdup_a')
event_b = _make_event('checkout.session.completed', session_obj,
event_id='evt_subdup_b')
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_retr.return_value = {'current_period_end': 1730000000}
with app.test_client() as client:
mock_verify.return_value = event_a
resp1 = _post_webhook(client)
assert resp1.status_code == 200
mock_verify.return_value = event_b
resp2 = _post_webhook(client)
assert resp2.status_code == 200
# Only one Subscription row despite two distinct events
assert Subscription.query.filter_by(stripe_subscription_id='sub_subdup').count() == 1
# But both events recorded
assert WebhookEvent.query.filter_by(stripe_event_id='evt_subdup_a').count() == 1
assert WebhookEvent.query.filter_by(stripe_event_id='evt_subdup_b').count() == 1
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 10. customer.subscription.updated
# ----------------------------------------------------------------------
def test_subscription_updated_updates_status_and_period_end():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
from datetime import datetime
user = _make_user(email='upd@example.qc.ca',
stripe_customer_id='cus_upd',
subscription_status='active')
existing = Subscription(
user_id=user.id,
stripe_customer_id='cus_upd',
stripe_subscription_id='sub_upd',
plan_slug='dictia-cloud',
period='monthly',
status='active',
current_period_end=datetime(2025, 1, 1),
)
db.session.add(existing)
db.session.commit()
event = _make_event(
'customer.subscription.updated',
_make_subscription_obj(sub_id='sub_upd', customer='cus_upd',
status='past_due',
period_end=1735689600), # 2025-01-01
event_id='evt_upd_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
db.session.refresh(existing)
assert existing.status == 'past_due'
assert existing.current_period_end is not None
db.session.refresh(user)
assert user.subscription_status == 'past_due'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 11. customer.subscription.updated when row missing
# ----------------------------------------------------------------------
def test_subscription_updated_creates_row_if_missing():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='race@example.qc.ca',
stripe_customer_id='cus_race')
event = _make_event(
'customer.subscription.updated',
_make_subscription_obj(sub_id='sub_race', customer='cus_race',
status='active'),
event_id='evt_race_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
sub = Subscription.query.filter_by(stripe_subscription_id='sub_race').first()
assert sub is not None
assert sub.user_id == user.id
assert sub.status == 'active'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 12. customer.subscription.deleted
# ----------------------------------------------------------------------
def test_subscription_deleted_marks_canceled():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='del@example.qc.ca',
stripe_customer_id='cus_del',
subscription_status='active')
existing = Subscription(
user_id=user.id,
stripe_customer_id='cus_del',
stripe_subscription_id='sub_del',
plan_slug='dictia-cloud',
period='monthly',
status='active',
)
db.session.add(existing)
db.session.commit()
event = _make_event(
'customer.subscription.deleted',
_make_subscription_obj(sub_id='sub_del', customer='cus_del',
status='canceled'),
event_id='evt_del_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
db.session.refresh(existing)
assert existing.status == 'canceled'
db.session.refresh(user)
assert user.subscription_status == 'canceled'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 13. invoice.payment_succeeded — recovers past_due
# ----------------------------------------------------------------------
def test_invoice_payment_succeeded_recovers_past_due():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='paysucc@example.qc.ca',
stripe_customer_id='cus_paysucc',
subscription_status='past_due')
existing = Subscription(
user_id=user.id,
stripe_customer_id='cus_paysucc',
stripe_subscription_id='sub_paysucc',
plan_slug='dictia-cloud',
period='monthly',
status='past_due',
)
db.session.add(existing)
db.session.commit()
event = _make_event(
'invoice.payment_succeeded',
_make_invoice(sub_id='sub_paysucc', customer='cus_paysucc'),
event_id='evt_paysucc_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
db.session.refresh(existing)
assert existing.status == 'active'
db.session.refresh(user)
assert user.subscription_status == 'active'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 14. invoice.payment_failed
# ----------------------------------------------------------------------
def test_invoice_payment_failed_marks_past_due():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='payfail@example.qc.ca',
stripe_customer_id='cus_payfail',
subscription_status='active')
existing = Subscription(
user_id=user.id,
stripe_customer_id='cus_payfail',
stripe_subscription_id='sub_payfail',
plan_slug='dictia-cloud',
period='monthly',
status='active',
)
db.session.add(existing)
db.session.commit()
event = _make_event(
'invoice.payment_failed',
_make_invoice(sub_id='sub_payfail', customer='cus_payfail'),
event_id='evt_payfail_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
db.session.refresh(existing)
assert existing.status == 'past_due'
db.session.refresh(user)
assert user.subscription_status == 'past_due'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 15. Handler exception → 500 (Stripe retry)
# ----------------------------------------------------------------------
def test_handler_exception_returns_500_for_stripe_retry():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='boom@example.qc.ca',
stripe_customer_id='cus_boom')
event = _make_event(
'checkout.session.completed',
_make_checkout_session(customer='cus_boom',
subscription='sub_boom',
user_id=str(user.id)),
event_id='evt_boom_1',
)
# Patch the dispatch table entry so our raising stub fires inside the
# try/except handler block in stripe_webhook(). Patching the module
# attribute alone wouldn't help because _HANDLERS captures the original
# reference at module load.
from src.billing import webhooks as wh_mod
def _boom(_event):
raise RuntimeError('boom')
with patch.dict(wh_mod._HANDLERS,
{'checkout.session.completed': _boom}), \
patch('src.billing.webhooks._verify_event') as mock_verify:
mock_verify.return_value = event
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 500
assert resp.get_json().get('error') == 'handler_failed'
# No WebhookEvent recorded (rolled back)
assert WebhookEvent.query.filter_by(stripe_event_id='evt_boom_1').count() == 0
assert Subscription.query.filter_by(stripe_subscription_id='sub_boom').count() == 0
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 16. Metadata 'dictia_user_id' invalid → falls through
# ----------------------------------------------------------------------
def test_resolve_user_metadata_dictia_user_id_invalid_falls_through():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='emailfb@example.qc.ca',
stripe_customer_id=None)
session_obj = _make_checkout_session(
customer='cus_unknown',
subscription='sub_emailfb',
user_id='not-a-number',
email='emailfb@example.qc.ca',
)
event = _make_event('checkout.session.completed', session_obj,
event_id='evt_emailfb_1')
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_verify.return_value = event
mock_retr.return_value = {'current_period_end': 1730000000}
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
sub = Subscription.query.filter_by(stripe_subscription_id='sub_emailfb').first()
assert sub is not None
# Resolved by email
assert sub.user_id == user.id
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
# ----------------------------------------------------------------------
# 17. WebhookEvent records subscription_id for audit
# ----------------------------------------------------------------------
def test_webhook_event_records_subscription_id_for_audit():
with app.app_context():
_disable_csrf()
_clear_webhook_env()
os.environ['STRIPE_SECRET_KEY'] = 'sk_test_fake'
os.environ['STRIPE_WEBHOOK_SECRET'] = 'whsec_fake'
db.create_all()
try:
user = _make_user(email='audit@example.qc.ca',
stripe_customer_id='cus_audit')
event = _make_event(
'checkout.session.completed',
_make_checkout_session(customer='cus_audit',
subscription='sub_audit',
user_id=str(user.id)),
event_id='evt_audit_1',
)
with patch('src.billing.webhooks._verify_event') as mock_verify, \
patch('src.billing.webhooks.stripe.Subscription.retrieve') as mock_retr:
mock_verify.return_value = event
mock_retr.return_value = {'current_period_end': 1730000000}
with app.test_client() as client:
resp = _post_webhook(client)
assert resp.status_code == 200
wh = WebhookEvent.query.filter_by(stripe_event_id='evt_audit_1').first()
assert wh is not None
assert wh.event_type == 'checkout.session.completed'
assert wh.stripe_subscription_id == 'sub_audit'
assert wh.stripe_customer_id == 'cus_audit'
finally:
db.session.rollback()
db.drop_all()
_clear_webhook_env()
Reference in New Issue View Git Blame Copy Permalink