Samir Boulahtit
b9f08b853f
Delete empty stub files from models/database/: - audit.py, backup.py, configuration.py, monitoring.py - notification.py, payment.py, search.py, task.py Delete re-export files: - models/database/subscription.py → app.modules.billing.models - models/database/architecture_scan.py → app.modules.dev_tools.models - models/database/test_run.py → app.modules.dev_tools.models - models/schema/subscription.py → app.modules.billing.schemas - models/schema/marketplace.py (empty) - models/schema/monitoring.py (empty) Migrate schemas to canonical module locations: - billing.py → app/modules/billing/schemas/ - vendor_product.py → app/modules/catalog/schemas/ - homepage_sections.py → app/modules/cms/schemas/ Keep as CORE (framework-level, used everywhere): - models/schema/: admin, auth, base, company, email, image, media, team, vendor* - models/database/: admin*, base, company, email, feature, media, platform*, user, vendor* Update 30+ files to use canonical import locations. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
394 lines
13 KiB
Python
394 lines
13 KiB
Python
# tests/unit/services/test_stripe_webhook_handler.py
|
|
"""Unit tests for StripeWebhookHandler."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.handlers.stripe_webhook import StripeWebhookHandler
|
|
from app.modules.billing.models import (
|
|
BillingHistory,
|
|
StripeWebhookEvent,
|
|
SubscriptionStatus,
|
|
SubscriptionTier,
|
|
VendorSubscription,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.billing
|
|
class TestStripeWebhookHandlerIdempotency:
|
|
"""Test suite for webhook handler idempotency."""
|
|
|
|
def setup_method(self):
|
|
"""Initialize handler instance before each test."""
|
|
self.handler = StripeWebhookHandler()
|
|
|
|
def test_handle_event_creates_webhook_event_record(self, db, mock_stripe_event):
|
|
"""Test that handling an event creates a webhook event record."""
|
|
self.handler.handle_event(db, mock_stripe_event)
|
|
|
|
record = (
|
|
db.query(StripeWebhookEvent)
|
|
.filter(StripeWebhookEvent.event_id == mock_stripe_event.id)
|
|
.first()
|
|
)
|
|
assert record is not None
|
|
assert record.event_type == mock_stripe_event.type
|
|
assert record.status == "processed"
|
|
|
|
def test_handle_event_skips_duplicate(self, db, mock_stripe_event):
|
|
"""Test that duplicate events are skipped."""
|
|
# Process first time
|
|
result1 = self.handler.handle_event(db, mock_stripe_event)
|
|
assert result1["status"] != "skipped"
|
|
|
|
# Process second time
|
|
result2 = self.handler.handle_event(db, mock_stripe_event)
|
|
assert result2["status"] == "skipped"
|
|
assert result2["reason"] == "duplicate"
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.billing
|
|
class TestStripeWebhookHandlerCheckout:
|
|
"""Test suite for checkout.session.completed event handling."""
|
|
|
|
def setup_method(self):
|
|
"""Initialize handler instance before each test."""
|
|
self.handler = StripeWebhookHandler()
|
|
|
|
@patch("app.handlers.stripe_webhook.stripe.Subscription.retrieve")
|
|
def test_handle_checkout_completed_success(
|
|
self, mock_stripe_retrieve, db, test_vendor, test_subscription, mock_checkout_event
|
|
):
|
|
"""Test successful checkout completion."""
|
|
# Mock Stripe subscription retrieve
|
|
mock_stripe_sub = MagicMock()
|
|
mock_stripe_sub.current_period_start = int(datetime.now(timezone.utc).timestamp())
|
|
mock_stripe_sub.current_period_end = int(datetime.now(timezone.utc).timestamp())
|
|
mock_stripe_sub.trial_end = None
|
|
mock_stripe_retrieve.return_value = mock_stripe_sub
|
|
|
|
mock_checkout_event.data.object.metadata = {"vendor_id": str(test_vendor.id)}
|
|
|
|
result = self.handler.handle_event(db, mock_checkout_event)
|
|
|
|
assert result["status"] == "processed"
|
|
db.refresh(test_subscription)
|
|
assert test_subscription.stripe_customer_id == "cus_test123"
|
|
assert test_subscription.status == SubscriptionStatus.ACTIVE
|
|
|
|
def test_handle_checkout_completed_no_vendor_id(self, db, mock_checkout_event):
|
|
"""Test checkout with missing vendor_id is skipped."""
|
|
mock_checkout_event.data.object.metadata = {}
|
|
|
|
result = self.handler.handle_event(db, mock_checkout_event)
|
|
|
|
assert result["status"] == "processed"
|
|
assert result["result"]["action"] == "skipped"
|
|
assert result["result"]["reason"] == "no vendor_id"
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.billing
|
|
class TestStripeWebhookHandlerSubscription:
|
|
"""Test suite for subscription event handling."""
|
|
|
|
def setup_method(self):
|
|
"""Initialize handler instance before each test."""
|
|
self.handler = StripeWebhookHandler()
|
|
|
|
def test_handle_subscription_updated_status_change(
|
|
self, db, test_vendor, test_active_subscription, mock_subscription_updated_event
|
|
):
|
|
"""Test subscription update changes status."""
|
|
result = self.handler.handle_event(db, mock_subscription_updated_event)
|
|
|
|
assert result["status"] == "processed"
|
|
|
|
def test_handle_subscription_deleted(
|
|
self, db, test_vendor, test_active_subscription, mock_subscription_deleted_event
|
|
):
|
|
"""Test subscription deletion."""
|
|
result = self.handler.handle_event(db, mock_subscription_deleted_event)
|
|
|
|
assert result["status"] == "processed"
|
|
db.refresh(test_active_subscription)
|
|
assert test_active_subscription.status == SubscriptionStatus.CANCELLED
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.billing
|
|
class TestStripeWebhookHandlerInvoice:
|
|
"""Test suite for invoice event handling."""
|
|
|
|
def setup_method(self):
|
|
"""Initialize handler instance before each test."""
|
|
self.handler = StripeWebhookHandler()
|
|
|
|
def test_handle_invoice_paid_creates_billing_record(
|
|
self, db, test_vendor, test_active_subscription, mock_invoice_paid_event
|
|
):
|
|
"""Test invoice.paid creates billing history record."""
|
|
result = self.handler.handle_event(db, mock_invoice_paid_event)
|
|
|
|
assert result["status"] == "processed"
|
|
|
|
# Check billing record created
|
|
record = (
|
|
db.query(BillingHistory)
|
|
.filter(BillingHistory.vendor_id == test_vendor.id)
|
|
.first()
|
|
)
|
|
assert record is not None
|
|
assert record.status == "paid"
|
|
assert record.total_cents == 4900
|
|
|
|
def test_handle_invoice_paid_resets_counters(
|
|
self, db, test_vendor, test_active_subscription, mock_invoice_paid_event
|
|
):
|
|
"""Test invoice.paid resets order counters."""
|
|
test_active_subscription.orders_this_period = 50
|
|
db.commit()
|
|
|
|
self.handler.handle_event(db, mock_invoice_paid_event)
|
|
|
|
db.refresh(test_active_subscription)
|
|
assert test_active_subscription.orders_this_period == 0
|
|
|
|
def test_handle_payment_failed_marks_past_due(
|
|
self, db, test_vendor, test_active_subscription, mock_payment_failed_event
|
|
):
|
|
"""Test payment failure marks subscription as past due."""
|
|
result = self.handler.handle_event(db, mock_payment_failed_event)
|
|
|
|
assert result["status"] == "processed"
|
|
db.refresh(test_active_subscription)
|
|
assert test_active_subscription.status == SubscriptionStatus.PAST_DUE
|
|
assert test_active_subscription.payment_retry_count == 1
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.billing
|
|
class TestStripeWebhookHandlerUnknownEvents:
|
|
"""Test suite for unknown event handling."""
|
|
|
|
def setup_method(self):
|
|
"""Initialize handler instance before each test."""
|
|
self.handler = StripeWebhookHandler()
|
|
|
|
def test_handle_unknown_event_type(self, db):
|
|
"""Test unknown event types are ignored."""
|
|
mock_event = MagicMock()
|
|
mock_event.id = "evt_unknown123"
|
|
mock_event.type = "customer.unknown_event"
|
|
mock_event.data.object = {}
|
|
|
|
result = self.handler.handle_event(db, mock_event)
|
|
|
|
assert result["status"] == "ignored"
|
|
assert "no handler" in result["reason"]
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.billing
|
|
class TestStripeWebhookHandlerStatusMapping:
|
|
"""Test suite for status mapping helper."""
|
|
|
|
def setup_method(self):
|
|
"""Initialize handler instance before each test."""
|
|
self.handler = StripeWebhookHandler()
|
|
|
|
def test_map_active_status(self):
|
|
"""Test mapping active status."""
|
|
result = self.handler._map_stripe_status("active")
|
|
assert result == SubscriptionStatus.ACTIVE
|
|
|
|
def test_map_trialing_status(self):
|
|
"""Test mapping trialing status."""
|
|
result = self.handler._map_stripe_status("trialing")
|
|
assert result == SubscriptionStatus.TRIAL
|
|
|
|
def test_map_past_due_status(self):
|
|
"""Test mapping past_due status."""
|
|
result = self.handler._map_stripe_status("past_due")
|
|
assert result == SubscriptionStatus.PAST_DUE
|
|
|
|
def test_map_canceled_status(self):
|
|
"""Test mapping canceled status."""
|
|
result = self.handler._map_stripe_status("canceled")
|
|
assert result == SubscriptionStatus.CANCELLED
|
|
|
|
def test_map_unknown_status(self):
|
|
"""Test mapping unknown status defaults to expired."""
|
|
result = self.handler._map_stripe_status("unknown_status")
|
|
assert result == SubscriptionStatus.EXPIRED
|
|
|
|
|
|
# ==================== Fixtures ====================
|
|
|
|
|
|
@pytest.fixture
|
|
def test_subscription_tier(db):
|
|
"""Create a basic subscription tier."""
|
|
tier = SubscriptionTier(
|
|
code="essential",
|
|
name="Essential",
|
|
price_monthly_cents=4900,
|
|
display_order=1,
|
|
is_active=True,
|
|
is_public=True,
|
|
)
|
|
db.add(tier)
|
|
db.commit()
|
|
db.refresh(tier)
|
|
return tier
|
|
|
|
|
|
@pytest.fixture
|
|
def test_subscription(db, test_vendor):
|
|
"""Create a basic subscription for testing."""
|
|
# Create tier first if not exists
|
|
tier = db.query(SubscriptionTier).filter(SubscriptionTier.code == "essential").first()
|
|
if not tier:
|
|
tier = SubscriptionTier(
|
|
code="essential",
|
|
name="Essential",
|
|
price_monthly_cents=4900,
|
|
display_order=1,
|
|
is_active=True,
|
|
is_public=True,
|
|
)
|
|
db.add(tier)
|
|
db.commit()
|
|
|
|
subscription = VendorSubscription(
|
|
vendor_id=test_vendor.id,
|
|
tier="essential",
|
|
status=SubscriptionStatus.TRIAL,
|
|
period_start=datetime.now(timezone.utc),
|
|
period_end=datetime.now(timezone.utc),
|
|
)
|
|
db.add(subscription)
|
|
db.commit()
|
|
db.refresh(subscription)
|
|
return subscription
|
|
|
|
|
|
@pytest.fixture
|
|
def test_active_subscription(db, test_vendor):
|
|
"""Create an active subscription with Stripe IDs."""
|
|
# Create tier first if not exists
|
|
tier = db.query(SubscriptionTier).filter(SubscriptionTier.code == "essential").first()
|
|
if not tier:
|
|
tier = SubscriptionTier(
|
|
code="essential",
|
|
name="Essential",
|
|
price_monthly_cents=4900,
|
|
display_order=1,
|
|
is_active=True,
|
|
is_public=True,
|
|
)
|
|
db.add(tier)
|
|
db.commit()
|
|
|
|
subscription = VendorSubscription(
|
|
vendor_id=test_vendor.id,
|
|
tier="essential",
|
|
status=SubscriptionStatus.ACTIVE,
|
|
stripe_customer_id="cus_test123",
|
|
stripe_subscription_id="sub_test123",
|
|
period_start=datetime.now(timezone.utc),
|
|
period_end=datetime.now(timezone.utc),
|
|
)
|
|
db.add(subscription)
|
|
db.commit()
|
|
db.refresh(subscription)
|
|
return subscription
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_stripe_event():
|
|
"""Create a mock Stripe event."""
|
|
event = MagicMock()
|
|
event.id = "evt_test123"
|
|
event.type = "customer.created"
|
|
event.data.object = {"id": "cus_test123"}
|
|
return event
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_checkout_event():
|
|
"""Create a mock checkout.session.completed event."""
|
|
event = MagicMock()
|
|
event.id = "evt_checkout123"
|
|
event.type = "checkout.session.completed"
|
|
event.data.object.id = "cs_test123"
|
|
event.data.object.customer = "cus_test123"
|
|
event.data.object.subscription = "sub_test123"
|
|
event.data.object.metadata = {}
|
|
return event
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_subscription_updated_event():
|
|
"""Create a mock customer.subscription.updated event."""
|
|
event = MagicMock()
|
|
event.id = "evt_subupdated123"
|
|
event.type = "customer.subscription.updated"
|
|
event.data.object.id = "sub_test123"
|
|
event.data.object.customer = "cus_test123"
|
|
event.data.object.status = "active"
|
|
event.data.object.current_period_start = int(datetime.now(timezone.utc).timestamp())
|
|
event.data.object.current_period_end = int(datetime.now(timezone.utc).timestamp())
|
|
event.data.object.cancel_at_period_end = False
|
|
event.data.object.items.data = []
|
|
event.data.object.metadata = {}
|
|
return event
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_subscription_deleted_event():
|
|
"""Create a mock customer.subscription.deleted event."""
|
|
event = MagicMock()
|
|
event.id = "evt_subdeleted123"
|
|
event.type = "customer.subscription.deleted"
|
|
event.data.object.id = "sub_test123"
|
|
event.data.object.customer = "cus_test123"
|
|
return event
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_invoice_paid_event():
|
|
"""Create a mock invoice.paid event."""
|
|
event = MagicMock()
|
|
event.id = "evt_invoicepaid123"
|
|
event.type = "invoice.paid"
|
|
event.data.object.id = "in_test123"
|
|
event.data.object.customer = "cus_test123"
|
|
event.data.object.payment_intent = "pi_test123"
|
|
event.data.object.number = "INV-001"
|
|
event.data.object.created = int(datetime.now(timezone.utc).timestamp())
|
|
event.data.object.subtotal = 4900
|
|
event.data.object.tax = 0
|
|
event.data.object.total = 4900
|
|
event.data.object.amount_paid = 4900
|
|
event.data.object.currency = "eur"
|
|
event.data.object.invoice_pdf = "https://stripe.com/invoice.pdf"
|
|
event.data.object.hosted_invoice_url = "https://invoice.stripe.com"
|
|
return event
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_payment_failed_event():
|
|
"""Create a mock invoice.payment_failed event."""
|
|
event = MagicMock()
|
|
event.id = "evt_paymentfailed123"
|
|
event.type = "invoice.payment_failed"
|
|
event.data.object.id = "in_test123"
|
|
event.data.object.customer = "cus_test123"
|
|
event.data.object.last_payment_error = {"message": "Card declined"}
|
|
return event
|