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

refactor: rename shop to storefront throughout codebase

Browse Source 
Rename "shop" to "storefront" as not all platforms sell items -
storefront is a more accurate term for the customer-facing interface.

Changes:
- Rename app/api/v1/shop/ → app/api/v1/storefront/
- Rename app/routes/shop_pages.py → app/routes/storefront_pages.py
- Rename app/modules/cms/routes/api/shop.py → storefront.py
- Rename tests/integration/api/v1/shop/ → storefront/
- Update API prefix from /api/v1/shop to /api/v1/storefront
- Update route tags from shop-* to storefront-*
- Rename get_shop_context() → get_storefront_context()
- Update architecture rules to reference storefront paths
- Update all test API endpoint paths

This is Phase 2 of the storefront module restructure plan.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-29 22:42:01 +01:00
parent 228163d920
commit 3e86d4b58b
20 changed files with 140 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -0,0 +1,621 @@
# tests/integration/api/v1/storefront/test_addresses.py
"""Integration tests for shop addresses API endpoints.
Tests the /api/v1/storefront/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 app.modules.customers.models.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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/addresses/{other_customer_address.id}/default",
headers=shop_customer_headers,
)
assert response.status_code == 404

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

@@ -0,0 +1,557 @@
# tests/integration/api/v1/storefront/test_orders.py
"""Integration tests for shop orders API endpoints.
Tests the /api/v1/storefront/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 app.modules.customers.models.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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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/storefront/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

451
tests/integration/api/v1/storefront/test_password_reset.py Normal file
View File

@@ -0,0 +1,451 @@
# tests/integration/api/v1/storefront/test_password_reset.py
"""Integration tests for shop password reset API endpoints.
Tests the /api/v1/storefront/auth/forgot-password and /api/v1/storefront/auth/reset-password endpoints.
"""
from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
from app.modules.customers.models.customer import Customer
from models.database.password_reset_token import PasswordResetToken
@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="customer@example.com",
hashed_password=auth_manager.hash_password("oldpassword123"),
first_name="Test",
last_name="Customer",
customer_number="CUST001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def inactive_customer(db, test_vendor):
"""Create an inactive customer for testing."""
from middleware.auth import AuthManager
auth_manager = AuthManager()
customer = Customer(
vendor_id=test_vendor.id,
email="inactive@example.com",
hashed_password=auth_manager.hash_password("password123"),
first_name="Inactive",
last_name="Customer",
customer_number="CUST002",
is_active=False,
)
db.add(customer)
db.commit()
db.refresh(customer)
return customer
@pytest.fixture
def valid_reset_token(db, shop_customer):
"""Create a valid password reset token."""
token = PasswordResetToken.create_for_customer(db, shop_customer.id)
db.commit()
return token
@pytest.fixture
def expired_reset_token(db, shop_customer):
"""Create an expired password reset token."""
import secrets
token = secrets.token_urlsafe(32)
token_hash = PasswordResetToken.hash_token(token)
reset_token = PasswordResetToken(
customer_id=shop_customer.id,
token_hash=token_hash,
expires_at=datetime.utcnow() - timedelta(hours=2), # Already expired
)
db.add(reset_token)
db.commit()
return token
@pytest.fixture
def used_reset_token(db, shop_customer):
"""Create a used password reset token."""
import secrets
token = secrets.token_urlsafe(32)
token_hash = PasswordResetToken.hash_token(token)
reset_token = PasswordResetToken(
customer_id=shop_customer.id,
token_hash=token_hash,
expires_at=datetime.utcnow() + timedelta(hours=1),
used_at=datetime.utcnow(), # Already used
)
db.add(reset_token)
db.commit()
return token
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestForgotPasswordAPI:
"""Test forgot password endpoint at /api/v1/storefront/auth/forgot-password."""
def test_forgot_password_existing_customer(
self, client, db, test_vendor, shop_customer
):
"""Test password reset request for existing customer."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
# Mock email service to avoid actual email sending
with patch("app.api.v1.shop.auth.EmailService") as mock_email_service:
mock_instance = MagicMock()
mock_email_service.return_value = mock_instance
response = client.post(
"/api/v1/storefront/auth/forgot-password",
params={"email": shop_customer.email},
)
assert response.status_code == 200
data = response.json()
assert "password reset link has been sent" in data["message"].lower()
# Verify email was sent
mock_instance.send_template.assert_called_once()
call_kwargs = mock_instance.send_template.call_args.kwargs
assert call_kwargs["template_code"] == "password_reset"
assert call_kwargs["to_email"] == shop_customer.email
def test_forgot_password_nonexistent_email(self, client, db, test_vendor):
"""Test password reset request for non-existent email (same response)."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/forgot-password",
params={"email": "nonexistent@example.com"},
)
# Should return same success message to prevent email enumeration
assert response.status_code == 200
data = response.json()
assert "password reset link has been sent" in data["message"].lower()
def test_forgot_password_inactive_customer(
self, client, db, test_vendor, inactive_customer
):
"""Test password reset request for inactive customer."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/forgot-password",
params={"email": inactive_customer.email},
)
# Should return same success message (inactive customers can't reset)
assert response.status_code == 200
data = response.json()
assert "password reset link has been sent" in data["message"].lower()
def test_forgot_password_creates_token(
self, client, db, test_vendor, shop_customer
):
"""Test that forgot password creates a token in the database."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.v1.shop.auth.EmailService"):
response = client.post(
"/api/v1/storefront/auth/forgot-password",
params={"email": shop_customer.email},
)
assert response.status_code == 200
# Verify token was created
token = (
db.query(PasswordResetToken)
.filter(PasswordResetToken.customer_id == shop_customer.id)
.first()
)
assert token is not None
assert token.used_at is None
assert token.expires_at > datetime.utcnow()
def test_forgot_password_invalidates_old_tokens(
self, client, db, test_vendor, shop_customer, valid_reset_token
):
"""Test that requesting new token invalidates old ones."""
# Get the old token record
old_token_count = (
db.query(PasswordResetToken)
.filter(
PasswordResetToken.customer_id == shop_customer.id,
PasswordResetToken.used_at.is_(None),
)
.count()
)
assert old_token_count == 1
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
with patch("app.api.v1.shop.auth.EmailService"):
response = client.post(
"/api/v1/storefront/auth/forgot-password",
params={"email": shop_customer.email},
)
assert response.status_code == 200
# Old token should be deleted, new one created
new_token_count = (
db.query(PasswordResetToken)
.filter(
PasswordResetToken.customer_id == shop_customer.id,
PasswordResetToken.used_at.is_(None),
)
.count()
)
assert new_token_count == 1
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestResetPasswordAPI:
"""Test reset password endpoint at /api/v1/storefront/auth/reset-password."""
def test_reset_password_success(
self, client, db, test_vendor, shop_customer, valid_reset_token
):
"""Test successful password reset."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
new_password = "newpassword123"
response = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": valid_reset_token,
"new_password": new_password,
},
)
assert response.status_code == 200
data = response.json()
assert "password reset successfully" in data["message"].lower()
# Verify password was changed
db.refresh(shop_customer)
from middleware.auth import AuthManager
auth_manager = AuthManager()
assert auth_manager.verify_password(
new_password, shop_customer.hashed_password
)
def test_reset_password_token_marked_used(
self, client, db, test_vendor, shop_customer, valid_reset_token
):
"""Test that token is marked as used after successful reset."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": valid_reset_token,
"new_password": "newpassword123",
},
)
assert response.status_code == 200
# Verify token is marked as used
token_record = (
db.query(PasswordResetToken)
.filter(PasswordResetToken.customer_id == shop_customer.id)
.first()
)
assert token_record.used_at is not None
def test_reset_password_invalid_token(self, client, db, test_vendor):
"""Test reset with invalid token."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": "invalid_token_12345",
"new_password": "newpassword123",
},
)
assert response.status_code == 400
def test_reset_password_expired_token(
self, client, db, test_vendor, shop_customer, expired_reset_token
):
"""Test reset with expired token."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": expired_reset_token,
"new_password": "newpassword123",
},
)
assert response.status_code == 400
def test_reset_password_used_token(
self, client, db, test_vendor, shop_customer, used_reset_token
):
"""Test reset with already used token."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": used_reset_token,
"new_password": "newpassword123",
},
)
assert response.status_code == 400
def test_reset_password_short_password(
self, client, db, test_vendor, shop_customer, valid_reset_token
):
"""Test reset with password that's too short."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
response = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": valid_reset_token,
"new_password": "short", # Less than 8 chars
},
)
assert response.status_code == 400
def test_reset_password_cannot_reuse_token(
self, client, db, test_vendor, shop_customer, valid_reset_token
):
"""Test that token cannot be reused after successful reset."""
with patch("app.api.v1.shop.auth.getattr") as mock_getattr:
mock_getattr.return_value = test_vendor
# First reset should succeed
response1 = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": valid_reset_token,
"new_password": "newpassword123",
},
)
assert response1.status_code == 200
# Second reset with same token should fail
response2 = client.post(
"/api/v1/storefront/auth/reset-password",
params={
"reset_token": valid_reset_token,
"new_password": "anotherpassword123",
},
)
assert response2.status_code == 400
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.shop
class TestPasswordResetTokenModel:
"""Test PasswordResetToken model functionality."""
def test_token_hash_is_deterministic(self):
"""Test that hashing the same token produces the same hash."""
token = "test_token_12345"
hash1 = PasswordResetToken.hash_token(token)
hash2 = PasswordResetToken.hash_token(token)
assert hash1 == hash2
def test_different_tokens_produce_different_hashes(self):
"""Test that different tokens produce different hashes."""
hash1 = PasswordResetToken.hash_token("token1")
hash2 = PasswordResetToken.hash_token("token2")
assert hash1 != hash2
def test_create_for_customer_returns_plaintext(self, db, shop_customer):
"""Test that create_for_customer returns plaintext token."""
token = PasswordResetToken.create_for_customer(db, shop_customer.id)
db.commit()
# Token should be URL-safe base64
assert token is not None
assert len(token) > 20 # secrets.token_urlsafe(32) produces ~43 chars
def test_find_valid_token_works_with_plaintext(self, db, shop_customer):
"""Test that find_valid_token works with plaintext token."""
plaintext_token = PasswordResetToken.create_for_customer(db, shop_customer.id)
db.commit()
found = PasswordResetToken.find_valid_token(db, plaintext_token)
assert found is not None
assert found.customer_id == shop_customer.id
def test_find_valid_token_returns_none_for_invalid(self, db):
"""Test that find_valid_token returns None for invalid token."""
found = PasswordResetToken.find_valid_token(db, "invalid_token")
assert found is None
def test_mark_used_sets_timestamp(self, db, shop_customer):
"""Test that mark_used sets the used_at timestamp."""
plaintext_token = PasswordResetToken.create_for_customer(db, shop_customer.id)
db.commit()
token_record = PasswordResetToken.find_valid_token(db, plaintext_token)
assert token_record.used_at is None
token_record.mark_used(db)
db.commit()
assert token_record.used_at is not None
def test_used_token_not_found_by_find_valid(self, db, shop_customer):
"""Test that used tokens are not returned by find_valid_token."""
plaintext_token = PasswordResetToken.create_for_customer(db, shop_customer.id)
db.commit()
token_record = PasswordResetToken.find_valid_token(db, plaintext_token)
token_record.mark_used(db)
db.commit()
# Should not find the used token
found = PasswordResetToken.find_valid_token(db, plaintext_token)
assert found is None