sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(billing): migrate frontend templates to feature provider system

Browse Source 
Replace hardcoded subscription fields (orders_limit, products_limit,
team_members_limit) across 5 frontend pages with dynamic feature
provider APIs. Add admin convenience endpoint for store subscription
lookup. Remove legacy stubs (StoreSubscription, FeatureCode, Feature,
TIER_LIMITS, FeatureInfo, FeatureUpgradeInfo) and schema aliases.

Pages updated:
- Admin subscriptions: dynamic feature overrides editor
- Admin tiers: correct feature catalog/limits API URLs
- Store billing: usage metrics from /store/billing/usage
- Merchant subscription detail: tier.feature_limits rendering
- Admin store detail: new GET /admin/subscriptions/store/{id} endpoint

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-07 15:18:16 +01:00
parent 922616c9e3
commit 1db7e8a087
19 changed files with 2508 additions and 1205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
tests/unit/services/test_billing_service.py
View File

@@ -6,7 +6,7 @@ from unittest.mock import MagicMock, patch
import pytest
from app.modules.tenancy.exceptions import VendorNotFoundException
from app.modules.tenancy.exceptions import StoreNotFoundException
from app.modules.billing.services.billing_service import (
BillingService,
NoActiveSubscriptionError,
@@ -18,10 +18,10 @@ from app.modules.billing.services.billing_service import (
from app.modules.billing.models import (
AddOnProduct,
BillingHistory,
MerchantSubscription,
SubscriptionStatus,
SubscriptionTier,
VendorAddOn,
VendorSubscription,
StoreAddOn,
)
@@ -35,22 +35,22 @@ class TestBillingServiceSubscription:
self.service = BillingService()
def test_get_subscription_with_tier_creates_if_not_exists(
self, db, test_vendor, test_subscription_tier
self, db, test_store, test_subscription_tier
):
"""Test get_subscription_with_tier creates subscription if needed."""
subscription, tier = self.service.get_subscription_with_tier(db, test_vendor.id)
subscription, tier = self.service.get_subscription_with_tier(db, test_store.id)
assert subscription is not None
assert subscription.vendor_id == test_vendor.id
assert subscription.store_id == test_store.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
self, db, test_store, 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)
subscription, tier = self.service.get_subscription_with_tier(db, test_store.id)
assert subscription.id == test_subscription.id
assert tier.code == test_subscription.tier
@@ -109,7 +109,7 @@ class TestBillingServiceCheckout:
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_checkout_session_stripe_not_configured(
self, mock_stripe, db, test_vendor, test_subscription_tier
self, mock_stripe, db, test_store, test_subscription_tier
):
"""Test checkout fails when Stripe not configured."""
mock_stripe.is_configured = False
@@ -117,7 +117,7 @@ class TestBillingServiceCheckout:
with pytest.raises(PaymentSystemNotConfiguredError):
self.service.create_checkout_session(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
tier_code="essential",
is_annual=False,
success_url="https://example.com/success",
@@ -126,7 +126,7 @@ class TestBillingServiceCheckout:
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_checkout_session_success(
self, mock_stripe, db, test_vendor, test_subscription_tier_with_stripe
self, mock_stripe, db, test_store, test_subscription_tier_with_stripe
):
"""Test successful checkout session creation."""
mock_stripe.is_configured = True
@@ -137,7 +137,7 @@ class TestBillingServiceCheckout:
result = self.service.create_checkout_session(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
tier_code="essential",
is_annual=False,
success_url="https://example.com/success",
@@ -149,7 +149,7 @@ class TestBillingServiceCheckout:
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_checkout_session_tier_not_found(
self, mock_stripe, db, test_vendor
self, mock_stripe, db, test_store
):
"""Test checkout fails with invalid tier."""
mock_stripe.is_configured = True
@@ -157,7 +157,7 @@ class TestBillingServiceCheckout:
with pytest.raises(TierNotFoundError):
self.service.create_checkout_session(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
tier_code="nonexistent",
is_annual=False,
success_url="https://example.com/success",
@@ -166,7 +166,7 @@ class TestBillingServiceCheckout:
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_checkout_session_no_price(
self, mock_stripe, db, test_vendor, test_subscription_tier
self, mock_stripe, db, test_store, test_subscription_tier
):
"""Test checkout fails when tier has no Stripe price."""
mock_stripe.is_configured = True
@@ -174,7 +174,7 @@ class TestBillingServiceCheckout:
with pytest.raises(StripePriceNotConfiguredError):
self.service.create_checkout_session(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
tier_code="essential",
is_annual=False,
success_url="https://example.com/success",
@@ -192,32 +192,32 @@ class TestBillingServicePortal:
self.service = BillingService()
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_portal_session_stripe_not_configured(self, mock_stripe, db, test_vendor):
def test_create_portal_session_stripe_not_configured(self, mock_stripe, db, test_store):
"""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,
store_id=test_store.id,
return_url="https://example.com/billing",
)
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_portal_session_no_subscription(self, mock_stripe, db, test_vendor):
def test_create_portal_session_no_subscription(self, mock_stripe, db, test_store):
"""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,
store_id=test_store.id,
return_url="https://example.com/billing",
)
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_create_portal_session_success(
self, mock_stripe, db, test_vendor, test_active_subscription
self, mock_stripe, db, test_store, test_active_subscription
):
"""Test successful portal session creation."""
mock_stripe.is_configured = True
@@ -227,7 +227,7 @@ class TestBillingServicePortal:
result = self.service.create_portal_session(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
return_url="https://example.com/billing",
)
@@ -243,30 +243,30 @@ class TestBillingServiceInvoices:
"""Initialize service instance before each test."""
self.service = BillingService()
def test_get_invoices_empty(self, db, test_vendor):
def test_get_invoices_empty(self, db, test_store):
"""Test getting invoices when none exist."""
invoices, total = self.service.get_invoices(db, test_vendor.id)
invoices, total = self.service.get_invoices(db, test_store.id)
assert invoices == []
assert total == 0
def test_get_invoices_with_data(self, db, test_vendor, test_billing_history):
def test_get_invoices_with_data(self, db, test_store, test_billing_history):
"""Test getting invoices returns data."""
invoices, total = self.service.get_invoices(db, test_vendor.id)
invoices, total = self.service.get_invoices(db, test_store.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):
def test_get_invoices_pagination(self, db, test_store, test_multiple_invoices):
"""Test invoice pagination."""
# Get first page
page1, total = self.service.get_invoices(db, test_vendor.id, skip=0, limit=2)
page1, total = self.service.get_invoices(db, test_store.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)
page2, _ = self.service.get_invoices(db, test_store.id, skip=2, limit=2)
assert len(page2) == 2
@@ -298,9 +298,9 @@ class TestBillingServiceAddons:
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)
def test_get_store_addons_empty(self, db, test_store):
"""Test getting store addons when none purchased."""
addons = self.service.get_store_addons(db, test_store.id)
assert addons == []
@@ -315,7 +315,7 @@ class TestBillingServiceCancellation:
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_cancel_subscription_no_subscription(
self, mock_stripe, db, test_vendor
self, mock_stripe, db, test_store
):
"""Test cancel fails when no subscription."""
mock_stripe.is_configured = True
@@ -323,21 +323,21 @@ class TestBillingServiceCancellation:
with pytest.raises(NoActiveSubscriptionError):
self.service.cancel_subscription(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
reason="Test reason",
immediately=False,
)
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_cancel_subscription_success(
self, mock_stripe, db, test_vendor, test_active_subscription
self, mock_stripe, db, test_store, test_active_subscription
):
"""Test successful subscription cancellation."""
mock_stripe.is_configured = True
result = self.service.cancel_subscription(
db=db,
vendor_id=test_vendor.id,
store_id=test_store.id,
reason="Too expensive",
immediately=False,
)
@@ -348,22 +348,22 @@ class TestBillingServiceCancellation:
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_reactivate_subscription_not_cancelled(
self, mock_stripe, db, test_vendor, test_active_subscription
self, mock_stripe, db, test_store, 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)
self.service.reactivate_subscription(db, test_store.id)
@patch("app.modules.billing.services.billing_service.stripe_service")
def test_reactivate_subscription_success(
self, mock_stripe, db, test_vendor, test_cancelled_subscription
self, mock_stripe, db, test_store, test_cancelled_subscription
):
"""Test successful subscription reactivation."""
mock_stripe.is_configured = True
result = self.service.reactivate_subscription(db, test_vendor.id)
result = self.service.reactivate_subscription(db, test_store.id)
assert result["message"] == "Subscription reactivated successfully"
assert test_cancelled_subscription.cancelled_at is None
@@ -372,23 +372,23 @@ class TestBillingServiceCancellation:
@pytest.mark.unit
@pytest.mark.billing
class TestBillingServiceVendor:
"""Test suite for BillingService vendor operations."""
class TestBillingServiceStore:
"""Test suite for BillingService store 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)
def test_get_store_success(self, db, test_store):
"""Test getting store by ID."""
store = self.service.get_store(db, test_store.id)
assert vendor.id == test_vendor.id
assert store.id == test_store.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)
def test_get_store_not_found(self, db):
"""Test getting non-existent store raises error."""
with pytest.raises(StoreNotFoundException):
self.service.get_store(db, 99999)
# ==================== Fixtures ====================
@@ -480,7 +480,7 @@ def test_subscription_tiers(db):
@pytest.fixture
def test_subscription(db, test_vendor):
def test_subscription(db, test_store):
"""Create a basic subscription for testing."""
# Create tier first
tier = SubscriptionTier(
@@ -494,8 +494,8 @@ def test_subscription(db, test_vendor):
db.add(tier)
db.commit()
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = MerchantSubscription(
store_id=test_store.id,
tier="essential",
status=SubscriptionStatus.ACTIVE,
period_start=datetime.now(timezone.utc),
@@ -508,7 +508,7 @@ def test_subscription(db, test_vendor):
@pytest.fixture
def test_active_subscription(db, test_vendor):
def test_active_subscription(db, test_store):
"""Create an active subscription with Stripe IDs."""
# Create tier first if not exists
tier = db.query(SubscriptionTier).filter(SubscriptionTier.code == "essential").first()
@@ -524,8 +524,8 @@ def test_active_subscription(db, test_vendor):
db.add(tier)
db.commit()
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = MerchantSubscription(
store_id=test_store.id,
tier="essential",
status=SubscriptionStatus.ACTIVE,
stripe_customer_id="cus_test123",
@@ -540,7 +540,7 @@ def test_active_subscription(db, test_vendor):
@pytest.fixture
def test_cancelled_subscription(db, test_vendor):
def test_cancelled_subscription(db, test_store):
"""Create a cancelled subscription."""
# Create tier first if not exists
tier = db.query(SubscriptionTier).filter(SubscriptionTier.code == "essential").first()
@@ -556,8 +556,8 @@ def test_cancelled_subscription(db, test_vendor):
db.add(tier)
db.commit()
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = MerchantSubscription(
store_id=test_store.id,
tier="essential",
status=SubscriptionStatus.ACTIVE,
stripe_customer_id="cus_test123",
@@ -574,10 +574,10 @@ def test_cancelled_subscription(db, test_vendor):
@pytest.fixture
def test_billing_history(db, test_vendor):
def test_billing_history(db, test_store):
"""Create a billing history record."""
record = BillingHistory(
vendor_id=test_vendor.id,
store_id=test_store.id,
stripe_invoice_id="in_test123",
invoice_number="INV-001",
invoice_date=datetime.now(timezone.utc),
@@ -595,12 +595,12 @@ def test_billing_history(db, test_vendor):
@pytest.fixture
def test_multiple_invoices(db, test_vendor):
def test_multiple_invoices(db, test_store):
"""Create multiple billing history records."""
records = []
for i in range(5):
record = BillingHistory(
vendor_id=test_vendor.id,
store_id=test_store.id,
stripe_invoice_id=f"in_test{i}",
invoice_number=f"INV-{i:03d}",
invoice_date=datetime.now(timezone.utc),

93
tests/unit/services/test_feature_service.py
View File

@@ -5,8 +5,7 @@ import pytest
from app.modules.billing.exceptions import FeatureNotFoundError, InvalidFeatureCodesError, TierNotFoundError
from app.modules.billing.services.feature_service import FeatureService, feature_service
from app.modules.billing.models import Feature
from app.modules.billing.models import SubscriptionTier, VendorSubscription
from app.modules.billing.models import SubscriptionTier, MerchantSubscription
@pytest.mark.unit
@@ -18,27 +17,27 @@ class TestFeatureServiceAvailability:
"""Initialize service instance before each test."""
self.service = FeatureService()
def test_has_feature_true(self, db, test_vendor_with_subscription):
def test_has_feature_true(self, db, test_store_with_subscription):
"""Test has_feature returns True for available feature."""
vendor_id = test_vendor_with_subscription.id
result = self.service.has_feature(db, vendor_id, "basic_reports")
store_id = test_store_with_subscription.id
result = self.service.has_feature(db, store_id, "basic_reports")
assert result is True
def test_has_feature_false(self, db, test_vendor_with_subscription):
def test_has_feature_false(self, db, test_store_with_subscription):
"""Test has_feature returns False for unavailable feature."""
vendor_id = test_vendor_with_subscription.id
result = self.service.has_feature(db, vendor_id, "api_access")
store_id = test_store_with_subscription.id
result = self.service.has_feature(db, store_id, "api_access")
assert result is False
def test_has_feature_no_subscription(self, db, test_vendor):
"""Test has_feature returns False for vendor without subscription."""
result = self.service.has_feature(db, test_vendor.id, "basic_reports")
def test_has_feature_no_subscription(self, db, test_store):
"""Test has_feature returns False for store without subscription."""
result = self.service.has_feature(db, test_store.id, "basic_reports")
assert result is False
def test_get_vendor_feature_codes(self, db, test_vendor_with_subscription):
"""Test getting all feature codes for vendor."""
vendor_id = test_vendor_with_subscription.id
features = self.service.get_vendor_feature_codes(db, vendor_id)
def test_get_store_feature_codes(self, db, test_store_with_subscription):
"""Test getting all feature codes for store."""
store_id = test_store_with_subscription.id
features = self.service.get_store_feature_codes(db, store_id)
assert isinstance(features, set)
assert "basic_reports" in features
@@ -54,10 +53,10 @@ class TestFeatureServiceListing:
"""Initialize service instance before each test."""
self.service = FeatureService()
def test_get_vendor_features(self, db, test_vendor_with_subscription, test_features):
def test_get_store_features(self, db, test_store_with_subscription, test_features):
"""Test getting all features with availability."""
vendor_id = test_vendor_with_subscription.id
features = self.service.get_vendor_features(db, vendor_id)
store_id = test_store_with_subscription.id
features = self.service.get_store_features(db, store_id)
assert len(features) > 0
basic_reports = next((f for f in features if f.code == "basic_reports"), None)
@@ -68,30 +67,30 @@ class TestFeatureServiceListing:
assert api_access is not None
assert api_access.is_available is False
def test_get_vendor_features_by_category(
self, db, test_vendor_with_subscription, test_features
def test_get_store_features_by_category(
self, db, test_store_with_subscription, test_features
):
"""Test filtering features by category."""
vendor_id = test_vendor_with_subscription.id
features = self.service.get_vendor_features(db, vendor_id, category="analytics")
store_id = test_store_with_subscription.id
features = self.service.get_store_features(db, store_id, category="analytics")
assert all(f.category == "analytics" for f in features)
def test_get_vendor_features_available_only(
self, db, test_vendor_with_subscription, test_features
def test_get_store_features_available_only(
self, db, test_store_with_subscription, test_features
):
"""Test getting only available features."""
vendor_id = test_vendor_with_subscription.id
features = self.service.get_vendor_features(
db, vendor_id, include_unavailable=False
store_id = test_store_with_subscription.id
features = self.service.get_store_features(
db, store_id, include_unavailable=False
)
assert all(f.is_available for f in features)
def test_get_available_feature_codes(self, db, test_vendor_with_subscription):
def test_get_available_feature_codes(self, db, test_store_with_subscription):
"""Test getting simple list of available codes."""
vendor_id = test_vendor_with_subscription.id
codes = self.service.get_available_feature_codes(db, vendor_id)
store_id = test_store_with_subscription.id
codes = self.service.get_available_feature_codes(db, store_id)
assert isinstance(codes, list)
assert "basic_reports" in codes
@@ -159,28 +158,28 @@ class TestFeatureServiceCache:
"""Initialize service instance before each test."""
self.service = FeatureService()
def test_cache_invalidation(self, db, test_vendor_with_subscription):
"""Test cache invalidation for vendor."""
vendor_id = test_vendor_with_subscription.id
def test_cache_invalidation(self, db, test_store_with_subscription):
"""Test cache invalidation for store."""
store_id = test_store_with_subscription.id
# Prime the cache
self.service.get_vendor_feature_codes(db, vendor_id)
assert self.service._cache.get(vendor_id) is not None
self.service.get_store_feature_codes(db, store_id)
assert self.service._cache.get(store_id) is not None
# Invalidate
self.service.invalidate_vendor_cache(vendor_id)
assert self.service._cache.get(vendor_id) is None
self.service.invalidate_store_cache(store_id)
assert self.service._cache.get(store_id) is None
def test_cache_invalidate_all(self, db, test_vendor_with_subscription):
def test_cache_invalidate_all(self, db, test_store_with_subscription):
"""Test invalidating entire cache."""
vendor_id = test_vendor_with_subscription.id
store_id = test_store_with_subscription.id
# Prime the cache
self.service.get_vendor_feature_codes(db, vendor_id)
self.service.get_store_feature_codes(db, store_id)
# Invalidate all
self.service.invalidate_all_cache()
assert self.service._cache.get(vendor_id) is None
assert self.service._cache.get(store_id) is None
@pytest.mark.unit
@@ -301,14 +300,14 @@ def test_subscription_tiers(db):
@pytest.fixture
def test_vendor_with_subscription(db, test_vendor, test_subscription_tiers):
"""Create a vendor with an active subscription."""
def test_store_with_subscription(db, test_store, test_subscription_tiers):
"""Create a store with an active subscription."""
from datetime import datetime, timezone
essential_tier = test_subscription_tiers[0] # Use the essential tier from tiers list
now = datetime.now(timezone.utc)
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = StoreSubscription(
store_id=test_store.id,
tier="essential",
tier_id=essential_tier.id,
status="active",
@@ -318,8 +317,8 @@ def test_vendor_with_subscription(db, test_vendor, test_subscription_tiers):
)
db.add(subscription)
db.commit()
db.refresh(test_vendor)
return test_vendor
db.refresh(test_store)
return test_store
@pytest.fixture

36
tests/unit/services/test_stripe_webhook_handler.py
View File

@@ -9,10 +9,10 @@ import pytest
from app.handlers.stripe_webhook import StripeWebhookHandler
from app.modules.billing.models import (
BillingHistory,
MerchantSubscription,
StripeWebhookEvent,
SubscriptionStatus,
SubscriptionTier,
VendorSubscription,
)
@@ -61,7 +61,7 @@ class TestStripeWebhookHandlerCheckout:
@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
self, mock_stripe_retrieve, db, test_store, test_subscription, mock_checkout_event
):
"""Test successful checkout completion."""
# Mock Stripe subscription retrieve
@@ -71,7 +71,7 @@ class TestStripeWebhookHandlerCheckout:
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)}
mock_checkout_event.data.object.metadata = {"store_id": str(test_store.id)}
result = self.handler.handle_event(db, mock_checkout_event)
@@ -80,15 +80,15 @@ class TestStripeWebhookHandlerCheckout:
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."""
def test_handle_checkout_completed_no_store_id(self, db, mock_checkout_event):
"""Test checkout with missing store_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"
assert result["result"]["reason"] == "no store_id"
@pytest.mark.unit
@@ -101,7 +101,7 @@ class TestStripeWebhookHandlerSubscription:
self.handler = StripeWebhookHandler()
def test_handle_subscription_updated_status_change(
self, db, test_vendor, test_active_subscription, mock_subscription_updated_event
self, db, test_store, test_active_subscription, mock_subscription_updated_event
):
"""Test subscription update changes status."""
result = self.handler.handle_event(db, mock_subscription_updated_event)
@@ -109,7 +109,7 @@ class TestStripeWebhookHandlerSubscription:
assert result["status"] == "processed"
def test_handle_subscription_deleted(
self, db, test_vendor, test_active_subscription, mock_subscription_deleted_event
self, db, test_store, test_active_subscription, mock_subscription_deleted_event
):
"""Test subscription deletion."""
result = self.handler.handle_event(db, mock_subscription_deleted_event)
@@ -129,7 +129,7 @@ class TestStripeWebhookHandlerInvoice:
self.handler = StripeWebhookHandler()
def test_handle_invoice_paid_creates_billing_record(
self, db, test_vendor, test_active_subscription, mock_invoice_paid_event
self, db, test_store, test_active_subscription, mock_invoice_paid_event
):
"""Test invoice.paid creates billing history record."""
result = self.handler.handle_event(db, mock_invoice_paid_event)
@@ -139,7 +139,7 @@ class TestStripeWebhookHandlerInvoice:
# Check billing record created
record = (
db.query(BillingHistory)
.filter(BillingHistory.vendor_id == test_vendor.id)
.filter(BillingHistory.store_id == test_store.id)
.first()
)
assert record is not None
@@ -147,7 +147,7 @@ class TestStripeWebhookHandlerInvoice:
assert record.total_cents == 4900
def test_handle_invoice_paid_resets_counters(
self, db, test_vendor, test_active_subscription, mock_invoice_paid_event
self, db, test_store, test_active_subscription, mock_invoice_paid_event
):
"""Test invoice.paid resets order counters."""
test_active_subscription.orders_this_period = 50
@@ -159,7 +159,7 @@ class TestStripeWebhookHandlerInvoice:
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
self, db, test_store, 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)
@@ -248,7 +248,7 @@ def test_subscription_tier(db):
@pytest.fixture
def test_subscription(db, test_vendor):
def test_subscription(db, test_store):
"""Create a basic subscription for testing."""
# Create tier first if not exists
tier = db.query(SubscriptionTier).filter(SubscriptionTier.code == "essential").first()
@@ -264,8 +264,8 @@ def test_subscription(db, test_vendor):
db.add(tier)
db.commit()
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = MerchantSubscription(
store_id=test_store.id,
tier="essential",
status=SubscriptionStatus.TRIAL,
period_start=datetime.now(timezone.utc),
@@ -278,7 +278,7 @@ def test_subscription(db, test_vendor):
@pytest.fixture
def test_active_subscription(db, test_vendor):
def test_active_subscription(db, test_store):
"""Create an active subscription with Stripe IDs."""
# Create tier first if not exists
tier = db.query(SubscriptionTier).filter(SubscriptionTier.code == "essential").first()
@@ -294,8 +294,8 @@ def test_active_subscription(db, test_vendor):
db.add(tier)
db.commit()
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = MerchantSubscription(
store_id=test_store.id,
tier="essential",
status=SubscriptionStatus.ACTIVE,
stripe_customer_id="cus_test123",

150
tests/unit/services/test_usage_service.py
View File

@@ -5,32 +5,32 @@ import pytest
from app.modules.analytics.services.usage_service import UsageService, usage_service
from app.modules.catalog.models import Product
from app.modules.billing.models import SubscriptionTier, VendorSubscription
from app.modules.tenancy.models import VendorUser
from app.modules.billing.models import SubscriptionTier, MerchantSubscription
from app.modules.tenancy.models import StoreUser
@pytest.mark.unit
@pytest.mark.usage
class TestUsageServiceGetUsage:
"""Test suite for get_vendor_usage operation."""
"""Test suite for get_store_usage operation."""
def setup_method(self):
"""Initialize service instance before each test."""
self.service = UsageService()
def test_get_vendor_usage_basic(self, db, test_vendor_with_subscription):
def test_get_store_usage_basic(self, db, test_store_with_subscription):
"""Test getting basic usage data."""
vendor_id = test_vendor_with_subscription.id
usage = self.service.get_vendor_usage(db, vendor_id)
store_id = test_store_with_subscription.id
usage = self.service.get_store_usage(db, store_id)
assert usage.tier.code == "essential"
assert usage.tier.name == "Essential"
assert len(usage.usage) == 3
def test_get_vendor_usage_metrics(self, db, test_vendor_with_subscription):
def test_get_store_usage_metrics(self, db, test_store_with_subscription):
"""Test usage metrics are calculated correctly."""
vendor_id = test_vendor_with_subscription.id
usage = self.service.get_vendor_usage(db, vendor_id)
store_id = test_store_with_subscription.id
usage = self.service.get_store_usage(db, store_id)
orders_metric = next((m for m in usage.usage if m.name == "orders"), None)
assert orders_metric is not None
@@ -39,39 +39,39 @@ class TestUsageServiceGetUsage:
assert orders_metric.percentage == 10.0
assert orders_metric.is_unlimited is False
def test_get_vendor_usage_at_limit(self, db, test_vendor_at_limit):
def test_get_store_usage_at_limit(self, db, test_store_at_limit):
"""Test usage shows at limit correctly."""
vendor_id = test_vendor_at_limit.id
usage = self.service.get_vendor_usage(db, vendor_id)
store_id = test_store_at_limit.id
usage = self.service.get_store_usage(db, store_id)
orders_metric = next((m for m in usage.usage if m.name == "orders"), None)
assert orders_metric.is_at_limit is True
assert usage.has_limits_reached is True
def test_get_vendor_usage_approaching_limit(self, db, test_vendor_approaching_limit):
def test_get_store_usage_approaching_limit(self, db, test_store_approaching_limit):
"""Test usage shows approaching limit correctly."""
vendor_id = test_vendor_approaching_limit.id
usage = self.service.get_vendor_usage(db, vendor_id)
store_id = test_store_approaching_limit.id
usage = self.service.get_store_usage(db, store_id)
orders_metric = next((m for m in usage.usage if m.name == "orders"), None)
assert orders_metric.is_approaching_limit is True
assert usage.has_limits_approaching is True
def test_get_vendor_usage_upgrade_available(
self, db, test_vendor_with_subscription, test_professional_tier
def test_get_store_usage_upgrade_available(
self, db, test_store_with_subscription, test_professional_tier
):
"""Test upgrade info when not on highest tier."""
vendor_id = test_vendor_with_subscription.id
usage = self.service.get_vendor_usage(db, vendor_id)
store_id = test_store_with_subscription.id
usage = self.service.get_store_usage(db, store_id)
assert usage.upgrade_available is True
assert usage.upgrade_tier is not None
assert usage.upgrade_tier.code == "professional"
def test_get_vendor_usage_highest_tier(self, db, test_vendor_on_professional):
def test_get_store_usage_highest_tier(self, db, test_store_on_professional):
"""Test no upgrade when on highest tier."""
vendor_id = test_vendor_on_professional.id
usage = self.service.get_vendor_usage(db, vendor_id)
store_id = test_store_on_professional.id
usage = self.service.get_store_usage(db, store_id)
assert usage.tier.is_highest_tier is True
assert usage.upgrade_available is False
@@ -87,28 +87,28 @@ class TestUsageServiceCheckLimit:
"""Initialize service instance before each test."""
self.service = UsageService()
def test_check_orders_limit_can_proceed(self, db, test_vendor_with_subscription):
def test_check_orders_limit_can_proceed(self, db, test_store_with_subscription):
"""Test checking orders limit when under limit."""
vendor_id = test_vendor_with_subscription.id
result = self.service.check_limit(db, vendor_id, "orders")
store_id = test_store_with_subscription.id
result = self.service.check_limit(db, store_id, "orders")
assert result.can_proceed is True
assert result.current == 10
assert result.limit == 100
def test_check_products_limit(self, db, test_vendor_with_products):
def test_check_products_limit(self, db, test_store_with_products):
"""Test checking products limit."""
vendor_id = test_vendor_with_products.id
result = self.service.check_limit(db, vendor_id, "products")
store_id = test_store_with_products.id
result = self.service.check_limit(db, store_id, "products")
assert result.can_proceed is True
assert result.current == 5
assert result.limit == 500
def test_check_team_members_limit(self, db, test_vendor_with_team):
def test_check_team_members_limit(self, db, test_store_with_team):
"""Test checking team members limit when at limit."""
vendor_id = test_vendor_with_team.id
result = self.service.check_limit(db, vendor_id, "team_members")
store_id = test_store_with_team.id
result = self.service.check_limit(db, store_id, "team_members")
# At limit (2/2) - can_proceed should be False
assert result.can_proceed is False
@@ -116,18 +116,18 @@ class TestUsageServiceCheckLimit:
assert result.limit == 2
assert result.percentage == 100.0
def test_check_unknown_limit_type(self, db, test_vendor_with_subscription):
def test_check_unknown_limit_type(self, db, test_store_with_subscription):
"""Test checking unknown limit type."""
vendor_id = test_vendor_with_subscription.id
result = self.service.check_limit(db, vendor_id, "unknown")
store_id = test_store_with_subscription.id
result = self.service.check_limit(db, store_id, "unknown")
assert result.can_proceed is True
assert "Unknown limit type" in result.message
def test_check_limit_upgrade_info_when_blocked(self, db, test_vendor_at_limit):
def test_check_limit_upgrade_info_when_blocked(self, db, test_store_at_limit):
"""Test upgrade info is provided when at limit."""
vendor_id = test_vendor_at_limit.id
result = self.service.check_limit(db, vendor_id, "orders")
store_id = test_store_at_limit.id
result = self.service.check_limit(db, store_id, "orders")
assert result.can_proceed is False
assert result.upgrade_tier_code == "professional"
@@ -182,13 +182,13 @@ def test_professional_tier(db, test_essential_tier):
@pytest.fixture
def test_vendor_with_subscription(db, test_vendor, test_essential_tier):
"""Create vendor with active subscription."""
def test_store_with_subscription(db, test_store, test_essential_tier):
"""Create store with active subscription."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = StoreSubscription(
store_id=test_store.id,
tier="essential",
tier_id=test_essential_tier.id,
status="active",
@@ -198,18 +198,18 @@ def test_vendor_with_subscription(db, test_vendor, test_essential_tier):
)
db.add(subscription)
db.commit()
db.refresh(test_vendor)
return test_vendor
db.refresh(test_store)
return test_store
@pytest.fixture
def test_vendor_at_limit(db, test_vendor, test_essential_tier, test_professional_tier):
"""Create vendor at order limit."""
def test_store_at_limit(db, test_store, test_essential_tier, test_professional_tier):
"""Create store at order limit."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = StoreSubscription(
store_id=test_store.id,
tier="essential",
tier_id=test_essential_tier.id,
status="active",
@@ -219,18 +219,18 @@ def test_vendor_at_limit(db, test_vendor, test_essential_tier, test_professional
)
db.add(subscription)
db.commit()
db.refresh(test_vendor)
return test_vendor
db.refresh(test_store)
return test_store
@pytest.fixture
def test_vendor_approaching_limit(db, test_vendor, test_essential_tier):
"""Create vendor approaching order limit (>=80%)."""
def test_store_approaching_limit(db, test_store, test_essential_tier):
"""Create store approaching order limit (>=80%)."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = StoreSubscription(
store_id=test_store.id,
tier="essential",
tier_id=test_essential_tier.id,
status="active",
@@ -240,18 +240,18 @@ def test_vendor_approaching_limit(db, test_vendor, test_essential_tier):
)
db.add(subscription)
db.commit()
db.refresh(test_vendor)
return test_vendor
db.refresh(test_store)
return test_store
@pytest.fixture
def test_vendor_on_professional(db, test_vendor, test_professional_tier):
"""Create vendor on highest tier."""
def test_store_on_professional(db, test_store, test_professional_tier):
"""Create store on highest tier."""
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
subscription = VendorSubscription(
vendor_id=test_vendor.id,
subscription = StoreSubscription(
store_id=test_store.id,
tier="professional",
tier_id=test_professional_tier.id,
status="active",
@@ -261,48 +261,48 @@ def test_vendor_on_professional(db, test_vendor, test_professional_tier):
)
db.add(subscription)
db.commit()
db.refresh(test_vendor)
return test_vendor
db.refresh(test_store)
return test_store
@pytest.fixture
def test_vendor_with_products(db, test_vendor_with_subscription, marketplace_product_factory):
"""Create vendor with products."""
def test_store_with_products(db, test_store_with_subscription, marketplace_product_factory):
"""Create store with products."""
for i in range(5):
# Create marketplace product first
mp = marketplace_product_factory(db, title=f"Test Product {i}")
product = Product(
vendor_id=test_vendor_with_subscription.id,
store_id=test_store_with_subscription.id,
marketplace_product_id=mp.id,
price_cents=1000,
is_active=True,
)
db.add(product)
db.commit()
return test_vendor_with_subscription
return test_store_with_subscription
@pytest.fixture
def test_vendor_with_team(db, test_vendor_with_subscription, test_user, other_user):
"""Create vendor with team members (owner + team member = 2)."""
from app.modules.tenancy.models import VendorUserType
def test_store_with_team(db, test_store_with_subscription, test_user, other_user):
"""Create store with team members (owner + team member = 2)."""
from app.modules.tenancy.models import StoreUserType
# Add owner
owner = VendorUser(
vendor_id=test_vendor_with_subscription.id,
owner = StoreUser(
store_id=test_store_with_subscription.id,
user_id=test_user.id,
user_type=VendorUserType.OWNER.value,
user_type=StoreUserType.OWNER.value,
is_active=True,
)
db.add(owner)
# Add team member
team_member = VendorUser(
vendor_id=test_vendor_with_subscription.id,
team_member = StoreUser(
store_id=test_store_with_subscription.id,
user_id=other_user.id,
user_type=VendorUserType.TEAM_MEMBER.value,
user_type=StoreUserType.TEAM_MEMBER.value,
is_active=True,
)
db.add(team_member)
db.commit()
return test_vendor_with_subscription
return test_store_with_subscription