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

feat: add customer multiple addresses management

Browse Source 
- Add CustomerAddressService with CRUD operations
- Add shop API endpoints for address management (GET, POST, PUT, DELETE)
- Add set default endpoint for address type
- Implement addresses.html with full UI (cards, modals, Alpine.js)
- Integrate saved addresses in checkout flow
  - Address selector dropdowns for shipping/billing
  - Auto-select default addresses
  - Save new address checkbox option
- Add country_iso field alongside country_name
- Add address exceptions (NotFound, LimitExceeded, InvalidType)
- Max 10 addresses per customer limit
- One default address per type (shipping/billing)
- Add unit tests for CustomerAddressService
- Add integration tests for shop addresses API

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-02 19:16:35 +01:00
parent ea0218746f
commit b5b32fb351
15 changed files with 3940 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
tests/fixtures/customer_fixtures.py vendored
View File

@@ -32,7 +32,7 @@ def test_customer(db, test_vendor):
@pytest.fixture
def test_customer_address(db, test_vendor, test_customer):
"""Create a test customer address."""
"""Create a test customer shipping address."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
@@ -42,7 +42,8 @@ def test_customer_address(db, test_vendor, test_customer):
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
country_name="Luxembourg",
country_iso="LU",
is_default=True,
)
db.add(address)
@@ -51,6 +52,55 @@ def test_customer_address(db, test_vendor, test_customer):
return address
@pytest.fixture
def test_customer_billing_address(db, test_vendor, test_customer):
"""Create a test customer billing address."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="billing",
first_name="John",
last_name="Doe",
company="Test Company S.A.",
address_line_1="456 Business Ave",
city="Luxembourg",
postal_code="L-5678",
country_name="Luxembourg",
country_iso="LU",
is_default=True,
)
db.add(address)
db.commit()
db.refresh(address)
return address
@pytest.fixture
def test_customer_multiple_addresses(db, test_vendor, test_customer):
"""Create multiple addresses for testing limits and listing."""
addresses = []
for i in range(3):
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping" if i % 2 == 0 else "billing",
first_name=f"Name{i}",
last_name="Test",
address_line_1=f"{i}00 Test Street",
city="Luxembourg",
postal_code=f"L-{1000+i}",
country_name="Luxembourg",
country_iso="LU",
is_default=(i == 0),
)
db.add(address)
addresses.append(address)
db.commit()
for addr in addresses:
db.refresh(addr)
return addresses
@pytest.fixture
def test_order(db, test_vendor, test_customer, test_customer_address):
"""Create a test order with customer/address snapshots."""

1
tests/integration/api/v1/shop/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Shop API integration tests

621
tests/integration/api/v1/shop/test_addresses.py Normal file
View File

@@ -0,0 +1,621 @@
# tests/integration/api/v1/shop/test_addresses.py
"""Integration tests for shop addresses API endpoints.
Tests the /api/v1/shop/addresses/* endpoints.
All endpoints require customer JWT authentication with vendor context.
"""
from datetime import UTC, datetime, timedelta
from unittest.mock import patch
import pytest
from jose import jwt
from models.database.customer import Customer, CustomerAddress
@pytest.fixture
def shop_customer(db, test_vendor):
"""Create a test customer for shop API tests."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
customer = Customer(
vendor_id=test_vendor.id,
email="shopcustomer@example.com",
hashed_password=auth_manager.hash_password("testpass123"),
first_name="Shop",
last_name="Customer",
customer_number="SHOP001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def shop_customer_token(shop_customer, test_vendor):
"""Create JWT token for shop customer."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
expires_delta = timedelta(minutes=auth_manager.token_expire_minutes)
expire = datetime.now(UTC) + expires_delta
payload = {
"sub": str(shop_customer.id),
"email": shop_customer.email,
"vendor_id": test_vendor.id,
"type": "customer",
"exp": expire,
"iat": datetime.now(UTC),
}
token = jwt.encode(
payload, auth_manager.secret_key, algorithm=auth_manager.algorithm
)
return token
@pytest.fixture
def shop_customer_headers(shop_customer_token):
"""Get authentication headers for shop customer."""
return {"Authorization": f"Bearer {shop_customer_token}"}
@pytest.fixture
def customer_address(db, test_vendor, shop_customer):
"""Create a test address for shop customer."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=shop_customer.id,
address_type="shipping",
first_name="Ship",
last_name="Address",
address_line_1="123 Shipping St",
city="Luxembourg",
postal_code="L-1234",
country_name="Luxembourg",
country_iso="LU",
is_default=True,
)
db.add(address)
db.commit()
db.refresh(address)
return address
@pytest.fixture
def customer_billing_address(db, test_vendor, shop_customer):
"""Create a billing address for shop customer."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=shop_customer.id,
address_type="billing",
first_name="Bill",
last_name="Address",
company="Test Company",
address_line_1="456 Billing Ave",
city="Esch-sur-Alzette",
postal_code="L-5678",
country_name="Luxembourg",
country_iso="LU",
is_default=True,
)
db.add(address)
db.commit()
db.refresh(address)
return address
@pytest.fixture
def other_customer(db, test_vendor):
"""Create another customer for testing access controls."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
customer = Customer(
vendor_id=test_vendor.id,
email="othercustomer@example.com",
hashed_password=auth_manager.hash_password("otherpass123"),
first_name="Other",
last_name="Customer",
customer_number="OTHER001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def other_customer_address(db, test_vendor, other_customer):
"""Create an address for another customer."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=other_customer.id,
address_type="shipping",
first_name="Other",
last_name="Address",
address_line_1="999 Other St",
city="Differdange",
postal_code="L-9999",
country_name="Luxembourg",
country_iso="LU",
is_default=True,
)
db.add(address)
db.commit()
db.refresh(address)
return address
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopAddressesListAPI:
"""Test shop addresses list endpoint at /api/v1/shop/addresses."""
def test_list_addresses_requires_authentication(self, client, test_vendor):
"""Test that listing addresses requires authentication."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.get("/api/v1/shop/addresses")
assert response.status_code in [401, 403]
def test_list_addresses_success(
self,
client,
shop_customer_headers,
customer_address,
test_vendor,
shop_customer,
):
"""Test listing customer addresses successfully."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/addresses",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert "addresses" in data
assert "total" in data
assert data["total"] == 1
assert data["addresses"][0]["first_name"] == "Ship"
def test_list_addresses_empty(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test listing addresses when customer has none."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/addresses",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["addresses"] == []
def test_list_addresses_multiple_types(
self,
client,
shop_customer_headers,
customer_address,
customer_billing_address,
test_vendor,
shop_customer,
):
"""Test listing addresses includes both shipping and billing."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/addresses",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 2
types = {addr["address_type"] for addr in data["addresses"]}
assert "shipping" in types
assert "billing" in types
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopAddressDetailAPI:
"""Test shop address detail endpoint at /api/v1/shop/addresses/{address_id}."""
def test_get_address_success(
self,
client,
shop_customer_headers,
customer_address,
test_vendor,
shop_customer,
):
"""Test getting address details successfully."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/addresses/{customer_address.id}",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert data["id"] == customer_address.id
assert data["first_name"] == "Ship"
assert data["country_iso"] == "LU"
assert data["country_name"] == "Luxembourg"
def test_get_address_not_found(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test getting non-existent address returns 404."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/addresses/99999",
headers=shop_customer_headers,
)
assert response.status_code == 404
def test_get_address_other_customer(
self,
client,
shop_customer_headers,
other_customer_address,
test_vendor,
shop_customer,
):
"""Test cannot access another customer's address."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/addresses/{other_customer_address.id}",
headers=shop_customer_headers,
)
# Should return 404 to prevent enumeration
assert response.status_code == 404
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopAddressCreateAPI:
"""Test shop address creation at POST /api/v1/shop/addresses."""
def test_create_address_success(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test creating a new address."""
address_data = {
"address_type": "shipping",
"first_name": "New",
"last_name": "Address",
"address_line_1": "789 New St",
"city": "Luxembourg",
"postal_code": "L-1111",
"country_name": "Luxembourg",
"country_iso": "LU",
"is_default": False,
}
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.post(
"/api/v1/shop/addresses",
headers=shop_customer_headers,
json=address_data,
)
assert response.status_code == 201
data = response.json()
assert data["first_name"] == "New"
assert data["last_name"] == "Address"
assert data["country_iso"] == "LU"
assert "id" in data
def test_create_address_with_company(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test creating address with company name."""
address_data = {
"address_type": "billing",
"first_name": "Business",
"last_name": "Address",
"company": "Acme Corp",
"address_line_1": "100 Business Park",
"city": "Luxembourg",
"postal_code": "L-2222",
"country_name": "Luxembourg",
"country_iso": "LU",
"is_default": True,
}
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.post(
"/api/v1/shop/addresses",
headers=shop_customer_headers,
json=address_data,
)
assert response.status_code == 201
data = response.json()
assert data["company"] == "Acme Corp"
def test_create_address_validation_error(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test validation error for missing required fields."""
address_data = {
"address_type": "shipping",
"first_name": "Test",
# Missing required fields
}
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.post(
"/api/v1/shop/addresses",
headers=shop_customer_headers,
json=address_data,
)
assert response.status_code == 422
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopAddressUpdateAPI:
"""Test shop address update at PUT /api/v1/shop/addresses/{address_id}."""
def test_update_address_success(
self,
client,
shop_customer_headers,
customer_address,
test_vendor,
shop_customer,
):
"""Test updating an address."""
update_data = {
"first_name": "Updated",
"city": "Esch-sur-Alzette",
}
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.put(
f"/api/v1/shop/addresses/{customer_address.id}",
headers=shop_customer_headers,
json=update_data,
)
assert response.status_code == 200
data = response.json()
assert data["first_name"] == "Updated"
assert data["city"] == "Esch-sur-Alzette"
def test_update_address_not_found(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test updating non-existent address returns 404."""
update_data = {"first_name": "Test"}
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.put(
"/api/v1/shop/addresses/99999",
headers=shop_customer_headers,
json=update_data,
)
assert response.status_code == 404
def test_update_address_other_customer(
self,
client,
shop_customer_headers,
other_customer_address,
test_vendor,
shop_customer,
):
"""Test cannot update another customer's address."""
update_data = {"first_name": "Hacked"}
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.put(
f"/api/v1/shop/addresses/{other_customer_address.id}",
headers=shop_customer_headers,
json=update_data,
)
assert response.status_code == 404
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopAddressDeleteAPI:
"""Test shop address deletion at DELETE /api/v1/shop/addresses/{address_id}."""
def test_delete_address_success(
self,
client,
shop_customer_headers,
customer_address,
test_vendor,
shop_customer,
):
"""Test deleting an address."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.delete(
f"/api/v1/shop/addresses/{customer_address.id}",
headers=shop_customer_headers,
)
assert response.status_code == 204
def test_delete_address_not_found(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test deleting non-existent address returns 404."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.delete(
"/api/v1/shop/addresses/99999",
headers=shop_customer_headers,
)
assert response.status_code == 404
def test_delete_address_other_customer(
self,
client,
shop_customer_headers,
other_customer_address,
test_vendor,
shop_customer,
):
"""Test cannot delete another customer's address."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.delete(
f"/api/v1/shop/addresses/{other_customer_address.id}",
headers=shop_customer_headers,
)
assert response.status_code == 404
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopAddressSetDefaultAPI:
"""Test set address as default at PUT /api/v1/shop/addresses/{address_id}/default."""
def test_set_default_success(
self,
client,
shop_customer_headers,
customer_address,
test_vendor,
shop_customer,
db,
):
"""Test setting address as default."""
# Create a second non-default address
second_address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=shop_customer.id,
address_type="shipping",
first_name="Second",
last_name="Address",
address_line_1="222 Second St",
city="Dudelange",
postal_code="L-3333",
country_name="Luxembourg",
country_iso="LU",
is_default=False,
)
db.add(second_address)
db.commit()
db.refresh(second_address)
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.put(
f"/api/v1/shop/addresses/{second_address.id}/default",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert data["is_default"] is True
def test_set_default_not_found(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test setting default on non-existent address returns 404."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.put(
"/api/v1/shop/addresses/99999/default",
headers=shop_customer_headers,
)
assert response.status_code == 404
def test_set_default_other_customer(
self,
client,
shop_customer_headers,
other_customer_address,
test_vendor,
shop_customer,
):
"""Test cannot set default on another customer's address."""
with patch("app.api.v1.shop.addresses.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.put(
f"/api/v1/shop/addresses/{other_customer_address.id}/default",
headers=shop_customer_headers,
)
assert response.status_code == 404

557
tests/integration/api/v1/shop/test_orders.py Normal file
View File

@@ -0,0 +1,557 @@
# tests/integration/api/v1/shop/test_orders.py
"""Integration tests for shop orders API endpoints.
Tests the /api/v1/shop/orders/* endpoints.
All endpoints require customer JWT authentication with vendor context.
"""
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from unittest.mock import patch, MagicMock
import pytest
from jose import jwt
from models.database.customer import Customer
from models.database.invoice import Invoice, InvoiceStatus, VendorInvoiceSettings
from models.database.order import Order, OrderItem
@pytest.fixture
def shop_customer(db, test_vendor):
"""Create a test customer for shop API tests."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
customer = Customer(
vendor_id=test_vendor.id,
email="shopcustomer@example.com",
hashed_password=auth_manager.hash_password("testpass123"),
first_name="Shop",
last_name="Customer",
customer_number="SHOP001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def shop_customer_token(shop_customer, test_vendor):
"""Create JWT token for shop customer."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
expires_delta = timedelta(minutes=auth_manager.token_expire_minutes)
expire = datetime.now(UTC) + expires_delta
payload = {
"sub": str(shop_customer.id),
"email": shop_customer.email,
"vendor_id": test_vendor.id,
"type": "customer",
"exp": expire,
"iat": datetime.now(UTC),
}
token = jwt.encode(
payload, auth_manager.secret_key, algorithm=auth_manager.algorithm
)
return token
@pytest.fixture
def shop_customer_headers(shop_customer_token):
"""Get authentication headers for shop customer."""
return {"Authorization": f"Bearer {shop_customer_token}"}
@pytest.fixture
def shop_order(db, test_vendor, shop_customer):
"""Create a test order for shop customer."""
order = Order(
vendor_id=test_vendor.id,
customer_id=shop_customer.id,
order_number="SHOP-ORD-001",
status="pending",
channel="direct",
order_date=datetime.now(UTC),
subtotal_cents=10000,
tax_amount_cents=1700,
shipping_amount_cents=500,
total_amount_cents=12200,
currency="EUR",
customer_email=shop_customer.email,
customer_first_name=shop_customer.first_name,
customer_last_name=shop_customer.last_name,
ship_first_name=shop_customer.first_name,
ship_last_name=shop_customer.last_name,
ship_address_line_1="123 Shop St",
ship_city="Luxembourg",
ship_postal_code="L-1234",
ship_country_iso="LU",
bill_first_name=shop_customer.first_name,
bill_last_name=shop_customer.last_name,
bill_address_line_1="123 Shop St",
bill_city="Luxembourg",
bill_postal_code="L-1234",
bill_country_iso="LU",
# VAT fields
vat_regime="domestic",
vat_rate=Decimal("17.00"),
vat_rate_label="Luxembourg VAT 17.00%",
vat_destination_country=None,
)
db.add(order)
db.commit()
db.refresh(order)
return order
@pytest.fixture
def shop_order_processing(db, test_vendor, shop_customer):
"""Create a test order with processing status (eligible for invoice)."""
order = Order(
vendor_id=test_vendor.id,
customer_id=shop_customer.id,
order_number="SHOP-ORD-002",
status="processing",
channel="direct",
order_date=datetime.now(UTC),
subtotal_cents=20000,
tax_amount_cents=3400,
shipping_amount_cents=500,
total_amount_cents=23900,
currency="EUR",
customer_email=shop_customer.email,
customer_first_name=shop_customer.first_name,
customer_last_name=shop_customer.last_name,
ship_first_name=shop_customer.first_name,
ship_last_name=shop_customer.last_name,
ship_address_line_1="456 Shop Ave",
ship_city="Luxembourg",
ship_postal_code="L-5678",
ship_country_iso="LU",
bill_first_name=shop_customer.first_name,
bill_last_name=shop_customer.last_name,
bill_address_line_1="456 Shop Ave",
bill_city="Luxembourg",
bill_postal_code="L-5678",
bill_country_iso="LU",
# VAT fields
vat_regime="domestic",
vat_rate=Decimal("17.00"),
vat_rate_label="Luxembourg VAT 17.00%",
)
db.add(order)
db.flush()
# Add order item
item = OrderItem(
order_id=order.id,
product_id=1,
product_sku="TEST-SKU-001",
product_name="Test Product",
quantity=2,
unit_price_cents=10000,
total_price_cents=20000,
)
db.add(item)
db.commit()
db.refresh(order)
return order
@pytest.fixture
def shop_invoice_settings(db, test_vendor):
"""Create invoice settings for the vendor."""
settings = VendorInvoiceSettings(
vendor_id=test_vendor.id,
company_name="Shop Test Company S.A.",
company_address="123 Business St",
company_city="Luxembourg",
company_postal_code="L-1234",
company_country="LU",
vat_number="LU12345678",
invoice_prefix="INV",
invoice_next_number=1,
default_vat_rate=Decimal("17.00"),
)
db.add(settings)
db.commit()
db.refresh(settings)
return settings
@pytest.fixture
def shop_order_with_invoice(db, test_vendor, shop_customer, shop_invoice_settings):
"""Create an order with an existing invoice."""
order = Order(
vendor_id=test_vendor.id,
customer_id=shop_customer.id,
order_number="SHOP-ORD-003",
status="shipped",
channel="direct",
order_date=datetime.now(UTC),
subtotal_cents=15000,
tax_amount_cents=2550,
shipping_amount_cents=500,
total_amount_cents=18050,
currency="EUR",
customer_email=shop_customer.email,
customer_first_name=shop_customer.first_name,
customer_last_name=shop_customer.last_name,
ship_first_name=shop_customer.first_name,
ship_last_name=shop_customer.last_name,
ship_address_line_1="789 Shop Blvd",
ship_city="Luxembourg",
ship_postal_code="L-9999",
ship_country_iso="LU",
bill_first_name=shop_customer.first_name,
bill_last_name=shop_customer.last_name,
bill_address_line_1="789 Shop Blvd",
bill_city="Luxembourg",
bill_postal_code="L-9999",
bill_country_iso="LU",
vat_regime="domestic",
vat_rate=Decimal("17.00"),
)
db.add(order)
db.flush()
# Create invoice for this order
invoice = Invoice(
vendor_id=test_vendor.id,
order_id=order.id,
invoice_number="INV00001",
invoice_date=datetime.now(UTC),
status=InvoiceStatus.ISSUED.value,
seller_details={"company_name": "Shop Test Company S.A."},
buyer_details={"name": f"{shop_customer.first_name} {shop_customer.last_name}"},
line_items=[],
vat_rate=Decimal("17.00"),
subtotal_cents=15000,
vat_amount_cents=2550,
total_cents=18050,
)
db.add(invoice)
db.commit()
db.refresh(order)
db.refresh(invoice)
return order, invoice
@pytest.fixture
def other_customer(db, test_vendor):
"""Create another customer for testing access controls."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
customer = Customer(
vendor_id=test_vendor.id,
email="othercustomer@example.com",
hashed_password=auth_manager.hash_password("otherpass123"),
first_name="Other",
last_name="Customer",
customer_number="OTHER001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def other_customer_order(db, test_vendor, other_customer):
"""Create an order for another customer."""
order = Order(
vendor_id=test_vendor.id,
customer_id=other_customer.id,
order_number="OTHER-ORD-001",
status="processing",
channel="direct",
order_date=datetime.now(UTC),
subtotal_cents=5000,
tax_amount_cents=850,
total_amount_cents=5850,
currency="EUR",
customer_email=other_customer.email,
customer_first_name=other_customer.first_name,
customer_last_name=other_customer.last_name,
ship_first_name=other_customer.first_name,
ship_last_name=other_customer.last_name,
ship_address_line_1="Other St",
ship_city="Other City",
ship_postal_code="00000",
ship_country_iso="LU",
bill_first_name=other_customer.first_name,
bill_last_name=other_customer.last_name,
bill_address_line_1="Other St",
bill_city="Other City",
bill_postal_code="00000",
bill_country_iso="LU",
)
db.add(order)
db.commit()
db.refresh(order)
return order
# Note: Shop API endpoints require vendor context from VendorContextMiddleware.
# In integration tests, we mock the middleware to inject the vendor.
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopOrdersListAPI:
"""Test shop orders list endpoint at /api/v1/shop/orders."""
def test_list_orders_requires_authentication(self, client, test_vendor):
"""Test that listing orders requires authentication."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.get("/api/v1/shop/orders")
# Without token, should get 401 or 403
assert response.status_code in [401, 403]
def test_list_orders_success(
self, client, shop_customer_headers, shop_order, test_vendor, shop_customer
):
"""Test listing customer orders successfully."""
# Mock vendor context and customer auth
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
# Mock the dependency to return our customer
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/orders",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert "orders" in data
assert "total" in data
def test_list_orders_empty(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test listing orders when customer has none."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/orders",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopOrderDetailAPI:
"""Test shop order detail endpoint at /api/v1/shop/orders/{order_id}."""
def test_get_order_detail_success(
self, client, shop_customer_headers, shop_order, test_vendor, shop_customer
):
"""Test getting order details successfully."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/orders/{shop_order.id}",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
assert data["order_number"] == "SHOP-ORD-001"
assert data["status"] == "pending"
# Check VAT fields are present
assert "vat_regime" in data
assert "vat_rate" in data
def test_get_order_detail_not_found(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test getting non-existent order returns 404."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/orders/99999",
headers=shop_customer_headers,
)
assert response.status_code == 404
def test_get_order_detail_other_customer(
self,
client,
shop_customer_headers,
other_customer_order,
test_vendor,
shop_customer,
):
"""Test cannot access another customer's order."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/orders/{other_customer_order.id}",
headers=shop_customer_headers,
)
# Should return 404 (not 403) to prevent enumeration
assert response.status_code == 404
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopOrderInvoiceDownloadAPI:
"""Test shop order invoice download at /api/v1/shop/orders/{order_id}/invoice."""
def test_download_invoice_pending_order_rejected(
self, client, shop_customer_headers, shop_order, test_vendor, shop_customer
):
"""Test cannot download invoice for pending orders."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/orders/{shop_order.id}/invoice",
headers=shop_customer_headers,
)
# Pending orders should not allow invoice download
assert response.status_code == 422
@pytest.mark.skip(reason="Requires PDF generation infrastructure")
def test_download_invoice_processing_order_creates_invoice(
self,
client,
shop_customer_headers,
shop_order_processing,
shop_invoice_settings,
test_vendor,
shop_customer,
):
"""Test downloading invoice for processing order creates it if needed."""
# This test requires actual PDF generation which may not be available
# in all environments. The logic is tested via:
# 1. test_download_invoice_pending_order_rejected - validates status check
# 2. Direct service tests for invoice creation
pass
@pytest.mark.skip(reason="Requires PDF generation infrastructure")
def test_download_invoice_existing_invoice(
self,
client,
shop_customer_headers,
shop_order_with_invoice,
test_vendor,
shop_customer,
):
"""Test downloading invoice when one already exists."""
# This test requires PDF file to exist on disk
# The service layer handles invoice retrieval properly
pass
def test_download_invoice_other_customer(
self,
client,
shop_customer_headers,
other_customer_order,
test_vendor,
shop_customer,
):
"""Test cannot download invoice for another customer's order."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/orders/{other_customer_order.id}/invoice",
headers=shop_customer_headers,
)
# Should return 404 to prevent enumeration
assert response.status_code == 404
def test_download_invoice_not_found(
self, client, shop_customer_headers, test_vendor, shop_customer
):
"""Test downloading invoice for non-existent order."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/orders/99999/invoice",
headers=shop_customer_headers,
)
assert response.status_code == 404
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestShopOrderVATFields:
"""Test VAT fields in order responses."""
def test_order_includes_vat_fields(
self, client, shop_customer_headers, shop_order, test_vendor, shop_customer
):
"""Test order response includes VAT fields."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
f"/api/v1/shop/orders/{shop_order.id}",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
# Verify VAT fields
assert data.get("vat_regime") == "domestic"
assert data.get("vat_rate") == 17.0
assert "Luxembourg VAT" in (data.get("vat_rate_label") or "")
def test_order_list_includes_vat_fields(
self, client, shop_customer_headers, shop_order, test_vendor, shop_customer
):
"""Test order list includes VAT fields."""
with patch("app.api.v1.shop.orders.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.deps._validate_customer_token") as mock_validate:
mock_validate.return_value = shop_customer
response = client.get(
"/api/v1/shop/orders",
headers=shop_customer_headers,
)
assert response.status_code == 200
data = response.json()
if data["orders"]:
order = data["orders"][0]
assert "vat_regime" in order
assert "vat_rate" in order

453
tests/unit/services/test_customer_address_service.py Normal file
View File

@@ -0,0 +1,453 @@
# tests/unit/services/test_customer_address_service.py
"""
Unit tests for CustomerAddressService.
"""
import pytest
from app.exceptions import AddressLimitExceededException, AddressNotFoundException
from app.services.customer_address_service import CustomerAddressService
from models.database.customer import CustomerAddress
from models.schema.customer import CustomerAddressCreate, CustomerAddressUpdate
@pytest.fixture
def address_service():
"""Create CustomerAddressService instance."""
return CustomerAddressService()
@pytest.fixture
def multiple_addresses(db, test_vendor, test_customer):
"""Create multiple addresses for testing."""
addresses = []
for i in range(3):
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping" if i < 2 else "billing",
first_name=f"First{i}",
last_name=f"Last{i}",
address_line_1=f"{i+1} Test Street",
city="Luxembourg",
postal_code=f"L-{1000+i}",
country_name="Luxembourg",
country_iso="LU",
is_default=(i == 0), # First shipping is default
)
db.add(address)
addresses.append(address)
db.commit()
for a in addresses:
db.refresh(a)
return addresses
@pytest.mark.unit
class TestCustomerAddressServiceList:
"""Tests for list_addresses method."""
def test_list_addresses_empty(self, db, address_service, test_vendor, test_customer):
"""Test listing addresses when none exist."""
addresses = address_service.list_addresses(
db, vendor_id=test_vendor.id, customer_id=test_customer.id
)
assert addresses == []
def test_list_addresses_basic(
self, db, address_service, test_vendor, test_customer, test_customer_address
):
"""Test basic address listing."""
addresses = address_service.list_addresses(
db, vendor_id=test_vendor.id, customer_id=test_customer.id
)
assert len(addresses) == 1
assert addresses[0].id == test_customer_address.id
def test_list_addresses_ordered_by_default(
self, db, address_service, test_vendor, test_customer, multiple_addresses
):
"""Test addresses are ordered by default flag first."""
addresses = address_service.list_addresses(
db, vendor_id=test_vendor.id, customer_id=test_customer.id
)
# Default address should be first
assert addresses[0].is_default is True
def test_list_addresses_vendor_isolation(
self, db, address_service, test_vendor, test_customer, test_customer_address
):
"""Test addresses are isolated by vendor."""
# Query with different vendor ID
addresses = address_service.list_addresses(
db, vendor_id=99999, customer_id=test_customer.id
)
assert addresses == []
@pytest.mark.unit
class TestCustomerAddressServiceGet:
"""Tests for get_address method."""
def test_get_address_success(
self, db, address_service, test_vendor, test_customer, test_customer_address
):
"""Test getting address by ID."""
address = address_service.get_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=test_customer_address.id,
)
assert address.id == test_customer_address.id
assert address.first_name == test_customer_address.first_name
def test_get_address_not_found(
self, db, address_service, test_vendor, test_customer
):
"""Test error when address not found."""
with pytest.raises(AddressNotFoundException):
address_service.get_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=99999,
)
def test_get_address_wrong_customer(
self, db, address_service, test_vendor, test_customer, test_customer_address
):
"""Test cannot get another customer's address."""
with pytest.raises(AddressNotFoundException):
address_service.get_address(
db,
vendor_id=test_vendor.id,
customer_id=99999, # Different customer
address_id=test_customer_address.id,
)
@pytest.mark.unit
class TestCustomerAddressServiceGetDefault:
"""Tests for get_default_address method."""
def test_get_default_address_exists(
self, db, address_service, test_vendor, test_customer, multiple_addresses
):
"""Test getting default shipping address."""
address = address_service.get_default_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
)
assert address is not None
assert address.is_default is True
assert address.address_type == "shipping"
def test_get_default_address_not_set(
self, db, address_service, test_vendor, test_customer, multiple_addresses
):
"""Test getting default billing when none is set."""
# Remove default from billing (none was set as default)
address = address_service.get_default_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="billing",
)
# The billing address exists but is not default
assert address is None
@pytest.mark.unit
class TestCustomerAddressServiceCreate:
"""Tests for create_address method."""
def test_create_address_success(
self, db, address_service, test_vendor, test_customer
):
"""Test creating a new address."""
address_data = CustomerAddressCreate(
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 New Street",
city="Luxembourg",
postal_code="L-1234",
country_name="Luxembourg",
country_iso="LU",
is_default=False,
)
address = address_service.create_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_data=address_data,
)
db.commit()
assert address.id is not None
assert address.first_name == "John"
assert address.last_name == "Doe"
assert address.country_iso == "LU"
assert address.country_name == "Luxembourg"
def test_create_address_with_company(
self, db, address_service, test_vendor, test_customer
):
"""Test creating address with company name."""
address_data = CustomerAddressCreate(
address_type="billing",
first_name="Jane",
last_name="Doe",
company="Acme Corp",
address_line_1="456 Business Ave",
city="Luxembourg",
postal_code="L-5678",
country_name="Luxembourg",
country_iso="LU",
is_default=False,
)
address = address_service.create_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_data=address_data,
)
db.commit()
assert address.company == "Acme Corp"
def test_create_address_default_clears_others(
self, db, address_service, test_vendor, test_customer, multiple_addresses
):
"""Test creating default address clears other defaults of same type."""
# First address is default shipping
assert multiple_addresses[0].is_default is True
address_data = CustomerAddressCreate(
address_type="shipping",
first_name="New",
last_name="Default",
address_line_1="789 Main St",
city="Luxembourg",
postal_code="L-9999",
country_name="Luxembourg",
country_iso="LU",
is_default=True,
)
new_address = address_service.create_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_data=address_data,
)
db.commit()
# New address should be default
assert new_address.is_default is True
# Old default should be cleared
db.refresh(multiple_addresses[0])
assert multiple_addresses[0].is_default is False
def test_create_address_limit_exceeded(
self, db, address_service, test_vendor, test_customer
):
"""Test error when max addresses reached."""
# Create 10 addresses (max limit)
for i in range(10):
addr = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name=f"Test{i}",
last_name="User",
address_line_1=f"{i} Street",
city="City",
postal_code="12345",
country_name="Luxembourg",
country_iso="LU",
)
db.add(addr)
db.commit()
# Try to create 11th address
address_data = CustomerAddressCreate(
address_type="shipping",
first_name="Eleventh",
last_name="User",
address_line_1="11 Street",
city="City",
postal_code="12345",
country_name="Luxembourg",
country_iso="LU",
is_default=False,
)
with pytest.raises(AddressLimitExceededException):
address_service.create_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_data=address_data,
)
@pytest.mark.unit
class TestCustomerAddressServiceUpdate:
"""Tests for update_address method."""
def test_update_address_success(
self, db, address_service, test_vendor, test_customer, test_customer_address
):
"""Test updating an address."""
update_data = CustomerAddressUpdate(
first_name="Updated",
city="New City",
)
address = address_service.update_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=test_customer_address.id,
address_data=update_data,
)
db.commit()
assert address.first_name == "Updated"
assert address.city == "New City"
# Unchanged fields should remain
assert address.last_name == test_customer_address.last_name
def test_update_address_set_default(
self, db, address_service, test_vendor, test_customer, multiple_addresses
):
"""Test setting address as default clears others."""
# Second address is not default
assert multiple_addresses[1].is_default is False
update_data = CustomerAddressUpdate(is_default=True)
address = address_service.update_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=multiple_addresses[1].id,
address_data=update_data,
)
db.commit()
assert address.is_default is True
# Old default should be cleared
db.refresh(multiple_addresses[0])
assert multiple_addresses[0].is_default is False
def test_update_address_not_found(
self, db, address_service, test_vendor, test_customer
):
"""Test error when address not found."""
update_data = CustomerAddressUpdate(first_name="Test")
with pytest.raises(AddressNotFoundException):
address_service.update_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=99999,
address_data=update_data,
)
@pytest.mark.unit
class TestCustomerAddressServiceDelete:
"""Tests for delete_address method."""
def test_delete_address_success(
self, db, address_service, test_vendor, test_customer, test_customer_address
):
"""Test deleting an address."""
address_id = test_customer_address.id
address_service.delete_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=address_id,
)
db.commit()
# Address should be gone
with pytest.raises(AddressNotFoundException):
address_service.get_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=address_id,
)
def test_delete_address_not_found(
self, db, address_service, test_vendor, test_customer
):
"""Test error when deleting non-existent address."""
with pytest.raises(AddressNotFoundException):
address_service.delete_address(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=99999,
)
@pytest.mark.unit
class TestCustomerAddressServiceSetDefault:
"""Tests for set_default method."""
def test_set_default_success(
self, db, address_service, test_vendor, test_customer, multiple_addresses
):
"""Test setting address as default."""
# Second shipping address is not default
assert multiple_addresses[1].is_default is False
assert multiple_addresses[1].address_type == "shipping"
address = address_service.set_default(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=multiple_addresses[1].id,
)
db.commit()
assert address.is_default is True
# Old default should be cleared
db.refresh(multiple_addresses[0])
assert multiple_addresses[0].is_default is False
def test_set_default_not_found(
self, db, address_service, test_vendor, test_customer
):
"""Test error when address not found."""
with pytest.raises(AddressNotFoundException):
address_service.set_default(
db,
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_id=99999,
)