Major changes: - Add AuditProvider protocol for cross-module audit logging - Move customer order operations to orders module (dependency inversion) - Add customer order metrics via MetricsProvider pattern - Fix missing db parameter in get_admin_context() calls - Move ProductMedia relationship to catalog module (proper ownership) - Add marketplace breakdown stats to marketplace_widgets New files: - contracts/audit.py - AuditProviderProtocol - core/services/audit_aggregator.py - Aggregates audit providers - monitoring/services/audit_provider.py - Monitoring audit implementation - orders/services/customer_order_service.py - Customer order operations - orders/routes/api/vendor_customer_orders.py - Customer order endpoints - catalog/services/product_media_service.py - Product media service - Architecture documentation for patterns Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
229 lines
6.6 KiB
Python
229 lines
6.6 KiB
Python
# tests/unit/services/test_customer_order_service.py
|
|
"""
|
|
Unit tests for CustomerOrderService.
|
|
|
|
Tests the orders module's customer-order relationship operations.
|
|
This service owns the customer-order relationship (customers module is agnostic).
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from app.modules.orders.models import Order
|
|
from app.modules.orders.services.customer_order_service import CustomerOrderService
|
|
|
|
|
|
@pytest.fixture
|
|
def customer_order_service():
|
|
"""Create CustomerOrderService instance."""
|
|
return CustomerOrderService()
|
|
|
|
|
|
@pytest.fixture
|
|
def customer_with_orders(db, test_vendor, test_customer):
|
|
"""Create a customer with multiple orders."""
|
|
orders = []
|
|
first_name = test_customer.first_name or "Test"
|
|
last_name = test_customer.last_name or "Customer"
|
|
|
|
for i in range(5):
|
|
order = Order(
|
|
vendor_id=test_vendor.id,
|
|
customer_id=test_customer.id,
|
|
order_number=f"ORD-{i:04d}",
|
|
status="pending" if i < 2 else "completed",
|
|
channel="direct",
|
|
order_date=datetime.now(UTC),
|
|
subtotal_cents=1000 * (i + 1),
|
|
total_amount_cents=1000 * (i + 1),
|
|
currency="EUR",
|
|
# Customer info
|
|
customer_email=test_customer.email,
|
|
customer_first_name=first_name,
|
|
customer_last_name=last_name,
|
|
# Shipping address
|
|
ship_first_name=first_name,
|
|
ship_last_name=last_name,
|
|
ship_address_line_1="123 Test St",
|
|
ship_city="Luxembourg",
|
|
ship_postal_code="L-1234",
|
|
ship_country_iso="LU",
|
|
# Billing address
|
|
bill_first_name=first_name,
|
|
bill_last_name=last_name,
|
|
bill_address_line_1="123 Test St",
|
|
bill_city="Luxembourg",
|
|
bill_postal_code="L-1234",
|
|
bill_country_iso="LU",
|
|
)
|
|
db.add(order)
|
|
orders.append(order)
|
|
|
|
db.commit()
|
|
for order in orders:
|
|
db.refresh(order)
|
|
|
|
return test_customer, orders
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCustomerOrderServiceGetOrders:
|
|
"""Tests for get_customer_orders method."""
|
|
|
|
def test_get_customer_orders_empty(
|
|
self, db, customer_order_service, test_vendor, test_customer
|
|
):
|
|
"""Test getting orders when customer has none."""
|
|
orders, total = customer_order_service.get_customer_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=test_customer.id,
|
|
)
|
|
|
|
assert orders == []
|
|
assert total == 0
|
|
|
|
def test_get_customer_orders_with_data(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test getting orders when customer has orders."""
|
|
customer, _ = customer_with_orders
|
|
|
|
orders, total = customer_order_service.get_customer_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
)
|
|
|
|
assert total == 5
|
|
assert len(orders) == 5
|
|
|
|
def test_get_customer_orders_pagination(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test pagination of customer orders."""
|
|
customer, _ = customer_with_orders
|
|
|
|
# Get first page
|
|
orders, total = customer_order_service.get_customer_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
skip=0,
|
|
limit=2,
|
|
)
|
|
|
|
assert total == 5
|
|
assert len(orders) == 2
|
|
|
|
# Get second page
|
|
orders, total = customer_order_service.get_customer_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
skip=2,
|
|
limit=2,
|
|
)
|
|
|
|
assert total == 5
|
|
assert len(orders) == 2
|
|
|
|
def test_get_customer_orders_ordered_by_date(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test that orders are returned in descending date order."""
|
|
customer, created_orders = customer_with_orders
|
|
|
|
orders, _ = customer_order_service.get_customer_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
)
|
|
|
|
# Most recent should be first
|
|
for i in range(len(orders) - 1):
|
|
assert orders[i].created_at >= orders[i + 1].created_at
|
|
|
|
def test_get_customer_orders_wrong_vendor(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test that orders from wrong vendor are not returned."""
|
|
customer, _ = customer_with_orders
|
|
|
|
# Use non-existent vendor ID
|
|
orders, total = customer_order_service.get_customer_orders(
|
|
db=db,
|
|
vendor_id=99999,
|
|
customer_id=customer.id,
|
|
)
|
|
|
|
assert orders == []
|
|
assert total == 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCustomerOrderServiceRecentOrders:
|
|
"""Tests for get_recent_orders method."""
|
|
|
|
def test_get_recent_orders(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test getting recent orders."""
|
|
customer, _ = customer_with_orders
|
|
|
|
orders = customer_order_service.get_recent_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
limit=3,
|
|
)
|
|
|
|
assert len(orders) == 3
|
|
|
|
def test_get_recent_orders_respects_limit(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test that limit is respected."""
|
|
customer, _ = customer_with_orders
|
|
|
|
orders = customer_order_service.get_recent_orders(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
limit=2,
|
|
)
|
|
|
|
assert len(orders) == 2
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCustomerOrderServiceOrderCount:
|
|
"""Tests for get_order_count method."""
|
|
|
|
def test_get_order_count_zero(
|
|
self, db, customer_order_service, test_vendor, test_customer
|
|
):
|
|
"""Test count when customer has no orders."""
|
|
count = customer_order_service.get_order_count(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=test_customer.id,
|
|
)
|
|
|
|
assert count == 0
|
|
|
|
def test_get_order_count_with_orders(
|
|
self, db, customer_order_service, test_vendor, customer_with_orders
|
|
):
|
|
"""Test count when customer has orders."""
|
|
customer, _ = customer_with_orders
|
|
|
|
count = customer_order_service.get_order_count(
|
|
db=db,
|
|
vendor_id=test_vendor.id,
|
|
customer_id=customer.id,
|
|
)
|
|
|
|
assert count == 5
|