mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-12 23:23:21 +00:00
feat: add appointments, campaigns, admin, storage, tests and various updates
- 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>
This commit is contained in:
81
tests/test_billing.py
Normal file
81
tests/test_billing.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user