mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-13 08:03:54 +00:00
92d4c2fc5e
- Add new routers: admin, appointments, campaigns - Add storage service and logging config - Add migrations directory and test suite with pytest config - Add supabase_migration_features.sql - Update models, dependencies, config, and existing routers - Remove whatsapp_service (deleted) - Update pyproject.toml and uv.lock dependencies Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
82 lines
3.3 KiB
Python
82 lines
3.3 KiB
Python
"""Tests for billing webhook idempotency and Stripe integration."""
|
|
import pytest
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestStripeWebhookIdempotency:
|
|
def test_duplicate_event_returns_200_without_processing(self, client):
|
|
"""Same Stripe event ID sent twice should only process once."""
|
|
event_id = "evt_test_123"
|
|
payload = json.dumps({
|
|
"id": event_id,
|
|
"type": "checkout.session.completed",
|
|
"data": {"object": {"metadata": {"user_id": "user-123", "plan": "starter"}, "customer": "cus_123", "subscription": "sub_123"}},
|
|
}).encode()
|
|
|
|
with patch("app.routers.billing.get_supabase") as mock_sb, \
|
|
patch("app.routers.billing.settings") as mock_settings:
|
|
|
|
mock_settings.stripe_webhook_secret = ""
|
|
mock_settings.stripe_secret_key = "sk_test_123"
|
|
mock_settings.app_env = "development"
|
|
mock_settings.n8n_handoff_webhook_url = None
|
|
|
|
sb = MagicMock()
|
|
# First call: event not found
|
|
first_check = MagicMock(data=[])
|
|
# Second call: event found (already processed)
|
|
second_check = MagicMock(data=[{"stripe_event_id": event_id}])
|
|
|
|
call_count = 0
|
|
def table_exec(*args, **kwargs):
|
|
return MagicMock(data=[])
|
|
|
|
sb.table.return_value.select.return_value.eq.return_value.execute.side_effect = [
|
|
first_check, # first idempotency check → not found
|
|
MagicMock(data=[]), # subscription upsert check
|
|
MagicMock(data=[]), # insert event record
|
|
]
|
|
sb.table.return_value.upsert.return_value.execute.return_value = MagicMock(data=[{}])
|
|
sb.table.return_value.insert.return_value.execute.return_value = MagicMock(data=[{}])
|
|
mock_sb.return_value = sb
|
|
|
|
# First request
|
|
resp1 = client.post(
|
|
"/api/v1/billing/webhook",
|
|
content=payload,
|
|
headers={"content-type": "application/json"},
|
|
)
|
|
assert resp1.status_code == 200
|
|
|
|
# Reset mock: now event IS found
|
|
sb.table.return_value.select.return_value.eq.return_value.execute.return_value = second_check
|
|
|
|
# Second request with same event ID
|
|
resp2 = client.post(
|
|
"/api/v1/billing/webhook",
|
|
content=payload,
|
|
headers={"content-type": "application/json"},
|
|
)
|
|
assert resp2.status_code == 200
|
|
assert resp2.json() == {"received": True}
|
|
|
|
def test_webhook_requires_stripe_signature_in_production(self, client):
|
|
"""In production, webhook without signature should fail."""
|
|
payload = json.dumps({"id": "evt_123", "type": "test"}).encode()
|
|
|
|
with patch("app.routers.billing.settings") as mock_settings:
|
|
mock_settings.stripe_webhook_secret = "whsec_real_secret"
|
|
mock_settings.stripe_secret_key = "sk_live_123"
|
|
mock_settings.app_env = "production"
|
|
|
|
resp = client.post(
|
|
"/api/v1/billing/webhook",
|
|
content=payload,
|
|
headers={"content-type": "application/json"},
|
|
# No stripe-signature header
|
|
)
|
|
|
|
# Should fail due to missing signature
|
|
assert resp.status_code in (400, 500)
|