feat: add subscription and billing system with Stripe integration
- Add database models for subscription tiers, vendor subscriptions, add-ons, billing history, and webhook events - Implement BillingService for subscription operations - Implement StripeService for Stripe API operations - Implement StripeWebhookHandler for webhook event processing - Add vendor billing API endpoints for subscription management - Create vendor billing page with Alpine.js frontend - Add limit enforcement for products and team members - Add billing exceptions for proper error handling - Create comprehensive unit tests (40 tests passing) - Add subscription billing documentation 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
658
tests/unit/services/test_billing_service.py
Normal file
658
tests/unit/services/test_billing_service.py
Normal file
@@ -0,0 +1,658 @@
|
||||
# tests/unit/services/test_billing_service.py
|
||||
"""Unit tests for BillingService."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.exceptions import VendorNotFoundException
|
||||
from app.services.billing_service import (
|
||||
BillingService,
|
||||
NoActiveSubscriptionError,
|
||||
PaymentSystemNotConfiguredError,
|
||||
StripePriceNotConfiguredError,
|
||||
SubscriptionNotCancelledError,
|
||||
TierNotFoundError,
|
||||
)
|
||||
from models.database.subscription import (
|
||||
AddOnProduct,
|
||||
BillingHistory,
|
||||
SubscriptionStatus,
|
||||
SubscriptionTier,
|
||||
VendorAddOn,
|
||||
VendorSubscription,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceSubscription:
|
||||
"""Test suite for BillingService subscription operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
def test_get_subscription_with_tier_creates_if_not_exists(
|
||||
self, db, test_vendor, test_subscription_tier
|
||||
):
|
||||
"""Test get_subscription_with_tier creates subscription if needed."""
|
||||
subscription, tier = self.service.get_subscription_with_tier(db, test_vendor.id)
|
||||
|
||||
assert subscription is not None
|
||||
assert subscription.vendor_id == test_vendor.id
|
||||
assert tier is not None
|
||||
assert tier.code == subscription.tier
|
||||
|
||||
def test_get_subscription_with_tier_returns_existing(
|
||||
self, db, test_vendor, test_subscription
|
||||
):
|
||||
"""Test get_subscription_with_tier returns existing subscription."""
|
||||
# Note: test_subscription fixture already creates the tier
|
||||
subscription, tier = self.service.get_subscription_with_tier(db, test_vendor.id)
|
||||
|
||||
assert subscription.id == test_subscription.id
|
||||
assert tier.code == test_subscription.tier
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceTiers:
|
||||
"""Test suite for BillingService tier operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
def test_get_available_tiers(self, db, test_subscription_tiers):
|
||||
"""Test getting available tiers."""
|
||||
tier_list, tier_order = self.service.get_available_tiers(db, "essential")
|
||||
|
||||
assert len(tier_list) > 0
|
||||
assert "essential" in tier_order
|
||||
assert "professional" in tier_order
|
||||
|
||||
# Check tier has expected fields
|
||||
essential_tier = next(t for t in tier_list if t["code"] == "essential")
|
||||
assert essential_tier["is_current"] is True
|
||||
assert essential_tier["can_upgrade"] is False
|
||||
assert essential_tier["can_downgrade"] is False
|
||||
|
||||
professional_tier = next(t for t in tier_list if t["code"] == "professional")
|
||||
assert professional_tier["can_upgrade"] is True
|
||||
assert professional_tier["can_downgrade"] is False
|
||||
|
||||
def test_get_tier_by_code_success(self, db, test_subscription_tier):
|
||||
"""Test getting tier by code."""
|
||||
tier = self.service.get_tier_by_code(db, "essential")
|
||||
|
||||
assert tier.code == "essential"
|
||||
assert tier.is_active is True
|
||||
|
||||
def test_get_tier_by_code_not_found(self, db):
|
||||
"""Test getting non-existent tier raises error."""
|
||||
with pytest.raises(TierNotFoundError) as exc_info:
|
||||
self.service.get_tier_by_code(db, "nonexistent")
|
||||
|
||||
assert exc_info.value.tier_code == "nonexistent"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceCheckout:
|
||||
"""Test suite for BillingService checkout operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_checkout_session_stripe_not_configured(
|
||||
self, mock_stripe, db, test_vendor, test_subscription_tier
|
||||
):
|
||||
"""Test checkout fails when Stripe not configured."""
|
||||
mock_stripe.is_configured = False
|
||||
|
||||
with pytest.raises(PaymentSystemNotConfiguredError):
|
||||
self.service.create_checkout_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
tier_code="essential",
|
||||
is_annual=False,
|
||||
success_url="https://example.com/success",
|
||||
cancel_url="https://example.com/cancel",
|
||||
)
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_checkout_session_success(
|
||||
self, mock_stripe, db, test_vendor, test_subscription_tier_with_stripe
|
||||
):
|
||||
"""Test successful checkout session creation."""
|
||||
mock_stripe.is_configured = True
|
||||
mock_session = MagicMock()
|
||||
mock_session.url = "https://checkout.stripe.com/test"
|
||||
mock_session.id = "cs_test_123"
|
||||
mock_stripe.create_checkout_session.return_value = mock_session
|
||||
|
||||
result = self.service.create_checkout_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
tier_code="essential",
|
||||
is_annual=False,
|
||||
success_url="https://example.com/success",
|
||||
cancel_url="https://example.com/cancel",
|
||||
)
|
||||
|
||||
assert result["checkout_url"] == "https://checkout.stripe.com/test"
|
||||
assert result["session_id"] == "cs_test_123"
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_checkout_session_tier_not_found(
|
||||
self, mock_stripe, db, test_vendor
|
||||
):
|
||||
"""Test checkout fails with invalid tier."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
with pytest.raises(TierNotFoundError):
|
||||
self.service.create_checkout_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
tier_code="nonexistent",
|
||||
is_annual=False,
|
||||
success_url="https://example.com/success",
|
||||
cancel_url="https://example.com/cancel",
|
||||
)
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_checkout_session_no_price(
|
||||
self, mock_stripe, db, test_vendor, test_subscription_tier
|
||||
):
|
||||
"""Test checkout fails when tier has no Stripe price."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
with pytest.raises(StripePriceNotConfiguredError):
|
||||
self.service.create_checkout_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
tier_code="essential",
|
||||
is_annual=False,
|
||||
success_url="https://example.com/success",
|
||||
cancel_url="https://example.com/cancel",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServicePortal:
|
||||
"""Test suite for BillingService portal operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_portal_session_stripe_not_configured(self, mock_stripe, db, test_vendor):
|
||||
"""Test portal fails when Stripe not configured."""
|
||||
mock_stripe.is_configured = False
|
||||
|
||||
with pytest.raises(PaymentSystemNotConfiguredError):
|
||||
self.service.create_portal_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
return_url="https://example.com/billing",
|
||||
)
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_portal_session_no_subscription(self, mock_stripe, db, test_vendor):
|
||||
"""Test portal fails when no subscription exists."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
with pytest.raises(NoActiveSubscriptionError):
|
||||
self.service.create_portal_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
return_url="https://example.com/billing",
|
||||
)
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_create_portal_session_success(
|
||||
self, mock_stripe, db, test_vendor, test_active_subscription
|
||||
):
|
||||
"""Test successful portal session creation."""
|
||||
mock_stripe.is_configured = True
|
||||
mock_session = MagicMock()
|
||||
mock_session.url = "https://billing.stripe.com/portal"
|
||||
mock_stripe.create_portal_session.return_value = mock_session
|
||||
|
||||
result = self.service.create_portal_session(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
return_url="https://example.com/billing",
|
||||
)
|
||||
|
||||
assert result["portal_url"] == "https://billing.stripe.com/portal"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceInvoices:
|
||||
"""Test suite for BillingService invoice operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
def test_get_invoices_empty(self, db, test_vendor):
|
||||
"""Test getting invoices when none exist."""
|
||||
invoices, total = self.service.get_invoices(db, test_vendor.id)
|
||||
|
||||
assert invoices == []
|
||||
assert total == 0
|
||||
|
||||
def test_get_invoices_with_data(self, db, test_vendor, test_billing_history):
|
||||
"""Test getting invoices returns data."""
|
||||
invoices, total = self.service.get_invoices(db, test_vendor.id)
|
||||
|
||||
assert len(invoices) == 1
|
||||
assert total == 1
|
||||
assert invoices[0].invoice_number == "INV-001"
|
||||
|
||||
def test_get_invoices_pagination(self, db, test_vendor, test_multiple_invoices):
|
||||
"""Test invoice pagination."""
|
||||
# Get first page
|
||||
page1, total = self.service.get_invoices(db, test_vendor.id, skip=0, limit=2)
|
||||
assert len(page1) == 2
|
||||
assert total == 5
|
||||
|
||||
# Get second page
|
||||
page2, _ = self.service.get_invoices(db, test_vendor.id, skip=2, limit=2)
|
||||
assert len(page2) == 2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceAddons:
|
||||
"""Test suite for BillingService addon operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
def test_get_available_addons_empty(self, db):
|
||||
"""Test getting addons when none exist."""
|
||||
addons = self.service.get_available_addons(db)
|
||||
assert addons == []
|
||||
|
||||
def test_get_available_addons_with_data(self, db, test_addon_products):
|
||||
"""Test getting all available addons."""
|
||||
addons = self.service.get_available_addons(db)
|
||||
|
||||
assert len(addons) == 3
|
||||
assert all(addon.is_active for addon in addons)
|
||||
|
||||
def test_get_available_addons_by_category(self, db, test_addon_products):
|
||||
"""Test filtering addons by category."""
|
||||
domain_addons = self.service.get_available_addons(db, category="domain")
|
||||
|
||||
assert len(domain_addons) == 1
|
||||
assert domain_addons[0].category == "domain"
|
||||
|
||||
def test_get_vendor_addons_empty(self, db, test_vendor):
|
||||
"""Test getting vendor addons when none purchased."""
|
||||
addons = self.service.get_vendor_addons(db, test_vendor.id)
|
||||
assert addons == []
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceCancellation:
|
||||
"""Test suite for BillingService cancellation operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_cancel_subscription_no_subscription(
|
||||
self, mock_stripe, db, test_vendor
|
||||
):
|
||||
"""Test cancel fails when no subscription."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
with pytest.raises(NoActiveSubscriptionError):
|
||||
self.service.cancel_subscription(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
reason="Test reason",
|
||||
immediately=False,
|
||||
)
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_cancel_subscription_success(
|
||||
self, mock_stripe, db, test_vendor, test_active_subscription
|
||||
):
|
||||
"""Test successful subscription cancellation."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
result = self.service.cancel_subscription(
|
||||
db=db,
|
||||
vendor_id=test_vendor.id,
|
||||
reason="Too expensive",
|
||||
immediately=False,
|
||||
)
|
||||
|
||||
assert result["message"] == "Subscription cancelled successfully"
|
||||
assert test_active_subscription.cancelled_at is not None
|
||||
assert test_active_subscription.cancellation_reason == "Too expensive"
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_reactivate_subscription_not_cancelled(
|
||||
self, mock_stripe, db, test_vendor, test_active_subscription
|
||||
):
|
||||
"""Test reactivate fails when subscription not cancelled."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
with pytest.raises(SubscriptionNotCancelledError):
|
||||
self.service.reactivate_subscription(db, test_vendor.id)
|
||||
|
||||
@patch("app.services.billing_service.stripe_service")
|
||||
def test_reactivate_subscription_success(
|
||||
self, mock_stripe, db, test_vendor, test_cancelled_subscription
|
||||
):
|
||||
"""Test successful subscription reactivation."""
|
||||
mock_stripe.is_configured = True
|
||||
|
||||
result = self.service.reactivate_subscription(db, test_vendor.id)
|
||||
|
||||
assert result["message"] == "Subscription reactivated successfully"
|
||||
assert test_cancelled_subscription.cancelled_at is None
|
||||
assert test_cancelled_subscription.cancellation_reason is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.billing
|
||||
class TestBillingServiceVendor:
|
||||
"""Test suite for BillingService vendor operations."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Initialize service instance before each test."""
|
||||
self.service = BillingService()
|
||||
|
||||
def test_get_vendor_success(self, db, test_vendor):
|
||||
"""Test getting vendor by ID."""
|
||||
vendor = self.service.get_vendor(db, test_vendor.id)
|
||||
|
||||
assert vendor.id == test_vendor.id
|
||||
|
||||
def test_get_vendor_not_found(self, db):
|
||||
"""Test getting non-existent vendor raises error."""
|
||||
with pytest.raises(VendorNotFoundException):
|
||||
self.service.get_vendor(db, 99999)
|
||||
|
||||
|
||||
# ==================== Fixtures ====================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_subscription_tier(db):
|
||||
"""Create a basic subscription tier."""
|
||||
tier = SubscriptionTier(
|
||||
code="essential",
|
||||
name="Essential",
|
||||
description="Essential plan",
|
||||
price_monthly_cents=4900,
|
||||
price_annual_cents=49000,
|
||||
orders_per_month=100,
|
||||
products_limit=200,
|
||||
team_members=1,
|
||||
features=["basic_support"],
|
||||
display_order=1,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
)
|
||||
db.add(tier)
|
||||
db.commit()
|
||||
db.refresh(tier)
|
||||
return tier
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_subscription_tier_with_stripe(db):
|
||||
"""Create a subscription tier with Stripe configuration."""
|
||||
tier = SubscriptionTier(
|
||||
code="essential",
|
||||
name="Essential",
|
||||
description="Essential plan",
|
||||
price_monthly_cents=4900,
|
||||
price_annual_cents=49000,
|
||||
orders_per_month=100,
|
||||
products_limit=200,
|
||||
team_members=1,
|
||||
features=["basic_support"],
|
||||
display_order=1,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
stripe_product_id="prod_test123",
|
||||
stripe_price_monthly_id="price_test123",
|
||||
stripe_price_annual_id="price_test456",
|
||||
)
|
||||
db.add(tier)
|
||||
db.commit()
|
||||
db.refresh(tier)
|
||||
return tier
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_subscription_tiers(db):
|
||||
"""Create multiple subscription tiers."""
|
||||
tiers = [
|
||||
SubscriptionTier(
|
||||
code="essential",
|
||||
name="Essential",
|
||||
price_monthly_cents=4900,
|
||||
display_order=1,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
),
|
||||
SubscriptionTier(
|
||||
code="professional",
|
||||
name="Professional",
|
||||
price_monthly_cents=9900,
|
||||
display_order=2,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
),
|
||||
SubscriptionTier(
|
||||
code="business",
|
||||
name="Business",
|
||||
price_monthly_cents=19900,
|
||||
display_order=3,
|
||||
is_active=True,
|
||||
is_public=True,
|
||||
),
|
||||
]
|
||||
db.add_all(tiers)
|
||||
db.commit()
|
||||
for tier in tiers:
|
||||
db.refresh(tier)
|
||||
return tiers
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_subscription(db, test_vendor):
|
||||
"""Create a basic subscription for testing."""
|
||||
# Create tier first
|
||||
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,
|
||||
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 test_cancelled_subscription(db, test_vendor):
|
||||
"""Create a cancelled subscription."""
|
||||
# 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),
|
||||
cancelled_at=datetime.now(timezone.utc),
|
||||
cancellation_reason="Too expensive",
|
||||
)
|
||||
db.add(subscription)
|
||||
db.commit()
|
||||
db.refresh(subscription)
|
||||
return subscription
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_billing_history(db, test_vendor):
|
||||
"""Create a billing history record."""
|
||||
record = BillingHistory(
|
||||
vendor_id=test_vendor.id,
|
||||
stripe_invoice_id="in_test123",
|
||||
invoice_number="INV-001",
|
||||
invoice_date=datetime.now(timezone.utc),
|
||||
subtotal_cents=4900,
|
||||
tax_cents=0,
|
||||
total_cents=4900,
|
||||
amount_paid_cents=4900,
|
||||
currency="EUR",
|
||||
status="paid",
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
db.refresh(record)
|
||||
return record
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_multiple_invoices(db, test_vendor):
|
||||
"""Create multiple billing history records."""
|
||||
records = []
|
||||
for i in range(5):
|
||||
record = BillingHistory(
|
||||
vendor_id=test_vendor.id,
|
||||
stripe_invoice_id=f"in_test{i}",
|
||||
invoice_number=f"INV-{i:03d}",
|
||||
invoice_date=datetime.now(timezone.utc),
|
||||
subtotal_cents=4900,
|
||||
tax_cents=0,
|
||||
total_cents=4900,
|
||||
amount_paid_cents=4900,
|
||||
currency="EUR",
|
||||
status="paid",
|
||||
)
|
||||
records.append(record)
|
||||
db.add_all(records)
|
||||
db.commit()
|
||||
return records
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_addon_products(db):
|
||||
"""Create test addon products."""
|
||||
addons = [
|
||||
AddOnProduct(
|
||||
code="domain",
|
||||
name="Custom Domain",
|
||||
category="domain",
|
||||
price_cents=1500,
|
||||
billing_period="annual",
|
||||
display_order=1,
|
||||
is_active=True,
|
||||
),
|
||||
AddOnProduct(
|
||||
code="email_5",
|
||||
name="5 Email Addresses",
|
||||
category="email",
|
||||
price_cents=500,
|
||||
billing_period="monthly",
|
||||
quantity_value=5,
|
||||
display_order=2,
|
||||
is_active=True,
|
||||
),
|
||||
AddOnProduct(
|
||||
code="email_10",
|
||||
name="10 Email Addresses",
|
||||
category="email",
|
||||
price_cents=900,
|
||||
billing_period="monthly",
|
||||
quantity_value=10,
|
||||
display_order=3,
|
||||
is_active=True,
|
||||
),
|
||||
]
|
||||
db.add_all(addons)
|
||||
db.commit()
|
||||
for addon in addons:
|
||||
db.refresh(addon)
|
||||
return addons
|
||||
393
tests/unit/services/test_stripe_webhook_handler.py
Normal file
393
tests/unit/services/test_stripe_webhook_handler.py
Normal file
@@ -0,0 +1,393 @@
|
||||
# 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.services.stripe_webhook_handler import StripeWebhookHandler
|
||||
from models.database.subscription 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.services.stripe_webhook_handler.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
|
||||
Reference in New Issue
Block a user