test: add tests for merchant dashboard metrics and fix invoice template location

Move invoice PDF template from app/templates/invoices/ to
app/modules/orders/templates/invoices/ where InvoicePDFService expects it.
Expand invoice PDF tests to validate template path and existence.

Add unit tests for get_merchant_metrics() in tenancy, billing, and
customer metrics providers. Add unit tests for StatsAggregatorService
merchant methods. Add integration tests for the merchant dashboard
stats endpoint.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-22 21:46:34 +01:00
parent b77952bf89
commit d7a383f3d7
8 changed files with 786 additions and 1 deletions

View File

@@ -0,0 +1,149 @@
# app/modules/billing/tests/unit/test_billing_metrics.py
"""Unit tests for BillingMetricsProvider.get_merchant_metrics."""
import uuid
from datetime import UTC, datetime, timedelta
import pytest
from app.modules.billing.models import (
MerchantSubscription,
SubscriptionStatus,
SubscriptionTier,
)
from app.modules.billing.services.billing_metrics import BillingMetricsProvider
from app.modules.tenancy.models import Merchant, Platform, User
@pytest.fixture
def billing_platform(db):
"""Create a platform for billing metrics tests."""
platform = Platform(
code=f"bm_{uuid.uuid4().hex[:8]}",
name="Billing Metrics Platform",
is_active=True,
)
db.add(platform)
db.commit()
db.refresh(platform)
return platform
@pytest.fixture
def billing_merchant(db):
"""Create a merchant for billing metrics tests."""
from middleware.auth import AuthManager
auth = AuthManager()
user = User(
email=f"billowner_{uuid.uuid4().hex[:8]}@test.com",
username=f"billowner_{uuid.uuid4().hex[:8]}",
hashed_password=auth.hash_password("pass123"),
role="merchant_owner",
is_active=True,
)
db.add(user)
db.flush()
merchant = Merchant(
name="Billing Metrics Merchant",
owner_user_id=user.id,
contact_email=user.email,
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
return merchant
@pytest.fixture
def billing_tier(db, billing_platform):
"""Create a subscription tier."""
tier = SubscriptionTier(
code="professional",
name="Professional",
description="Pro tier",
price_monthly_cents=2900,
price_annual_cents=29000,
display_order=1,
is_active=True,
is_public=True,
platform_id=billing_platform.id,
)
db.add(tier)
db.commit()
db.refresh(tier)
return tier
@pytest.fixture
def billing_extra_platforms(db):
"""Create additional platforms for multiple subscriptions (unique constraint: merchant+platform)."""
platforms = []
for i in range(2):
p = Platform(
code=f"bm_extra_{uuid.uuid4().hex[:8]}",
name=f"Extra Platform {i}",
is_active=True,
)
db.add(p)
platforms.append(p)
db.commit()
for p in platforms:
db.refresh(p)
return platforms
@pytest.fixture
def billing_subscriptions(db, billing_merchant, billing_platform, billing_tier, billing_extra_platforms):
"""Create subscriptions: 1 active, 1 trial, 1 cancelled (each on a different platform)."""
platforms = [billing_platform, billing_extra_platforms[0], billing_extra_platforms[1]]
subs = []
for status, platform in zip(
[SubscriptionStatus.ACTIVE, SubscriptionStatus.TRIAL, SubscriptionStatus.CANCELLED],
platforms, strict=False,
):
sub = MerchantSubscription(
merchant_id=billing_merchant.id,
platform_id=platform.id,
tier_id=billing_tier.id,
status=status.value,
is_annual=False,
period_start=datetime.now(UTC),
period_end=datetime.now(UTC) + timedelta(days=30),
)
db.add(sub)
subs.append(sub)
db.commit()
for s in subs:
db.refresh(s)
return subs
@pytest.mark.unit
@pytest.mark.billing
class TestBillingMetricsProviderMerchant:
"""Tests for BillingMetricsProvider.get_merchant_metrics."""
def setup_method(self):
self.provider = BillingMetricsProvider()
def test_active_subscriptions_count(self, db, billing_merchant, billing_subscriptions):
"""Counts active + trial subscriptions, excludes cancelled."""
metrics = self.provider.get_merchant_metrics(db, billing_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["billing.active_subscriptions"] == 2
def test_no_subscriptions(self, db, billing_merchant):
"""Returns zero when merchant has no subscriptions."""
metrics = self.provider.get_merchant_metrics(db, billing_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["billing.active_subscriptions"] == 0
def test_nonexistent_merchant(self, db):
"""Returns zero for a non-existent merchant ID."""
metrics = self.provider.get_merchant_metrics(db, 999999)
by_key = {m.key: m.value for m in metrics}
assert by_key["billing.active_subscriptions"] == 0

View File

@@ -0,0 +1,241 @@
# app/modules/core/tests/integration/test_merchant_dashboard_routes.py
"""
Integration tests for merchant dashboard API routes.
Tests the merchant dashboard endpoint at:
GET /api/v1/merchants/core/dashboard/stats
"""
import uuid
from datetime import UTC, datetime, timedelta
import pytest
from app.api.deps import get_current_merchant_api, get_merchant_for_current_user
from app.modules.billing.models import (
MerchantSubscription,
SubscriptionStatus,
SubscriptionTier,
)
from app.modules.customers.models.customer import Customer
from app.modules.tenancy.models import Merchant, Platform, Store, StoreUser, User
from main import app
from models.schema.auth import UserContext
# ============================================================================
# Fixtures
# ============================================================================
BASE = "/api/v1/merchants/core"
@pytest.fixture
def dash_owner(db):
"""Create a merchant owner user."""
from middleware.auth import AuthManager
auth = AuthManager()
user = User(
email=f"dashowner_{uuid.uuid4().hex[:8]}@test.com",
username=f"dashowner_{uuid.uuid4().hex[:8]}",
hashed_password=auth.hash_password("pass123"),
role="merchant_owner",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@pytest.fixture
def dash_platform(db):
"""Create a platform."""
platform = Platform(
code=f"dp_{uuid.uuid4().hex[:8]}",
name="Dashboard Test Platform",
is_active=True,
)
db.add(platform)
db.commit()
db.refresh(platform)
return platform
@pytest.fixture
def dash_merchant(db, dash_owner):
"""Create a merchant."""
merchant = Merchant(
name="Dashboard Test Merchant",
owner_user_id=dash_owner.id,
contact_email=dash_owner.email,
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
return merchant
@pytest.fixture
def dash_stores(db, dash_merchant):
"""Create 3 stores (2 active, 1 inactive)."""
stores = []
for i, active in enumerate([True, True, False]):
uid = uuid.uuid4().hex[:8].upper()
store = Store(
merchant_id=dash_merchant.id,
store_code=f"DSTORE_{uid}",
subdomain=f"dstore{uid.lower()}",
name=f"Dashboard Store {i}",
is_active=active,
is_verified=True,
)
db.add(store)
stores.append(store)
db.commit()
for s in stores:
db.refresh(s)
return stores
@pytest.fixture
def dash_team_members(db, dash_stores, dash_owner):
"""Create team members across stores."""
from middleware.auth import AuthManager
auth = AuthManager()
users = []
for _ in range(2):
u = User(
email=f"dteam_{uuid.uuid4().hex[:8]}@test.com",
username=f"dteam_{uuid.uuid4().hex[:8]}",
hashed_password=auth.hash_password("pass123"),
role="store_user",
is_active=True,
)
db.add(u)
users.append(u)
db.flush()
db.add(StoreUser(store_id=dash_stores[0].id, user_id=users[0].id, is_active=True))
db.add(StoreUser(store_id=dash_stores[1].id, user_id=users[1].id, is_active=True))
db.commit()
return users
@pytest.fixture
def dash_customers(db, dash_stores):
"""Create customers in the merchant's stores."""
for i in range(4):
uid = uuid.uuid4().hex[:8]
db.add(
Customer(
store_id=dash_stores[0].id,
email=f"dc_{uid}@test.com",
hashed_password="hashed", # noqa: SEC001
first_name=f"F{i}",
last_name=f"L{i}",
customer_number=f"DC{uid}",
is_active=True,
)
)
db.commit()
@pytest.fixture
def dash_subscription(db, dash_merchant, dash_platform):
"""Create an active subscription."""
tier = SubscriptionTier(
code=f"pro_{uuid.uuid4().hex[:8]}",
name="Professional",
description="Pro",
price_monthly_cents=2900,
price_annual_cents=29000,
display_order=1,
is_active=True,
is_public=True,
platform_id=dash_platform.id,
)
db.add(tier)
db.flush()
sub = MerchantSubscription(
merchant_id=dash_merchant.id,
platform_id=dash_platform.id,
tier_id=tier.id,
status=SubscriptionStatus.ACTIVE.value,
is_annual=False,
period_start=datetime.now(UTC),
period_end=datetime.now(UTC) + timedelta(days=30),
)
db.add(sub)
db.commit()
db.refresh(sub)
return sub
@pytest.fixture
def dash_auth(dash_owner, dash_merchant):
"""Override auth dependencies for dashboard merchant."""
user_context = UserContext(
id=dash_owner.id,
email=dash_owner.email,
username=dash_owner.username,
role="merchant_owner",
is_active=True,
)
def _override_merchant():
return dash_merchant
def _override_user():
return user_context
app.dependency_overrides[get_merchant_for_current_user] = _override_merchant
app.dependency_overrides[get_current_merchant_api] = _override_user
yield {"Authorization": "Bearer fake-token"}
app.dependency_overrides.pop(get_merchant_for_current_user, None)
app.dependency_overrides.pop(get_current_merchant_api, None)
# ============================================================================
# Tests
# ============================================================================
@pytest.mark.integration
@pytest.mark.core
class TestMerchantDashboardStats:
"""Tests for GET /api/v1/merchants/core/dashboard/stats."""
def test_returns_correct_stats(
self, client, db, dash_auth, dash_stores, dash_team_members, dash_customers, dash_subscription
):
"""Endpoint returns correct aggregated stats."""
response = client.get(f"{BASE}/dashboard/stats", headers=dash_auth)
assert response.status_code == 200
data = response.json()
assert data["total_stores"] == 3
assert data["active_subscriptions"] == 1
assert data["total_customers"] == 4
assert data["team_members"] == 2
def test_returns_zeros_when_empty(self, client, db, dash_auth):
"""Returns zero counts when merchant has no data."""
response = client.get(f"{BASE}/dashboard/stats", headers=dash_auth)
assert response.status_code == 200
data = response.json()
assert data["total_stores"] == 0
assert data["active_subscriptions"] == 0
assert data["total_customers"] == 0
assert data["team_members"] == 0
def test_requires_auth(self, client):
"""Returns 401 without auth."""
# Remove any overrides
app.dependency_overrides.pop(get_merchant_for_current_user, None)
app.dependency_overrides.pop(get_current_merchant_api, None)
response = client.get(f"{BASE}/dashboard/stats")
assert response.status_code == 401

View File

@@ -0,0 +1,106 @@
# app/modules/core/tests/unit/test_stats_aggregator.py
"""Unit tests for StatsAggregatorService merchant methods."""
from unittest.mock import MagicMock, patch
import pytest
from app.modules.contracts.metrics import MetricValue
from app.modules.core.services.stats_aggregator import StatsAggregatorService
def _make_metric(key: str, value: int, category: str = "test") -> MetricValue:
return MetricValue(key=key, value=value, label=key, category=category)
@pytest.mark.unit
@pytest.mark.core
class TestStatsAggregatorMerchant:
"""Tests for StatsAggregatorService merchant aggregation methods."""
def setup_method(self):
self.service = StatsAggregatorService()
def test_get_merchant_dashboard_stats_aggregates_providers(self, db):
"""Aggregates metrics from providers that implement get_merchant_metrics."""
provider_a = MagicMock()
provider_a.metrics_category = "billing"
provider_a.get_merchant_metrics.return_value = [
_make_metric("billing.active_subscriptions", 2, "billing"),
]
provider_b = MagicMock()
provider_b.metrics_category = "tenancy"
provider_b.get_merchant_metrics.return_value = [
_make_metric("tenancy.total_stores", 3, "tenancy"),
]
module_a = MagicMock()
module_a.code = "billing"
module_b = MagicMock()
module_b.code = "tenancy"
with patch.object(
self.service, "_get_enabled_providers", return_value=[(module_a, provider_a), (module_b, provider_b)]
):
result = self.service.get_merchant_dashboard_stats(db, merchant_id=1, platform_id=1)
assert "billing" in result
assert "tenancy" in result
assert result["billing"][0].value == 2
assert result["tenancy"][0].value == 3
def test_get_merchant_dashboard_stats_skips_providers_without_method(self, db):
"""Skips providers that don't have get_merchant_metrics."""
provider = MagicMock(spec=[]) # No attributes at all
provider.metrics_category = "legacy"
module = MagicMock()
module.code = "legacy"
with patch.object(
self.service, "_get_enabled_providers", return_value=[(module, provider)]
):
result = self.service.get_merchant_dashboard_stats(db, merchant_id=1, platform_id=1)
assert result == {}
def test_get_merchant_dashboard_stats_handles_provider_error(self, db):
"""Gracefully handles a provider raising an exception."""
provider = MagicMock()
provider.metrics_category = "broken"
provider.get_merchant_metrics.side_effect = RuntimeError("DB error")
module = MagicMock()
module.code = "broken"
with patch.object(
self.service, "_get_enabled_providers", return_value=[(module, provider)]
):
result = self.service.get_merchant_dashboard_stats(db, merchant_id=1, platform_id=1)
assert result == {}
def test_get_merchant_stats_flat(self, db):
"""Flattens categorized metrics into a single dict."""
provider = MagicMock()
provider.metrics_category = "tenancy"
provider.get_merchant_metrics.return_value = [
_make_metric("tenancy.total_stores", 5, "tenancy"),
_make_metric("tenancy.team_members", 12, "tenancy"),
]
module = MagicMock()
module.code = "tenancy"
with patch.object(
self.service, "_get_enabled_providers", return_value=[(module, provider)]
):
flat = self.service.get_merchant_stats_flat(db, merchant_id=1, platform_id=1)
assert flat["tenancy.total_stores"] == 5
assert flat["tenancy.team_members"] == 12
def test_get_merchant_stats_flat_empty(self, db):
"""Returns empty dict when no providers have merchant metrics."""
with patch.object(self.service, "_get_enabled_providers", return_value=[]):
flat = self.service.get_merchant_stats_flat(db, merchant_id=1, platform_id=1)
assert flat == {}

View File

@@ -0,0 +1,130 @@
# app/modules/customers/tests/unit/test_customer_metrics.py
"""Unit tests for CustomerMetricsProvider.get_merchant_metrics."""
import uuid
import pytest
from app.modules.customers.models.customer import Customer
from app.modules.customers.services.customer_metrics import CustomerMetricsProvider
from app.modules.tenancy.models import Merchant, Store, User
@pytest.fixture
def cust_merchant(db):
"""Create a merchant for customer metrics tests."""
from middleware.auth import AuthManager
auth = AuthManager()
user = User(
email=f"custowner_{uuid.uuid4().hex[:8]}@test.com",
username=f"custowner_{uuid.uuid4().hex[:8]}",
hashed_password=auth.hash_password("pass123"),
role="merchant_owner",
is_active=True,
)
db.add(user)
db.flush()
merchant = Merchant(
name="Customer Metrics Merchant",
owner_user_id=user.id,
contact_email=user.email,
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
return merchant
@pytest.fixture
def cust_stores(db, cust_merchant):
"""Create 2 stores for the merchant."""
stores = []
for i in range(2):
uid = uuid.uuid4().hex[:8].upper()
store = Store(
merchant_id=cust_merchant.id,
store_code=f"CSTORE_{uid}",
subdomain=f"cstore{uid.lower()}",
name=f"Cust Store {i}",
is_active=True,
is_verified=True,
)
db.add(store)
stores.append(store)
db.commit()
for s in stores:
db.refresh(s)
return stores
@pytest.fixture
def cust_customers(db, cust_stores):
"""Create customers across the merchant's stores."""
customers = []
# 3 customers in store 0
for i in range(3):
uid = uuid.uuid4().hex[:8]
c = Customer(
store_id=cust_stores[0].id,
email=f"cust_{uid}@test.com",
hashed_password="hashed", # noqa: SEC001
first_name=f"First{i}",
last_name=f"Last{i}",
customer_number=f"C{uid}",
is_active=True,
)
db.add(c)
customers.append(c)
# 2 customers in store 1
for i in range(2):
uid = uuid.uuid4().hex[:8]
c = Customer(
store_id=cust_stores[1].id,
email=f"cust_{uid}@test.com",
hashed_password="hashed", # noqa: SEC001
first_name=f"First{i}",
last_name=f"Last{i}",
customer_number=f"C{uid}",
is_active=True,
)
db.add(c)
customers.append(c)
db.commit()
return customers
@pytest.mark.unit
@pytest.mark.customers
class TestCustomerMetricsProviderMerchant:
"""Tests for CustomerMetricsProvider.get_merchant_metrics."""
def setup_method(self):
self.provider = CustomerMetricsProvider()
def test_total_customers_across_stores(self, db, cust_merchant, cust_stores, cust_customers):
"""Aggregates customers across all merchant stores."""
metrics = self.provider.get_merchant_metrics(db, cust_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["customers.total"] == 5
def test_no_customers(self, db, cust_merchant, cust_stores):
"""Returns zero when stores have no customers."""
metrics = self.provider.get_merchant_metrics(db, cust_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["customers.total"] == 0
def test_no_stores(self, db, cust_merchant):
"""Returns zero when merchant has no stores."""
metrics = self.provider.get_merchant_metrics(db, cust_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["customers.total"] == 0
def test_nonexistent_merchant(self, db):
"""Returns zero for a non-existent merchant ID."""
metrics = self.provider.get_merchant_metrics(db, 999999)
by_key = {m.key: m.value for m in metrics}
assert by_key["customers.total"] == 0

View File

@@ -1,8 +1,13 @@
"""Unit tests for InvoicePDFService."""
from pathlib import Path
import pytest
from app.modules.orders.services.invoice_pdf_service import InvoicePDFService
from app.modules.orders.services.invoice_pdf_service import (
TEMPLATE_DIR,
InvoicePDFService,
)
@pytest.mark.unit
@@ -16,3 +21,25 @@ class TestInvoicePDFService:
def test_service_instantiation(self):
"""Service can be instantiated."""
assert self.service is not None
def test_template_directory_exists(self):
"""Template directory must exist at the expected path."""
assert TEMPLATE_DIR.exists(), f"Template directory missing: {TEMPLATE_DIR}"
assert TEMPLATE_DIR.is_dir()
def test_invoice_template_exists(self):
"""invoice.html template must exist in the template directory."""
template_path = TEMPLATE_DIR / "invoice.html"
assert template_path.exists(), f"Invoice template missing: {template_path}"
def test_template_can_be_loaded(self):
"""Jinja2 environment can load the invoice template."""
template = self.service.env.get_template("invoice.html")
assert template is not None
def test_template_dir_is_inside_orders_module(self):
"""Template directory should be inside the orders module, not the global templates."""
orders_module_dir = Path(__file__).parent.parent.parent
assert str(TEMPLATE_DIR).startswith(str(orders_module_dir)), (
f"TEMPLATE_DIR ({TEMPLATE_DIR}) should be inside orders module ({orders_module_dir})"
)

View File

@@ -0,0 +1,132 @@
# app/modules/tenancy/tests/unit/test_tenancy_metrics.py
"""Unit tests for TenancyMetricsProvider.get_merchant_metrics."""
import uuid
import pytest
from app.modules.tenancy.models import Merchant, Store, StoreUser, User
from app.modules.tenancy.services.tenancy_metrics import TenancyMetricsProvider
@pytest.fixture
def metrics_merchant(db):
"""Create a merchant owner and merchant for metrics tests."""
from middleware.auth import AuthManager
auth = AuthManager()
user = User(
email=f"metricsowner_{uuid.uuid4().hex[:8]}@test.com",
username=f"metricsowner_{uuid.uuid4().hex[:8]}",
hashed_password=auth.hash_password("pass123"),
role="merchant_owner",
is_active=True,
)
db.add(user)
db.flush()
merchant = Merchant(
name="Metrics Test Merchant",
owner_user_id=user.id,
contact_email=user.email,
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
return merchant
@pytest.fixture
def metrics_stores(db, metrics_merchant):
"""Create 3 stores (2 active, 1 inactive) for the merchant."""
stores = []
for i, active in enumerate([True, True, False]):
uid = uuid.uuid4().hex[:8].upper()
store = Store(
merchant_id=metrics_merchant.id,
store_code=f"MSTORE_{uid}",
subdomain=f"mstore{uid.lower()}",
name=f"Metrics Store {i}",
is_active=active,
is_verified=True,
)
db.add(store)
stores.append(store)
db.commit()
for s in stores:
db.refresh(s)
return stores
@pytest.fixture
def metrics_team_members(db, metrics_stores):
"""Create team members across merchant stores."""
from middleware.auth import AuthManager
auth = AuthManager()
users = []
for i in range(3):
u = User(
email=f"team_{uuid.uuid4().hex[:8]}@test.com",
username=f"team_{uuid.uuid4().hex[:8]}",
hashed_password=auth.hash_password("pass123"),
role="store_user",
is_active=True,
)
db.add(u)
users.append(u)
db.flush()
# User 0 on store 0 and store 1 (should be counted once)
db.add(StoreUser(store_id=metrics_stores[0].id, user_id=users[0].id, is_active=True))
db.add(StoreUser(store_id=metrics_stores[1].id, user_id=users[0].id, is_active=True))
# User 1 on store 0 only
db.add(StoreUser(store_id=metrics_stores[0].id, user_id=users[1].id, is_active=True))
# User 2 on store 0 but inactive
db.add(StoreUser(store_id=metrics_stores[0].id, user_id=users[2].id, is_active=False))
db.commit()
return users
@pytest.mark.unit
@pytest.mark.tenancy
class TestTenancyMetricsProviderMerchant:
"""Tests for TenancyMetricsProvider.get_merchant_metrics."""
def setup_method(self):
self.provider = TenancyMetricsProvider()
def test_total_stores(self, db, metrics_merchant, metrics_stores):
"""Returns correct total store count for merchant."""
metrics = self.provider.get_merchant_metrics(db, metrics_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["tenancy.total_stores"] == 3
def test_active_stores(self, db, metrics_merchant, metrics_stores):
"""Returns correct active store count (excludes inactive)."""
metrics = self.provider.get_merchant_metrics(db, metrics_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["tenancy.active_stores"] == 2
def test_team_members_distinct(self, db, metrics_merchant, metrics_stores, metrics_team_members):
"""Counts distinct active team members across stores."""
metrics = self.provider.get_merchant_metrics(db, metrics_merchant.id)
by_key = {m.key: m.value for m in metrics}
# 2 active distinct users (user 0 on 2 stores counted once, user 1, user 2 inactive)
assert by_key["tenancy.team_members"] == 2
def test_no_stores(self, db, metrics_merchant):
"""Returns zero counts when merchant has no stores."""
metrics = self.provider.get_merchant_metrics(db, metrics_merchant.id)
by_key = {m.key: m.value for m in metrics}
assert by_key["tenancy.total_stores"] == 0
assert by_key["tenancy.active_stores"] == 0
assert by_key["tenancy.team_members"] == 0
def test_nonexistent_merchant(self, db):
"""Returns zero counts for a non-existent merchant ID."""
metrics = self.provider.get_merchant_metrics(db, 999999)
by_key = {m.key: m.value for m in metrics}
assert by_key["tenancy.total_stores"] == 0