refactor: split database model tests into organized modules

Split the monolithic test_database_models.py into focused test modules:

Database model tests (tests/unit/models/database/):
- test_customer.py: Customer model and authentication tests
- test_inventory.py: Inventory model tests
- test_marketplace_import_job.py: Import job model tests
- test_marketplace_product.py: Marketplace product model tests
- test_order.py: Order and OrderItem model tests
- test_product.py: Product model tests
- test_team.py: Team invitation and membership tests
- test_user.py: User model tests
- test_vendor.py: Vendor model tests

Schema validation tests (tests/unit/models/schema/):
- test_auth.py: Auth schema validation tests
- test_customer.py: Customer schema validation tests
- test_inventory.py: Inventory schema validation tests
- test_marketplace_import_job.py: Import job schema tests
- test_order.py: Order schema validation tests
- test_product.py: Product schema validation tests

This improves test organization and makes it easier to find
and maintain tests for specific models.

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

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-12-05 21:42:26 +01:00
parent aaff799b5e
commit ca9f17fc37
19 changed files with 3533 additions and 598 deletions

View File

@@ -0,0 +1,2 @@
# tests/unit/models/database/__init__.py
"""Database model unit tests."""

View File

@@ -0,0 +1,240 @@
# tests/unit/models/database/test_customer.py
"""Unit tests for Customer and CustomerAddress database models."""
import pytest
from models.database.customer import Customer, CustomerAddress
@pytest.mark.unit
@pytest.mark.database
class TestCustomerModel:
"""Test Customer model."""
def test_customer_creation(self, db, test_vendor):
"""Test Customer model with vendor isolation."""
customer = Customer(
vendor_id=test_vendor.id,
email="customer@example.com",
hashed_password="hashed_password",
first_name="John",
last_name="Doe",
customer_number="CUST001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.id is not None
assert customer.vendor_id == test_vendor.id
assert customer.email == "customer@example.com"
assert customer.customer_number == "CUST001"
assert customer.first_name == "John"
assert customer.last_name == "Doe"
assert customer.vendor.vendor_code == test_vendor.vendor_code
def test_customer_default_values(self, db, test_vendor):
"""Test Customer model default values."""
customer = Customer(
vendor_id=test_vendor.id,
email="defaults@example.com",
hashed_password="hash",
customer_number="CUST_DEFAULTS",
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.is_active is True # Default
assert customer.marketing_consent is False # Default
assert customer.total_orders == 0 # Default
assert customer.total_spent == 0 # Default
def test_customer_full_name_property(self, db, test_vendor):
"""Test Customer full_name computed property."""
customer = Customer(
vendor_id=test_vendor.id,
email="fullname@example.com",
hashed_password="hash",
customer_number="CUST_FULLNAME",
first_name="Jane",
last_name="Smith",
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.full_name == "Jane Smith"
def test_customer_full_name_fallback_to_email(self, db, test_vendor):
"""Test Customer full_name falls back to email when names not set."""
customer = Customer(
vendor_id=test_vendor.id,
email="noname@example.com",
hashed_password="hash",
customer_number="CUST_NONAME",
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.full_name == "noname@example.com"
def test_customer_optional_fields(self, db, test_vendor):
"""Test Customer with optional fields."""
customer = Customer(
vendor_id=test_vendor.id,
email="optional@example.com",
hashed_password="hash",
customer_number="CUST_OPT",
phone="+352123456789",
preferences={"language": "en", "currency": "EUR"},
marketing_consent=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.phone == "+352123456789"
assert customer.preferences == {"language": "en", "currency": "EUR"}
assert customer.marketing_consent is True
def test_customer_vendor_relationship(self, db, test_vendor):
"""Test Customer-Vendor relationship."""
customer = Customer(
vendor_id=test_vendor.id,
email="relationship@example.com",
hashed_password="hash",
customer_number="CUST_REL",
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.vendor is not None
assert customer.vendor.id == test_vendor.id
@pytest.mark.unit
@pytest.mark.database
class TestCustomerAddressModel:
"""Test CustomerAddress model."""
def test_customer_address_creation(self, db, test_vendor, test_customer):
"""Test CustomerAddress model."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
is_default=True,
)
db.add(address)
db.commit()
db.refresh(address)
assert address.id is not None
assert address.vendor_id == test_vendor.id
assert address.customer_id == test_customer.id
assert address.address_type == "shipping"
assert address.is_default is True
def test_customer_address_types(self, db, test_vendor, test_customer):
"""Test CustomerAddress with different address types."""
shipping_address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Shipping St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
db.add(shipping_address)
billing_address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="billing",
first_name="John",
last_name="Doe",
address_line_1="456 Billing Ave",
city="Luxembourg",
postal_code="L-5678",
country="Luxembourg",
)
db.add(billing_address)
db.commit()
assert shipping_address.address_type == "shipping"
assert billing_address.address_type == "billing"
def test_customer_address_optional_fields(self, db, test_vendor, test_customer):
"""Test CustomerAddress with optional fields."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name="John",
last_name="Doe",
company="ACME Corp",
address_line_1="123 Main St",
address_line_2="Suite 100",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
db.add(address)
db.commit()
db.refresh(address)
assert address.company == "ACME Corp"
assert address.address_line_2 == "Suite 100"
def test_customer_address_default_values(self, db, test_vendor, test_customer):
"""Test CustomerAddress default values."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
db.add(address)
db.commit()
db.refresh(address)
assert address.is_default is False # Default
def test_customer_address_relationships(self, db, test_vendor, test_customer):
"""Test CustomerAddress relationships."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
db.add(address)
db.commit()
db.refresh(address)
assert address.customer is not None
assert address.customer.id == test_customer.id

View File

@@ -0,0 +1,143 @@
# tests/unit/models/database/test_inventory.py
"""Unit tests for Inventory database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.inventory import Inventory
@pytest.mark.unit
@pytest.mark.database
class TestInventoryModel:
"""Test Inventory model."""
def test_inventory_creation_with_product(self, db, test_vendor, test_product):
"""Test Inventory model linked to product."""
inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=150,
reserved_quantity=10,
gtin=test_product.marketplace_product.gtin,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
assert inventory.id is not None
assert inventory.product_id == test_product.id
assert inventory.vendor_id == test_vendor.id
assert inventory.location == "WAREHOUSE_A"
assert inventory.quantity == 150
assert inventory.reserved_quantity == 10
assert inventory.available_quantity == 140 # 150 - 10
def test_inventory_unique_product_location(self, db, test_vendor, test_product):
"""Test unique constraint on product_id + location."""
inventory1 = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=100,
)
db.add(inventory1)
db.commit()
# Same product + location should fail
with pytest.raises(IntegrityError):
inventory2 = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=50,
)
db.add(inventory2)
db.commit()
def test_inventory_same_product_different_location(self, db, test_vendor, test_product):
"""Test same product can have inventory in different locations."""
inventory1 = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=100,
)
db.add(inventory1)
db.commit()
# Same product in different location should succeed
inventory2 = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_B",
quantity=50,
)
db.add(inventory2)
db.commit()
db.refresh(inventory2)
assert inventory2.id is not None
assert inventory2.location == "WAREHOUSE_B"
def test_inventory_default_values(self, db, test_vendor, test_product):
"""Test Inventory model default values."""
inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="DEFAULT_LOC",
quantity=100,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
assert inventory.reserved_quantity == 0 # Default
assert inventory.available_quantity == 100 # quantity - reserved
def test_inventory_available_quantity_property(self, db, test_vendor, test_product):
"""Test available_quantity computed property."""
inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="PROP_TEST",
quantity=200,
reserved_quantity=50,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
assert inventory.available_quantity == 150 # 200 - 50
def test_inventory_relationships(self, db, test_vendor, test_product):
"""Test Inventory relationships."""
inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="REL_TEST",
quantity=100,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
assert inventory.product is not None
assert inventory.vendor is not None
assert inventory.product.id == test_product.id
assert inventory.vendor.id == test_vendor.id
def test_inventory_without_gtin(self, db, test_vendor, test_product):
"""Test Inventory can be created without GTIN."""
inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="NO_GTIN",
quantity=100,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
assert inventory.gtin is None

View File

@@ -0,0 +1,129 @@
# tests/unit/models/database/test_marketplace_import_job.py
"""Unit tests for MarketplaceImportJob database model."""
import pytest
from models.database.marketplace_import_job import MarketplaceImportJob
@pytest.mark.unit
@pytest.mark.database
class TestMarketplaceImportJobModel:
"""Test MarketplaceImportJob model."""
def test_import_job_creation(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob model with relationships."""
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
marketplace="Letzshop",
source_url="https://example.com/feed.csv",
status="pending",
imported_count=0,
updated_count=0,
error_count=0,
total_processed=0,
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.id is not None
assert import_job.vendor_id == test_vendor.id
assert import_job.user_id == test_user.id
assert import_job.marketplace == "Letzshop"
assert import_job.source_url == "https://example.com/feed.csv"
assert import_job.status == "pending"
assert import_job.vendor.vendor_code == test_vendor.vendor_code
assert import_job.user.username == test_user.username
def test_import_job_default_values(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob default values."""
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
source_url="https://example.com/feed.csv",
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.marketplace == "Letzshop" # Default
assert import_job.status == "pending" # Default
assert import_job.imported_count == 0 # Default
assert import_job.updated_count == 0 # Default
assert import_job.error_count == 0 # Default
assert import_job.total_processed == 0 # Default
def test_import_job_status_values(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob with different status values."""
statuses = ["pending", "processing", "completed", "failed", "completed_with_errors"]
for i, status in enumerate(statuses):
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
source_url=f"https://example.com/feed_{i}.csv",
status=status,
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.status == status
def test_import_job_counts(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob count fields."""
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
source_url="https://example.com/feed.csv",
status="completed",
imported_count=100,
updated_count=50,
error_count=5,
total_processed=155,
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.imported_count == 100
assert import_job.updated_count == 50
assert import_job.error_count == 5
assert import_job.total_processed == 155
def test_import_job_error_message(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob with error message."""
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
source_url="https://example.com/feed.csv",
status="failed",
error_message="Connection timeout while fetching CSV",
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.error_message == "Connection timeout while fetching CSV"
def test_import_job_relationships(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob relationships."""
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
source_url="https://example.com/feed.csv",
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.vendor is not None
assert import_job.user is not None
assert import_job.vendor.id == test_vendor.id
assert import_job.user.id == test_user.id

View File

@@ -0,0 +1,131 @@
# tests/unit/models/database/test_marketplace_product.py
"""Unit tests for MarketplaceProduct database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.marketplace_product import MarketplaceProduct
@pytest.mark.unit
@pytest.mark.database
class TestMarketplaceProductModel:
"""Test MarketplaceProduct model."""
def test_marketplace_product_creation(self, db):
"""Test MarketplaceProduct model creation."""
marketplace_product = MarketplaceProduct(
marketplace_product_id="DB_TEST_001",
title="Database Test Product",
description="Testing product model",
price="25.99",
currency="USD",
brand="DBTest",
gtin="1234567890123",
availability="in stock",
marketplace="Letzshop",
vendor_name="Test Vendor",
)
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
assert marketplace_product.id is not None
assert marketplace_product.marketplace_product_id == "DB_TEST_001"
assert marketplace_product.title == "Database Test Product"
assert marketplace_product.marketplace == "Letzshop"
assert marketplace_product.created_at is not None
def test_marketplace_product_id_uniqueness(self, db):
"""Test unique marketplace_product_id constraint."""
product1 = MarketplaceProduct(
marketplace_product_id="UNIQUE_001",
title="Product 1",
marketplace="Letzshop",
)
db.add(product1)
db.commit()
# Duplicate marketplace_product_id should raise error
with pytest.raises(IntegrityError):
product2 = MarketplaceProduct(
marketplace_product_id="UNIQUE_001",
title="Product 2",
marketplace="Letzshop",
)
db.add(product2)
db.commit()
def test_marketplace_product_all_fields(self, db):
"""Test MarketplaceProduct with all optional fields."""
marketplace_product = MarketplaceProduct(
marketplace_product_id="FULL_001",
title="Full Product",
description="Complete product description",
link="https://example.com/product",
image_link="https://example.com/image.jpg",
availability="in stock",
price="99.99",
brand="TestBrand",
gtin="9876543210123",
mpn="MPN123",
condition="new",
adult="no",
age_group="adult",
color="blue",
gender="unisex",
material="cotton",
pattern="solid",
size="M",
google_product_category="Apparel & Accessories",
product_type="Clothing",
currency="EUR",
marketplace="Letzshop",
vendor_name="Full Vendor",
)
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
assert marketplace_product.brand == "TestBrand"
assert marketplace_product.gtin == "9876543210123"
assert marketplace_product.color == "blue"
assert marketplace_product.size == "M"
def test_marketplace_product_custom_labels(self, db):
"""Test MarketplaceProduct with custom labels."""
marketplace_product = MarketplaceProduct(
marketplace_product_id="LABELS_001",
title="Labeled Product",
marketplace="Letzshop",
custom_label_0="Label0",
custom_label_1="Label1",
custom_label_2="Label2",
custom_label_3="Label3",
custom_label_4="Label4",
)
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
assert marketplace_product.custom_label_0 == "Label0"
assert marketplace_product.custom_label_4 == "Label4"
def test_marketplace_product_minimal_fields(self, db):
"""Test MarketplaceProduct with only required fields."""
marketplace_product = MarketplaceProduct(
marketplace_product_id="MINIMAL_001",
title="Minimal Product",
)
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
assert marketplace_product.id is not None
assert marketplace_product.marketplace_product_id == "MINIMAL_001"
assert marketplace_product.title == "Minimal Product"
assert marketplace_product.description is None
assert marketplace_product.price is None

View File

@@ -0,0 +1,242 @@
# tests/unit/models/database/test_order.py
"""Unit tests for Order and OrderItem database models."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.order import Order, OrderItem
@pytest.mark.unit
@pytest.mark.database
class TestOrderModel:
"""Test Order model."""
def test_order_creation(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test Order model with customer relationship."""
order = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="ORD-001",
status="pending",
subtotal=99.99,
total_amount=99.99,
currency="EUR",
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order)
db.commit()
db.refresh(order)
assert order.id is not None
assert order.vendor_id == test_vendor.id
assert order.customer_id == test_customer.id
assert order.order_number == "ORD-001"
assert order.status == "pending"
assert float(order.total_amount) == 99.99
def test_order_number_uniqueness(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test order_number unique constraint."""
order1 = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="UNIQUE-ORD-001",
status="pending",
subtotal=50.00,
total_amount=50.00,
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order1)
db.commit()
# Duplicate order number should fail
with pytest.raises(IntegrityError):
order2 = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="UNIQUE-ORD-001",
status="pending",
subtotal=75.00,
total_amount=75.00,
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order2)
db.commit()
def test_order_status_values(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test Order with different status values."""
statuses = ["pending", "confirmed", "processing", "shipped", "delivered", "cancelled"]
for i, status in enumerate(statuses):
order = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number=f"STATUS-ORD-{i:03d}",
status=status,
subtotal=50.00,
total_amount=50.00,
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order)
db.commit()
db.refresh(order)
assert order.status == status
def test_order_amounts(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test Order amount fields."""
order = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="AMOUNTS-ORD-001",
status="pending",
subtotal=100.00,
tax_amount=20.00,
shipping_amount=10.00,
discount_amount=5.00,
total_amount=125.00,
currency="EUR",
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order)
db.commit()
db.refresh(order)
assert float(order.subtotal) == 100.00
assert float(order.tax_amount) == 20.00
assert float(order.shipping_amount) == 10.00
assert float(order.discount_amount) == 5.00
assert float(order.total_amount) == 125.00
def test_order_relationships(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test Order relationships."""
order = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="REL-ORD-001",
status="pending",
subtotal=50.00,
total_amount=50.00,
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order)
db.commit()
db.refresh(order)
assert order.vendor is not None
assert order.customer is not None
assert order.vendor.id == test_vendor.id
assert order.customer.id == test_customer.id
@pytest.mark.unit
@pytest.mark.database
class TestOrderItemModel:
"""Test OrderItem model."""
def test_order_item_creation(self, db, test_order, test_product):
"""Test OrderItem model."""
order_item = OrderItem(
order_id=test_order.id,
product_id=test_product.id,
product_name=test_product.marketplace_product.title,
product_sku=test_product.product_id or "SKU001",
quantity=2,
unit_price=49.99,
total_price=99.98,
)
db.add(order_item)
db.commit()
db.refresh(order_item)
assert order_item.id is not None
assert order_item.order_id == test_order.id
assert order_item.product_id == test_product.id
assert order_item.quantity == 2
assert float(order_item.unit_price) == 49.99
assert float(order_item.total_price) == 99.98
def test_order_item_stores_product_snapshot(self, db, test_order, test_product):
"""Test OrderItem stores product name and SKU as snapshot."""
order_item = OrderItem(
order_id=test_order.id,
product_id=test_product.id,
product_name="Snapshot Product Name",
product_sku="SNAPSHOT-SKU-001",
quantity=1,
unit_price=25.00,
total_price=25.00,
)
db.add(order_item)
db.commit()
db.refresh(order_item)
assert order_item.id is not None
assert order_item.product_name == "Snapshot Product Name"
assert order_item.product_sku == "SNAPSHOT-SKU-001"
def test_order_item_relationships(self, db, test_order, test_product):
"""Test OrderItem relationships."""
order_item = OrderItem(
order_id=test_order.id,
product_id=test_product.id,
product_name="Test Product",
product_sku="SKU001",
quantity=1,
unit_price=50.00,
total_price=50.00,
)
db.add(order_item)
db.commit()
db.refresh(order_item)
assert order_item.order is not None
assert order_item.order.id == test_order.id
def test_multiple_items_per_order(self, db, test_order, test_product):
"""Test multiple OrderItems can belong to same Order."""
# Create two order items for the same product (different quantities)
item1 = OrderItem(
order_id=test_order.id,
product_id=test_product.id,
product_name="Product - Size M",
product_sku="SKU001-M",
quantity=1,
unit_price=25.00,
total_price=25.00,
)
item2 = OrderItem(
order_id=test_order.id,
product_id=test_product.id,
product_name="Product - Size L",
product_sku="SKU001-L",
quantity=2,
unit_price=30.00,
total_price=60.00,
)
db.add_all([item1, item2])
db.commit()
assert item1.order_id == item2.order_id
assert item1.id != item2.id
assert item1.product_id == item2.product_id # Same product, different items

View File

@@ -0,0 +1,123 @@
# tests/unit/models/database/test_product.py
"""Unit tests for Product (vendor catalog) database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.product import Product
@pytest.mark.unit
@pytest.mark.database
class TestProductModel:
"""Test Product (vendor catalog) model."""
def test_product_creation(self, db, test_vendor, test_marketplace_product):
"""Test Product model linking vendor catalog to marketplace product."""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
product_id="VENDOR_PROD_001",
price=89.99,
currency="EUR",
availability="in stock",
is_featured=True,
is_active=True,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.id is not None
assert product.vendor_id == test_vendor.id
assert product.marketplace_product_id == test_marketplace_product.id
assert product.price == 89.99
assert product.is_featured is True
assert product.vendor.vendor_code == test_vendor.vendor_code
assert product.marketplace_product.title == test_marketplace_product.title
def test_product_unique_per_vendor(self, db, test_vendor, test_marketplace_product):
"""Test that same marketplace product can't be added twice to vendor catalog."""
product1 = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
is_active=True,
)
db.add(product1)
db.commit()
# Same marketplace product to same vendor should fail
with pytest.raises(IntegrityError):
product2 = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
is_active=True,
)
db.add(product2)
db.commit()
def test_product_default_values(self, db, test_vendor, test_marketplace_product):
"""Test Product model default values."""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.is_active is True # Default
assert product.is_featured is False # Default
assert product.min_quantity == 1 # Default
assert product.display_order == 0 # Default
def test_product_vendor_override_fields(self, db, test_vendor, test_marketplace_product):
"""Test Product model vendor-specific override fields."""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
product_id="CUSTOM_SKU_001",
price=49.99,
sale_price=39.99,
currency="USD",
availability="limited",
condition="new",
)
db.add(product)
db.commit()
db.refresh(product)
assert product.product_id == "CUSTOM_SKU_001"
assert product.price == 49.99
assert product.sale_price == 39.99
assert product.currency == "USD"
assert product.availability == "limited"
def test_product_inventory_settings(self, db, test_vendor, test_marketplace_product):
"""Test Product model inventory settings."""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
min_quantity=2,
max_quantity=10,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.min_quantity == 2
assert product.max_quantity == 10
def test_product_relationships(self, db, test_vendor, test_marketplace_product):
"""Test Product relationships."""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.vendor is not None
assert product.marketplace_product is not None
assert product.inventory_entries == [] # No inventory yet

View File

@@ -0,0 +1,179 @@
# tests/unit/models/database/test_team.py
"""Unit tests for VendorUser and Role database models."""
import pytest
from models.database.vendor import Role, Vendor, VendorUser
@pytest.mark.unit
@pytest.mark.database
class TestRoleModel:
"""Test Role model."""
def test_role_creation(self, db, test_vendor):
"""Test Role model creation."""
role = Role(
vendor_id=test_vendor.id,
name="Manager",
permissions=["products.create", "orders.view"],
)
db.add(role)
db.commit()
db.refresh(role)
assert role.id is not None
assert role.vendor_id == test_vendor.id
assert role.name == "Manager"
assert "products.create" in role.permissions
assert "orders.view" in role.permissions
def test_role_default_permissions(self, db, test_vendor):
"""Test Role model with default empty permissions."""
role = Role(
vendor_id=test_vendor.id,
name="Viewer",
)
db.add(role)
db.commit()
db.refresh(role)
assert role.permissions == [] or role.permissions is None
def test_role_vendor_relationship(self, db, test_vendor):
"""Test Role-Vendor relationship."""
role = Role(
vendor_id=test_vendor.id,
name="Admin",
permissions=["*"],
)
db.add(role)
db.commit()
db.refresh(role)
assert role.vendor is not None
assert role.vendor.id == test_vendor.id
@pytest.mark.unit
@pytest.mark.database
class TestVendorUserModel:
"""Test VendorUser model."""
def test_vendor_user_creation(self, db, test_vendor, test_user):
"""Test VendorUser model for team management."""
# Create a role
role = Role(
vendor_id=test_vendor.id,
name="Manager",
permissions=["products.create", "orders.view"],
)
db.add(role)
db.commit()
# Create vendor user
vendor_user = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(vendor_user)
db.commit()
db.refresh(vendor_user)
assert vendor_user.id is not None
assert vendor_user.vendor_id == test_vendor.id
assert vendor_user.user_id == test_user.id
assert vendor_user.role.name == "Manager"
assert "products.create" in vendor_user.role.permissions
def test_vendor_user_multiple_vendors(self, db, test_vendor, test_user, other_company):
"""Test same user can be added to multiple vendors."""
# Create another vendor
other_vendor = Vendor(
company_id=other_company.id,
vendor_code="OTHER_VENDOR",
subdomain="othervendor",
name="Other Vendor",
)
db.add(other_vendor)
db.commit()
role1 = Role(
vendor_id=test_vendor.id,
name="Editor1",
permissions=["products.view"],
)
role2 = Role(
vendor_id=other_vendor.id,
name="Editor2",
permissions=["products.view"],
)
db.add_all([role1, role2])
db.commit()
# Same user can be added to different vendors
vendor_user1 = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role1.id,
)
vendor_user2 = VendorUser(
vendor_id=other_vendor.id,
user_id=test_user.id,
role_id=role2.id,
)
db.add_all([vendor_user1, vendor_user2])
db.commit()
assert vendor_user1.vendor_id != vendor_user2.vendor_id
assert vendor_user1.user_id == vendor_user2.user_id
def test_vendor_user_relationships(self, db, test_vendor, test_user):
"""Test VendorUser relationships."""
role = Role(
vendor_id=test_vendor.id,
name="Staff",
permissions=["orders.view"],
)
db.add(role)
db.commit()
vendor_user = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(vendor_user)
db.commit()
db.refresh(vendor_user)
assert vendor_user.vendor is not None
assert vendor_user.user is not None
assert vendor_user.role is not None
assert vendor_user.vendor.vendor_code == test_vendor.vendor_code
assert vendor_user.user.email == test_user.email
def test_vendor_user_with_active_flag(self, db, test_vendor, test_user):
"""Test VendorUser is_active field."""
role = Role(
vendor_id=test_vendor.id,
name="Default",
permissions=[],
)
db.add(role)
db.commit()
# Create with explicit is_active=True
vendor_user = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(vendor_user)
db.commit()
db.refresh(vendor_user)
assert vendor_user.is_active is True

View File

@@ -0,0 +1,104 @@
# tests/unit/models/database/test_user.py
"""Unit tests for User database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.user import User
@pytest.mark.unit
@pytest.mark.database
class TestUserModel:
"""Test User model."""
def test_user_creation(self, db):
"""Test User model creation and relationships."""
user = User(
email="db_test@example.com",
username="dbtest",
hashed_password="hashed_password_123",
role="user",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
assert user.id is not None
assert user.email == "db_test@example.com"
assert user.username == "dbtest"
assert user.role == "user"
assert user.is_active is True
assert user.created_at is not None
assert user.updated_at is not None
def test_user_email_uniqueness(self, db):
"""Test email unique constraint."""
user1 = User(
email="unique@example.com",
username="user1",
hashed_password="hash1",
)
db.add(user1)
db.commit()
# Duplicate email should raise error
with pytest.raises(IntegrityError):
user2 = User(
email="unique@example.com",
username="user2",
hashed_password="hash2",
)
db.add(user2)
db.commit()
def test_user_username_uniqueness(self, db):
"""Test username unique constraint."""
user1 = User(
email="user1@example.com",
username="sameusername",
hashed_password="hash1",
)
db.add(user1)
db.commit()
# Duplicate username should raise error
with pytest.raises(IntegrityError):
user2 = User(
email="user2@example.com",
username="sameusername",
hashed_password="hash2",
)
db.add(user2)
db.commit()
def test_user_default_values(self, db):
"""Test User model default values."""
user = User(
email="defaults@example.com",
username="defaultuser",
hashed_password="hash",
)
db.add(user)
db.commit()
db.refresh(user)
assert user.is_active is True # Default
assert user.role == "vendor" # Default (UserRole.VENDOR)
def test_user_optional_fields(self, db):
"""Test User model with optional fields."""
user = User(
email="optional@example.com",
username="optionaluser",
hashed_password="hash",
first_name="John",
last_name="Doe",
)
db.add(user)
db.commit()
db.refresh(user)
assert user.first_name == "John"
assert user.last_name == "Doe"

View File

@@ -0,0 +1,137 @@
# tests/unit/models/database/test_vendor.py
"""Unit tests for Vendor database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.vendor import Vendor
@pytest.mark.unit
@pytest.mark.database
class TestVendorModel:
"""Test Vendor model."""
def test_vendor_creation(self, db, test_company):
"""Test Vendor model creation with company relationship."""
vendor = Vendor(
company_id=test_company.id,
vendor_code="DBTEST",
subdomain="dbtest",
name="Database Test Vendor",
description="Testing vendor model",
contact_email="contact@dbtest.com",
contact_phone="+1234567890",
business_address="123 Test Street",
is_active=True,
is_verified=False,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
assert vendor.id is not None
assert vendor.vendor_code == "DBTEST"
assert vendor.subdomain == "dbtest"
assert vendor.name == "Database Test Vendor"
assert vendor.company_id == test_company.id
assert vendor.contact_email == "contact@dbtest.com"
assert vendor.is_active is True
assert vendor.is_verified is False
assert vendor.created_at is not None
def test_vendor_with_letzshop_urls(self, db, test_company):
"""Test Vendor model with multi-language Letzshop URLs."""
vendor = Vendor(
company_id=test_company.id,
vendor_code="MULTILANG",
subdomain="multilang",
name="Multi-Language Vendor",
letzshop_csv_url_fr="https://example.com/feed_fr.csv",
letzshop_csv_url_en="https://example.com/feed_en.csv",
letzshop_csv_url_de="https://example.com/feed_de.csv",
is_active=True,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
assert vendor.letzshop_csv_url_fr == "https://example.com/feed_fr.csv"
assert vendor.letzshop_csv_url_en == "https://example.com/feed_en.csv"
assert vendor.letzshop_csv_url_de == "https://example.com/feed_de.csv"
def test_vendor_code_uniqueness(self, db, test_company):
"""Test vendor_code unique constraint."""
vendor1 = Vendor(
company_id=test_company.id,
vendor_code="UNIQUE",
subdomain="unique1",
name="Unique Vendor 1",
)
db.add(vendor1)
db.commit()
# Duplicate vendor_code should raise error
with pytest.raises(IntegrityError):
vendor2 = Vendor(
company_id=test_company.id,
vendor_code="UNIQUE",
subdomain="unique2",
name="Unique Vendor 2",
)
db.add(vendor2)
db.commit()
def test_subdomain_uniqueness(self, db, test_company):
"""Test subdomain unique constraint."""
vendor1 = Vendor(
company_id=test_company.id,
vendor_code="VENDOR1",
subdomain="testsubdomain",
name="Vendor 1",
)
db.add(vendor1)
db.commit()
# Duplicate subdomain should raise error
with pytest.raises(IntegrityError):
vendor2 = Vendor(
company_id=test_company.id,
vendor_code="VENDOR2",
subdomain="testsubdomain",
name="Vendor 2",
)
db.add(vendor2)
db.commit()
def test_vendor_default_values(self, db, test_company):
"""Test Vendor model default values."""
vendor = Vendor(
company_id=test_company.id,
vendor_code="DEFAULTS",
subdomain="defaults",
name="Default Vendor",
)
db.add(vendor)
db.commit()
db.refresh(vendor)
assert vendor.is_active is True # Default
assert vendor.is_verified is False # Default
def test_vendor_company_relationship(self, db, test_company):
"""Test Vendor-Company relationship."""
vendor = Vendor(
company_id=test_company.id,
vendor_code="RELTEST",
subdomain="reltest",
name="Relationship Test Vendor",
)
db.add(vendor)
db.commit()
db.refresh(vendor)
assert vendor.company is not None
assert vendor.company.id == test_company.id
assert vendor.company.name == test_company.name

View File

@@ -0,0 +1,2 @@
# tests/unit/models/schema/__init__.py
"""Pydantic schema unit tests."""

View File

@@ -0,0 +1,275 @@
# tests/unit/models/schema/test_auth.py
"""Unit tests for auth Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.auth import (
UserCreate,
UserLogin,
UserRegister,
UserResponse,
UserUpdate,
)
@pytest.mark.unit
@pytest.mark.schema
class TestUserRegisterSchema:
"""Test UserRegister schema validation."""
def test_valid_registration(self):
"""Test valid registration data."""
user = UserRegister(
email="test@example.com",
username="testuser",
password="password123",
)
assert user.email == "test@example.com"
assert user.username == "testuser"
assert user.password == "password123"
def test_username_normalized_to_lowercase(self):
"""Test username is normalized to lowercase."""
user = UserRegister(
email="test@example.com",
username="TestUser",
password="password123",
)
assert user.username == "testuser"
def test_username_with_whitespace_invalid(self):
"""Test username with whitespace is invalid (validation before strip)."""
with pytest.raises(ValidationError) as exc_info:
UserRegister(
email="test@example.com",
username=" testuser ",
password="password123",
)
assert "username" in str(exc_info.value).lower()
def test_invalid_email(self):
"""Test invalid email raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
UserRegister(
email="not-an-email",
username="testuser",
password="password123",
)
assert "email" in str(exc_info.value).lower()
def test_invalid_username_special_chars(self):
"""Test username with special characters raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
UserRegister(
email="test@example.com",
username="test@user!",
password="password123",
)
assert "username" in str(exc_info.value).lower()
def test_valid_username_with_underscore(self):
"""Test username with underscore is valid."""
user = UserRegister(
email="test@example.com",
username="test_user_123",
password="password123",
)
assert user.username == "test_user_123"
def test_password_too_short(self):
"""Test password shorter than 6 characters raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
UserRegister(
email="test@example.com",
username="testuser",
password="12345",
)
assert "password" in str(exc_info.value).lower()
def test_password_exactly_6_chars(self):
"""Test password with exactly 6 characters is valid."""
user = UserRegister(
email="test@example.com",
username="testuser",
password="123456",
)
assert user.password == "123456"
def test_missing_required_fields(self):
"""Test missing required fields raises ValidationError."""
with pytest.raises(ValidationError):
UserRegister(email="test@example.com")
@pytest.mark.unit
@pytest.mark.schema
class TestUserLoginSchema:
"""Test UserLogin schema validation."""
def test_valid_login(self):
"""Test valid login data."""
login = UserLogin(
email_or_username="testuser",
password="password123",
)
assert login.email_or_username == "testuser"
assert login.password == "password123"
def test_login_with_email(self):
"""Test login with email."""
login = UserLogin(
email_or_username="test@example.com",
password="password123",
)
assert login.email_or_username == "test@example.com"
def test_login_with_vendor_code(self):
"""Test login with optional vendor code."""
login = UserLogin(
email_or_username="testuser",
password="password123",
vendor_code="VENDOR001",
)
assert login.vendor_code == "VENDOR001"
def test_email_or_username_stripped(self):
"""Test email_or_username is stripped of whitespace."""
login = UserLogin(
email_or_username=" testuser ",
password="password123",
)
assert login.email_or_username == "testuser"
@pytest.mark.unit
@pytest.mark.schema
class TestUserCreateSchema:
"""Test UserCreate schema validation."""
def test_valid_user_create(self):
"""Test valid user creation data."""
user = UserCreate(
email="admin@example.com",
username="adminuser",
password="securepass",
first_name="Admin",
last_name="User",
role="admin",
)
assert user.email == "admin@example.com"
assert user.role == "admin"
def test_default_role_is_vendor(self):
"""Test default role is vendor."""
user = UserCreate(
email="vendor@example.com",
username="vendoruser",
password="securepass",
)
assert user.role == "vendor"
def test_invalid_role(self):
"""Test invalid role raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
UserCreate(
email="test@example.com",
username="testuser",
password="securepass",
role="superadmin",
)
assert "role" in str(exc_info.value).lower()
def test_username_too_short(self):
"""Test username shorter than 3 characters raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
UserCreate(
email="test@example.com",
username="ab",
password="securepass",
)
assert "username" in str(exc_info.value).lower()
def test_password_min_length(self):
"""Test password minimum length validation."""
with pytest.raises(ValidationError) as exc_info:
UserCreate(
email="test@example.com",
username="testuser",
password="12345",
)
assert "password" in str(exc_info.value).lower()
@pytest.mark.unit
@pytest.mark.schema
class TestUserUpdateSchema:
"""Test UserUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = UserUpdate(first_name="NewName")
assert update.first_name == "NewName"
assert update.username is None
assert update.email is None
def test_username_update_normalized(self):
"""Test username in update is normalized."""
update = UserUpdate(username="NewUser")
assert update.username == "newuser"
def test_invalid_role_update(self):
"""Test invalid role in update raises ValidationError."""
with pytest.raises(ValidationError):
UserUpdate(role="superadmin")
def test_valid_role_update(self):
"""Test valid role values."""
admin_update = UserUpdate(role="admin")
vendor_update = UserUpdate(role="vendor")
assert admin_update.role == "admin"
assert vendor_update.role == "vendor"
def test_empty_update(self):
"""Test empty update is valid (all fields optional)."""
update = UserUpdate()
assert update.model_dump(exclude_unset=True) == {}
@pytest.mark.unit
@pytest.mark.schema
class TestUserResponseSchema:
"""Test UserResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"email": "test@example.com",
"username": "testuser",
"role": "vendor",
"is_active": True,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = UserResponse(**data)
assert response.id == 1
assert response.email == "test@example.com"
assert response.is_active is True
def test_optional_last_login(self):
"""Test last_login is optional."""
from datetime import datetime
data = {
"id": 1,
"email": "test@example.com",
"username": "testuser",
"role": "vendor",
"is_active": True,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = UserResponse(**data)
assert response.last_login is None

View File

@@ -0,0 +1,364 @@
# tests/unit/models/schema/test_customer.py
"""Unit tests for customer Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.customer import (
CustomerRegister,
CustomerUpdate,
CustomerResponse,
CustomerAddressCreate,
CustomerAddressUpdate,
CustomerAddressResponse,
CustomerPreferencesUpdate,
)
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerRegisterSchema:
"""Test CustomerRegister schema validation."""
def test_valid_registration(self):
"""Test valid registration data."""
customer = CustomerRegister(
email="customer@example.com",
password="Password123",
first_name="John",
last_name="Doe",
)
assert customer.email == "customer@example.com"
assert customer.first_name == "John"
assert customer.last_name == "Doe"
def test_email_normalized_to_lowercase(self):
"""Test email is normalized to lowercase."""
customer = CustomerRegister(
email="Customer@Example.COM",
password="Password123",
first_name="John",
last_name="Doe",
)
assert customer.email == "customer@example.com"
def test_invalid_email(self):
"""Test invalid email raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
CustomerRegister(
email="not-an-email",
password="Password123",
first_name="John",
last_name="Doe",
)
assert "email" in str(exc_info.value).lower()
def test_password_min_length(self):
"""Test password must be at least 8 characters."""
with pytest.raises(ValidationError) as exc_info:
CustomerRegister(
email="customer@example.com",
password="Pass1",
first_name="John",
last_name="Doe",
)
assert "password" in str(exc_info.value).lower()
def test_password_requires_digit(self):
"""Test password must contain at least one digit."""
with pytest.raises(ValidationError) as exc_info:
CustomerRegister(
email="customer@example.com",
password="Password",
first_name="John",
last_name="Doe",
)
assert "digit" in str(exc_info.value).lower()
def test_password_requires_letter(self):
"""Test password must contain at least one letter."""
with pytest.raises(ValidationError) as exc_info:
CustomerRegister(
email="customer@example.com",
password="12345678",
first_name="John",
last_name="Doe",
)
assert "letter" in str(exc_info.value).lower()
def test_first_name_required(self):
"""Test first_name is required."""
with pytest.raises(ValidationError) as exc_info:
CustomerRegister(
email="customer@example.com",
password="Password123",
last_name="Doe",
)
assert "first_name" in str(exc_info.value).lower()
def test_marketing_consent_default(self):
"""Test marketing_consent defaults to False."""
customer = CustomerRegister(
email="customer@example.com",
password="Password123",
first_name="John",
last_name="Doe",
)
assert customer.marketing_consent is False
def test_optional_phone(self):
"""Test optional phone field."""
customer = CustomerRegister(
email="customer@example.com",
password="Password123",
first_name="John",
last_name="Doe",
phone="+352 123 456",
)
assert customer.phone == "+352 123 456"
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerUpdateSchema:
"""Test CustomerUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = CustomerUpdate(first_name="Jane")
assert update.first_name == "Jane"
assert update.last_name is None
assert update.email is None
def test_empty_update_is_valid(self):
"""Test empty update is valid."""
update = CustomerUpdate()
assert update.model_dump(exclude_unset=True) == {}
def test_email_normalized_to_lowercase(self):
"""Test email is normalized to lowercase."""
update = CustomerUpdate(email="NewEmail@Example.COM")
assert update.email == "newemail@example.com"
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerResponseSchema:
"""Test CustomerResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
from decimal import Decimal
data = {
"id": 1,
"vendor_id": 1,
"email": "customer@example.com",
"first_name": "John",
"last_name": "Doe",
"phone": None,
"customer_number": "CUST001",
"marketing_consent": False,
"last_order_date": None,
"total_orders": 5,
"total_spent": Decimal("500.00"),
"is_active": True,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = CustomerResponse(**data)
assert response.id == 1
assert response.customer_number == "CUST001"
assert response.total_orders == 5
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerAddressCreateSchema:
"""Test CustomerAddressCreate schema validation."""
def test_valid_shipping_address(self):
"""Test valid shipping address creation."""
address = CustomerAddressCreate(
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.address_type == "shipping"
assert address.city == "Luxembourg"
def test_valid_billing_address(self):
"""Test valid billing address creation."""
address = CustomerAddressCreate(
address_type="billing",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.address_type == "billing"
def test_invalid_address_type(self):
"""Test invalid address_type raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
CustomerAddressCreate(
address_type="delivery",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert "address_type" in str(exc_info.value).lower()
def test_is_default_defaults_to_false(self):
"""Test is_default defaults to False."""
address = CustomerAddressCreate(
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.is_default is False
def test_optional_company(self):
"""Test optional company field."""
address = CustomerAddressCreate(
address_type="shipping",
first_name="John",
last_name="Doe",
company="Tech Corp",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.company == "Tech Corp"
def test_optional_address_line_2(self):
"""Test optional address_line_2 field."""
address = CustomerAddressCreate(
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
address_line_2="Apt 4B",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.address_line_2 == "Apt 4B"
def test_required_fields(self):
"""Test required fields."""
with pytest.raises(ValidationError):
CustomerAddressCreate(
address_type="shipping",
first_name="John",
# missing last_name, address_line_1, city, postal_code, country
)
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerAddressUpdateSchema:
"""Test CustomerAddressUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = CustomerAddressUpdate(city="Esch-sur-Alzette")
assert update.city == "Esch-sur-Alzette"
assert update.first_name is None
def test_empty_update_is_valid(self):
"""Test empty update is valid."""
update = CustomerAddressUpdate()
assert update.model_dump(exclude_unset=True) == {}
def test_address_type_validation(self):
"""Test address_type validation in update."""
with pytest.raises(ValidationError):
CustomerAddressUpdate(address_type="invalid")
def test_valid_address_type_update(self):
"""Test valid address_type values in update."""
shipping = CustomerAddressUpdate(address_type="shipping")
billing = CustomerAddressUpdate(address_type="billing")
assert shipping.address_type == "shipping"
assert billing.address_type == "billing"
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerAddressResponseSchema:
"""Test CustomerAddressResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"vendor_id": 1,
"customer_id": 1,
"address_type": "shipping",
"first_name": "John",
"last_name": "Doe",
"company": None,
"address_line_1": "123 Main St",
"address_line_2": None,
"city": "Luxembourg",
"postal_code": "L-1234",
"country": "Luxembourg",
"is_default": True,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = CustomerAddressResponse(**data)
assert response.id == 1
assert response.is_default is True
assert response.address_type == "shipping"
@pytest.mark.unit
@pytest.mark.schema
class TestCustomerPreferencesUpdateSchema:
"""Test CustomerPreferencesUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = CustomerPreferencesUpdate(marketing_consent=True)
assert update.marketing_consent is True
assert update.language is None
def test_language_update(self):
"""Test language preference update."""
update = CustomerPreferencesUpdate(language="fr")
assert update.language == "fr"
def test_currency_update(self):
"""Test currency preference update."""
update = CustomerPreferencesUpdate(currency="EUR")
assert update.currency == "EUR"
def test_notification_preferences(self):
"""Test notification preferences dict."""
update = CustomerPreferencesUpdate(
notification_preferences={
"email": True,
"sms": False,
"push": True,
}
)
assert update.notification_preferences["email"] is True
assert update.notification_preferences["sms"] is False

View File

@@ -0,0 +1,299 @@
# tests/unit/models/schema/test_inventory.py
"""Unit tests for inventory Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.inventory import (
InventoryBase,
InventoryCreate,
InventoryAdjust,
InventoryUpdate,
InventoryReserve,
InventoryResponse,
InventoryLocationResponse,
ProductInventorySummary,
)
@pytest.mark.unit
@pytest.mark.schema
class TestInventoryCreateSchema:
"""Test InventoryCreate schema validation."""
def test_valid_inventory_create(self):
"""Test valid inventory creation data."""
inventory = InventoryCreate(
product_id=1,
location="Warehouse A",
quantity=100,
)
assert inventory.product_id == 1
assert inventory.location == "Warehouse A"
assert inventory.quantity == 100
def test_product_id_required(self):
"""Test product_id is required."""
with pytest.raises(ValidationError) as exc_info:
InventoryCreate(location="Warehouse A", quantity=10)
assert "product_id" in str(exc_info.value).lower()
def test_location_required(self):
"""Test location is required."""
with pytest.raises(ValidationError) as exc_info:
InventoryCreate(product_id=1, quantity=10)
assert "location" in str(exc_info.value).lower()
def test_quantity_required(self):
"""Test quantity is required."""
with pytest.raises(ValidationError) as exc_info:
InventoryCreate(product_id=1, location="Warehouse A")
assert "quantity" in str(exc_info.value).lower()
def test_quantity_must_be_non_negative(self):
"""Test quantity must be >= 0."""
with pytest.raises(ValidationError) as exc_info:
InventoryCreate(
product_id=1,
location="Warehouse A",
quantity=-5,
)
assert "quantity" in str(exc_info.value).lower()
def test_quantity_zero_is_valid(self):
"""Test quantity of 0 is valid."""
inventory = InventoryCreate(
product_id=1,
location="Warehouse A",
quantity=0,
)
assert inventory.quantity == 0
@pytest.mark.unit
@pytest.mark.schema
class TestInventoryAdjustSchema:
"""Test InventoryAdjust schema validation."""
def test_positive_adjustment(self):
"""Test positive quantity adjustment (add stock)."""
adjustment = InventoryAdjust(
product_id=1,
location="Warehouse A",
quantity=50,
)
assert adjustment.quantity == 50
def test_negative_adjustment(self):
"""Test negative quantity adjustment (remove stock)."""
adjustment = InventoryAdjust(
product_id=1,
location="Warehouse A",
quantity=-20,
)
assert adjustment.quantity == -20
def test_zero_adjustment_is_valid(self):
"""Test zero adjustment is technically valid."""
adjustment = InventoryAdjust(
product_id=1,
location="Warehouse A",
quantity=0,
)
assert adjustment.quantity == 0
@pytest.mark.unit
@pytest.mark.schema
class TestInventoryUpdateSchema:
"""Test InventoryUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = InventoryUpdate(quantity=150)
assert update.quantity == 150
assert update.reserved_quantity is None
assert update.location is None
def test_empty_update_is_valid(self):
"""Test empty update is valid."""
update = InventoryUpdate()
assert update.model_dump(exclude_unset=True) == {}
def test_quantity_must_be_non_negative(self):
"""Test quantity must be >= 0 in update."""
with pytest.raises(ValidationError):
InventoryUpdate(quantity=-10)
def test_reserved_quantity_must_be_non_negative(self):
"""Test reserved_quantity must be >= 0."""
with pytest.raises(ValidationError):
InventoryUpdate(reserved_quantity=-5)
def test_location_update(self):
"""Test location can be updated."""
update = InventoryUpdate(location="Warehouse B")
assert update.location == "Warehouse B"
@pytest.mark.unit
@pytest.mark.schema
class TestInventoryReserveSchema:
"""Test InventoryReserve schema validation."""
def test_valid_reservation(self):
"""Test valid inventory reservation."""
reservation = InventoryReserve(
product_id=1,
location="Warehouse A",
quantity=10,
)
assert reservation.product_id == 1
assert reservation.quantity == 10
def test_quantity_must_be_positive(self):
"""Test reservation quantity must be > 0."""
with pytest.raises(ValidationError) as exc_info:
InventoryReserve(
product_id=1,
location="Warehouse A",
quantity=0,
)
assert "quantity" in str(exc_info.value).lower()
def test_negative_quantity_invalid(self):
"""Test negative reservation quantity is invalid."""
with pytest.raises(ValidationError):
InventoryReserve(
product_id=1,
location="Warehouse A",
quantity=-5,
)
@pytest.mark.unit
@pytest.mark.schema
class TestInventoryResponseSchema:
"""Test InventoryResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"product_id": 1,
"vendor_id": 1,
"location": "Warehouse A",
"quantity": 100,
"reserved_quantity": 20,
"gtin": "1234567890123",
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = InventoryResponse(**data)
assert response.id == 1
assert response.quantity == 100
assert response.reserved_quantity == 20
def test_available_quantity_property(self):
"""Test available_quantity calculated property."""
from datetime import datetime
data = {
"id": 1,
"product_id": 1,
"vendor_id": 1,
"location": "Warehouse A",
"quantity": 100,
"reserved_quantity": 30,
"gtin": None,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = InventoryResponse(**data)
assert response.available_quantity == 70
def test_available_quantity_never_negative(self):
"""Test available_quantity is never negative."""
from datetime import datetime
data = {
"id": 1,
"product_id": 1,
"vendor_id": 1,
"location": "Warehouse A",
"quantity": 10,
"reserved_quantity": 50, # Over-reserved
"gtin": None,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = InventoryResponse(**data)
assert response.available_quantity == 0
@pytest.mark.unit
@pytest.mark.schema
class TestInventoryLocationResponseSchema:
"""Test InventoryLocationResponse schema."""
def test_valid_location_response(self):
"""Test valid location response."""
location = InventoryLocationResponse(
location="Warehouse A",
quantity=100,
reserved_quantity=20,
available_quantity=80,
)
assert location.location == "Warehouse A"
assert location.quantity == 100
assert location.available_quantity == 80
@pytest.mark.unit
@pytest.mark.schema
class TestProductInventorySummarySchema:
"""Test ProductInventorySummary schema."""
def test_valid_summary(self):
"""Test valid inventory summary."""
summary = ProductInventorySummary(
product_id=1,
vendor_id=1,
product_sku="SKU-001",
product_title="Test Product",
total_quantity=200,
total_reserved=50,
total_available=150,
locations=[
InventoryLocationResponse(
location="Warehouse A",
quantity=100,
reserved_quantity=25,
available_quantity=75,
),
InventoryLocationResponse(
location="Warehouse B",
quantity=100,
reserved_quantity=25,
available_quantity=75,
),
],
)
assert summary.product_id == 1
assert summary.total_quantity == 200
assert len(summary.locations) == 2
def test_empty_locations(self):
"""Test summary with no locations."""
summary = ProductInventorySummary(
product_id=1,
vendor_id=1,
product_sku=None,
product_title="Test Product",
total_quantity=0,
total_reserved=0,
total_available=0,
locations=[],
)
assert summary.locations == []

View File

@@ -0,0 +1,239 @@
# tests/unit/models/schema/test_marketplace_import_job.py
"""Unit tests for marketplace import job Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.marketplace_import_job import (
MarketplaceImportJobRequest,
MarketplaceImportJobResponse,
MarketplaceImportJobListResponse,
MarketplaceImportJobStatusUpdate,
)
@pytest.mark.unit
@pytest.mark.schema
class TestMarketplaceImportJobRequestSchema:
"""Test MarketplaceImportJobRequest schema validation."""
def test_valid_request(self):
"""Test valid import job request."""
request = MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
marketplace="Letzshop",
)
assert request.source_url == "https://example.com/products.csv"
assert request.marketplace == "Letzshop"
def test_source_url_required(self):
"""Test source_url is required."""
with pytest.raises(ValidationError) as exc_info:
MarketplaceImportJobRequest(marketplace="Letzshop")
assert "source_url" in str(exc_info.value).lower()
def test_source_url_must_be_http_or_https(self):
"""Test source_url must start with http:// or https://."""
with pytest.raises(ValidationError) as exc_info:
MarketplaceImportJobRequest(
source_url="ftp://example.com/products.csv",
)
assert "url" in str(exc_info.value).lower()
def test_source_url_http_is_valid(self):
"""Test http:// URLs are valid."""
request = MarketplaceImportJobRequest(
source_url="http://example.com/products.csv",
)
assert request.source_url == "http://example.com/products.csv"
def test_source_url_with_leading_whitespace_invalid(self):
"""Test source_url with leading whitespace is invalid (validation before strip)."""
with pytest.raises(ValidationError) as exc_info:
MarketplaceImportJobRequest(
source_url=" https://example.com/products.csv",
)
assert "url" in str(exc_info.value).lower()
def test_marketplace_default(self):
"""Test marketplace defaults to Letzshop."""
request = MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
)
assert request.marketplace == "Letzshop"
def test_marketplace_stripped(self):
"""Test marketplace is stripped of whitespace."""
request = MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
marketplace=" CustomMarket ",
)
assert request.marketplace == "CustomMarket"
def test_batch_size_default(self):
"""Test batch_size defaults to 1000."""
request = MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
)
assert request.batch_size == 1000
def test_batch_size_minimum(self):
"""Test batch_size must be >= 100."""
with pytest.raises(ValidationError) as exc_info:
MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
batch_size=50,
)
assert "batch_size" in str(exc_info.value).lower()
def test_batch_size_maximum(self):
"""Test batch_size must be <= 10000."""
with pytest.raises(ValidationError) as exc_info:
MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
batch_size=15000,
)
assert "batch_size" in str(exc_info.value).lower()
def test_batch_size_valid_range(self):
"""Test batch_size in valid range."""
request = MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
batch_size=5000,
)
assert request.batch_size == 5000
@pytest.mark.unit
@pytest.mark.schema
class TestMarketplaceImportJobResponseSchema:
"""Test MarketplaceImportJobResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"job_id": 1,
"vendor_id": 1,
"vendor_code": "TEST_VENDOR",
"vendor_name": "Test Vendor",
"marketplace": "Letzshop",
"source_url": "https://example.com/products.csv",
"status": "pending",
"created_at": datetime.now(),
}
response = MarketplaceImportJobResponse(**data)
assert response.job_id == 1
assert response.vendor_code == "TEST_VENDOR"
assert response.status == "pending"
def test_default_counts(self):
"""Test count fields default to 0."""
from datetime import datetime
data = {
"job_id": 1,
"vendor_id": 1,
"vendor_code": "TEST_VENDOR",
"vendor_name": "Test Vendor",
"marketplace": "Letzshop",
"source_url": "https://example.com/products.csv",
"status": "completed",
"created_at": datetime.now(),
}
response = MarketplaceImportJobResponse(**data)
assert response.imported == 0
assert response.updated == 0
assert response.total_processed == 0
assert response.error_count == 0
def test_optional_timestamps(self):
"""Test optional timestamp fields."""
from datetime import datetime
now = datetime.now()
data = {
"job_id": 1,
"vendor_id": 1,
"vendor_code": "TEST_VENDOR",
"vendor_name": "Test Vendor",
"marketplace": "Letzshop",
"source_url": "https://example.com/products.csv",
"status": "completed",
"created_at": now,
"started_at": now,
"completed_at": now,
}
response = MarketplaceImportJobResponse(**data)
assert response.started_at == now
assert response.completed_at == now
def test_error_message_optional(self):
"""Test error_message is optional."""
from datetime import datetime
data = {
"job_id": 1,
"vendor_id": 1,
"vendor_code": "TEST_VENDOR",
"vendor_name": "Test Vendor",
"marketplace": "Letzshop",
"source_url": "https://example.com/products.csv",
"status": "failed",
"created_at": datetime.now(),
"error_message": "Connection timeout",
}
response = MarketplaceImportJobResponse(**data)
assert response.error_message == "Connection timeout"
@pytest.mark.unit
@pytest.mark.schema
class TestMarketplaceImportJobListResponseSchema:
"""Test MarketplaceImportJobListResponse schema."""
def test_valid_list_response(self):
"""Test valid list response structure."""
response = MarketplaceImportJobListResponse(
jobs=[],
total=0,
skip=0,
limit=10,
)
assert response.jobs == []
assert response.total == 0
@pytest.mark.unit
@pytest.mark.schema
class TestMarketplaceImportJobStatusUpdateSchema:
"""Test MarketplaceImportJobStatusUpdate schema."""
def test_status_only_update(self):
"""Test update with only status."""
update = MarketplaceImportJobStatusUpdate(status="processing")
assert update.status == "processing"
assert update.imported_count is None
def test_full_update(self):
"""Test update with all fields."""
update = MarketplaceImportJobStatusUpdate(
status="completed",
imported_count=100,
updated_count=50,
error_count=2,
total_processed=152,
)
assert update.imported_count == 100
assert update.updated_count == 50
assert update.error_count == 2
assert update.total_processed == 152
def test_error_update(self):
"""Test update with error message."""
update = MarketplaceImportJobStatusUpdate(
status="failed",
error_message="File not found",
)
assert update.status == "failed"
assert update.error_message == "File not found"

View File

@@ -0,0 +1,366 @@
# tests/unit/models/schema/test_order.py
"""Unit tests for order Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.order import (
OrderItemCreate,
OrderItemResponse,
OrderAddressCreate,
OrderAddressResponse,
OrderCreate,
OrderUpdate,
OrderResponse,
OrderDetailResponse,
OrderListResponse,
)
@pytest.mark.unit
@pytest.mark.schema
class TestOrderItemCreateSchema:
"""Test OrderItemCreate schema validation."""
def test_valid_order_item(self):
"""Test valid order item creation."""
item = OrderItemCreate(
product_id=1,
quantity=2,
)
assert item.product_id == 1
assert item.quantity == 2
def test_product_id_required(self):
"""Test product_id is required."""
with pytest.raises(ValidationError) as exc_info:
OrderItemCreate(quantity=2)
assert "product_id" in str(exc_info.value).lower()
def test_quantity_required(self):
"""Test quantity is required."""
with pytest.raises(ValidationError) as exc_info:
OrderItemCreate(product_id=1)
assert "quantity" in str(exc_info.value).lower()
def test_quantity_must_be_at_least_1(self):
"""Test quantity must be >= 1."""
with pytest.raises(ValidationError) as exc_info:
OrderItemCreate(
product_id=1,
quantity=0,
)
assert "quantity" in str(exc_info.value).lower()
def test_negative_quantity_invalid(self):
"""Test negative quantity is invalid."""
with pytest.raises(ValidationError):
OrderItemCreate(
product_id=1,
quantity=-1,
)
@pytest.mark.unit
@pytest.mark.schema
class TestOrderAddressCreateSchema:
"""Test OrderAddressCreate schema validation."""
def test_valid_address(self):
"""Test valid order address creation."""
address = OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.first_name == "John"
assert address.city == "Luxembourg"
def test_required_fields(self):
"""Test required fields validation."""
with pytest.raises(ValidationError):
OrderAddressCreate(
first_name="John",
# missing required fields
)
def test_optional_company(self):
"""Test optional company field."""
address = OrderAddressCreate(
first_name="John",
last_name="Doe",
company="Tech Corp",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.company == "Tech Corp"
def test_optional_address_line_2(self):
"""Test optional address_line_2 field."""
address = OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
address_line_2="Suite 500",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
assert address.address_line_2 == "Suite 500"
def test_first_name_min_length(self):
"""Test first_name minimum length."""
with pytest.raises(ValidationError):
OrderAddressCreate(
first_name="",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
)
def test_country_min_length(self):
"""Test country minimum length (2)."""
with pytest.raises(ValidationError):
OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="L",
)
@pytest.mark.unit
@pytest.mark.schema
class TestOrderCreateSchema:
"""Test OrderCreate schema validation."""
def test_valid_order(self):
"""Test valid order creation."""
order = OrderCreate(
items=[
OrderItemCreate(product_id=1, quantity=2),
OrderItemCreate(product_id=2, quantity=1),
],
shipping_address=OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
),
)
assert len(order.items) == 2
assert order.customer_id is None # Optional for guest checkout
def test_items_required(self):
"""Test items are required."""
with pytest.raises(ValidationError) as exc_info:
OrderCreate(
shipping_address=OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
),
)
assert "items" in str(exc_info.value).lower()
def test_items_must_not_be_empty(self):
"""Test items list must have at least 1 item."""
with pytest.raises(ValidationError) as exc_info:
OrderCreate(
items=[],
shipping_address=OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
),
)
assert "items" in str(exc_info.value).lower()
def test_shipping_address_required(self):
"""Test shipping_address is required."""
with pytest.raises(ValidationError) as exc_info:
OrderCreate(
items=[OrderItemCreate(product_id=1, quantity=1)],
)
assert "shipping_address" in str(exc_info.value).lower()
def test_optional_billing_address(self):
"""Test billing_address is optional."""
order = OrderCreate(
items=[OrderItemCreate(product_id=1, quantity=1)],
shipping_address=OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
),
billing_address=OrderAddressCreate(
first_name="Jane",
last_name="Doe",
address_line_1="456 Other St",
city="Esch",
postal_code="L-4321",
country="Luxembourg",
),
)
assert order.billing_address is not None
assert order.billing_address.first_name == "Jane"
def test_optional_customer_notes(self):
"""Test optional customer_notes."""
order = OrderCreate(
items=[OrderItemCreate(product_id=1, quantity=1)],
shipping_address=OrderAddressCreate(
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
),
customer_notes="Please leave at door",
)
assert order.customer_notes == "Please leave at door"
@pytest.mark.unit
@pytest.mark.schema
class TestOrderUpdateSchema:
"""Test OrderUpdate schema validation."""
def test_status_update(self):
"""Test valid status update."""
update = OrderUpdate(status="processing")
assert update.status == "processing"
def test_valid_status_values(self):
"""Test all valid status values."""
valid_statuses = ["pending", "processing", "shipped", "delivered", "cancelled", "refunded"]
for status in valid_statuses:
update = OrderUpdate(status=status)
assert update.status == status
def test_invalid_status(self):
"""Test invalid status raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
OrderUpdate(status="invalid_status")
assert "status" in str(exc_info.value).lower()
def test_tracking_number_update(self):
"""Test tracking number update."""
update = OrderUpdate(tracking_number="TRACK123456")
assert update.tracking_number == "TRACK123456"
def test_internal_notes_update(self):
"""Test internal notes update."""
update = OrderUpdate(internal_notes="Customer requested expedited shipping")
assert update.internal_notes == "Customer requested expedited shipping"
def test_empty_update_is_valid(self):
"""Test empty update is valid."""
update = OrderUpdate()
assert update.model_dump(exclude_unset=True) == {}
@pytest.mark.unit
@pytest.mark.schema
class TestOrderResponseSchema:
"""Test OrderResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"vendor_id": 1,
"customer_id": 1,
"order_number": "ORD-001",
"status": "pending",
"subtotal": 100.00,
"tax_amount": 20.00,
"shipping_amount": 10.00,
"discount_amount": 5.00,
"total_amount": 125.00,
"currency": "EUR",
"shipping_method": "standard",
"tracking_number": None,
"customer_notes": None,
"internal_notes": None,
"created_at": datetime.now(),
"updated_at": datetime.now(),
"paid_at": None,
"shipped_at": None,
"delivered_at": None,
"cancelled_at": None,
}
response = OrderResponse(**data)
assert response.id == 1
assert response.order_number == "ORD-001"
assert response.total_amount == 125.00
@pytest.mark.unit
@pytest.mark.schema
class TestOrderItemResponseSchema:
"""Test OrderItemResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"order_id": 1,
"product_id": 1,
"product_name": "Test Product",
"product_sku": "SKU-001",
"quantity": 2,
"unit_price": 50.00,
"total_price": 100.00,
"inventory_reserved": True,
"inventory_fulfilled": False,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = OrderItemResponse(**data)
assert response.id == 1
assert response.quantity == 2
assert response.total_price == 100.00
@pytest.mark.unit
@pytest.mark.schema
class TestOrderListResponseSchema:
"""Test OrderListResponse schema."""
def test_valid_list_response(self):
"""Test valid list response structure."""
response = OrderListResponse(
orders=[],
total=0,
skip=0,
limit=10,
)
assert response.orders == []
assert response.total == 0
assert response.skip == 0
assert response.limit == 10

View File

@@ -0,0 +1,243 @@
# tests/unit/models/schema/test_product.py
"""Unit tests for product Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.product import (
ProductCreate,
ProductUpdate,
ProductResponse,
ProductDetailResponse,
ProductListResponse,
)
@pytest.mark.unit
@pytest.mark.schema
class TestProductCreateSchema:
"""Test ProductCreate schema validation."""
def test_valid_product_create(self):
"""Test valid product creation data."""
product = ProductCreate(
marketplace_product_id=1,
product_id="SKU-001",
price=99.99,
currency="EUR",
)
assert product.marketplace_product_id == 1
assert product.product_id == "SKU-001"
assert product.price == 99.99
def test_marketplace_product_id_required(self):
"""Test marketplace_product_id is required."""
with pytest.raises(ValidationError) as exc_info:
ProductCreate(price=99.99)
assert "marketplace_product_id" in str(exc_info.value).lower()
def test_price_must_be_non_negative(self):
"""Test price must be >= 0."""
with pytest.raises(ValidationError) as exc_info:
ProductCreate(
marketplace_product_id=1,
price=-10.00,
)
assert "price" in str(exc_info.value).lower()
def test_price_zero_is_valid(self):
"""Test price of 0 is valid."""
product = ProductCreate(
marketplace_product_id=1,
price=0,
)
assert product.price == 0
def test_sale_price_must_be_non_negative(self):
"""Test sale_price must be >= 0."""
with pytest.raises(ValidationError) as exc_info:
ProductCreate(
marketplace_product_id=1,
sale_price=-5.00,
)
assert "sale_price" in str(exc_info.value).lower()
def test_min_quantity_default(self):
"""Test min_quantity defaults to 1."""
product = ProductCreate(marketplace_product_id=1)
assert product.min_quantity == 1
def test_min_quantity_must_be_at_least_1(self):
"""Test min_quantity must be >= 1."""
with pytest.raises(ValidationError) as exc_info:
ProductCreate(
marketplace_product_id=1,
min_quantity=0,
)
assert "min_quantity" in str(exc_info.value).lower()
def test_max_quantity_must_be_at_least_1(self):
"""Test max_quantity must be >= 1 if provided."""
with pytest.raises(ValidationError) as exc_info:
ProductCreate(
marketplace_product_id=1,
max_quantity=0,
)
assert "max_quantity" in str(exc_info.value).lower()
def test_is_featured_default(self):
"""Test is_featured defaults to False."""
product = ProductCreate(marketplace_product_id=1)
assert product.is_featured is False
def test_all_optional_fields(self):
"""Test product with all optional fields."""
product = ProductCreate(
marketplace_product_id=1,
product_id="SKU-001",
price=100.00,
sale_price=80.00,
currency="EUR",
availability="in_stock",
condition="new",
is_featured=True,
min_quantity=2,
max_quantity=10,
)
assert product.sale_price == 80.00
assert product.availability == "in_stock"
assert product.condition == "new"
assert product.is_featured is True
assert product.max_quantity == 10
@pytest.mark.unit
@pytest.mark.schema
class TestProductUpdateSchema:
"""Test ProductUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = ProductUpdate(price=150.00)
assert update.price == 150.00
assert update.product_id is None
assert update.is_active is None
def test_empty_update_is_valid(self):
"""Test empty update is valid (all fields optional)."""
update = ProductUpdate()
assert update.model_dump(exclude_unset=True) == {}
def test_price_validation(self):
"""Test price must be >= 0 in update."""
with pytest.raises(ValidationError):
ProductUpdate(price=-10.00)
def test_is_active_update(self):
"""Test is_active can be updated."""
update = ProductUpdate(is_active=False)
assert update.is_active is False
def test_is_featured_update(self):
"""Test is_featured can be updated."""
update = ProductUpdate(is_featured=True)
assert update.is_featured is True
@pytest.mark.unit
@pytest.mark.schema
class TestProductResponseSchema:
"""Test ProductResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"vendor_id": 1,
"marketplace_product": {
"id": 1,
"gtin": "1234567890123",
"title": "Test Product",
"description": "A test product",
"brand": "Test Brand",
"category": "Electronics",
"image_url": "https://example.com/image.jpg",
"created_at": datetime.now(),
"updated_at": datetime.now(),
},
"product_id": "SKU-001",
"price": 99.99,
"sale_price": None,
"currency": "EUR",
"availability": "in_stock",
"condition": "new",
"is_featured": False,
"is_active": True,
"display_order": 0,
"min_quantity": 1,
"max_quantity": None,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = ProductResponse(**data)
assert response.id == 1
assert response.vendor_id == 1
assert response.is_active is True
def test_optional_inventory_fields(self):
"""Test optional inventory summary fields."""
from datetime import datetime
data = {
"id": 1,
"vendor_id": 1,
"marketplace_product": {
"id": 1,
"gtin": "1234567890123",
"title": "Test Product",
"description": None,
"brand": None,
"category": None,
"image_url": None,
"created_at": datetime.now(),
"updated_at": datetime.now(),
},
"product_id": None,
"price": None,
"sale_price": None,
"currency": None,
"availability": None,
"condition": None,
"is_featured": False,
"is_active": True,
"display_order": 0,
"min_quantity": 1,
"max_quantity": None,
"created_at": datetime.now(),
"updated_at": datetime.now(),
"total_inventory": 100,
"available_inventory": 80,
}
response = ProductResponse(**data)
assert response.total_inventory == 100
assert response.available_inventory == 80
@pytest.mark.unit
@pytest.mark.schema
class TestProductListResponseSchema:
"""Test ProductListResponse schema."""
def test_valid_list_response(self):
"""Test valid list response structure."""
response = ProductListResponse(
products=[],
total=0,
skip=0,
limit=10,
)
assert response.products == []
assert response.total == 0
assert response.skip == 0
assert response.limit == 10

View File

@@ -0,0 +1,315 @@
# tests/unit/models/schema/test_vendor.py
"""Unit tests for vendor Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.vendor import (
VendorCreate,
VendorUpdate,
VendorResponse,
VendorDetailResponse,
VendorListResponse,
VendorSummary,
)
@pytest.mark.unit
@pytest.mark.schema
class TestVendorCreateSchema:
"""Test VendorCreate schema validation."""
def test_valid_vendor_create(self):
"""Test valid vendor creation data."""
vendor = VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
)
assert vendor.company_id == 1
assert vendor.vendor_code == "TECHSTORE"
assert vendor.subdomain == "techstore"
assert vendor.name == "Tech Store"
def test_company_id_required(self):
"""Test company_id is required."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
vendor_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
)
assert "company_id" in str(exc_info.value).lower()
def test_company_id_must_be_positive(self):
"""Test company_id must be > 0."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=0,
vendor_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
)
assert "company_id" in str(exc_info.value).lower()
def test_vendor_code_required(self):
"""Test vendor_code is required."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
subdomain="techstore",
name="Tech Store",
)
assert "vendor_code" in str(exc_info.value).lower()
def test_vendor_code_uppercase_normalized(self):
"""Test vendor_code is normalized to uppercase."""
vendor = VendorCreate(
company_id=1,
vendor_code="techstore",
subdomain="techstore",
name="Tech Store",
)
assert vendor.vendor_code == "TECHSTORE"
def test_vendor_code_min_length(self):
"""Test vendor_code minimum length (2)."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
vendor_code="T",
subdomain="techstore",
name="Tech Store",
)
assert "vendor_code" in str(exc_info.value).lower()
def test_subdomain_required(self):
"""Test subdomain is required."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
name="Tech Store",
)
assert "subdomain" in str(exc_info.value).lower()
def test_subdomain_uppercase_invalid(self):
"""Test subdomain with uppercase is invalid (validated before normalization)."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="TechStore",
name="Tech Store",
)
assert "subdomain" in str(exc_info.value).lower()
def test_subdomain_valid_format(self):
"""Test subdomain with valid format."""
vendor = VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="tech-store-123",
name="Tech Store",
)
assert vendor.subdomain == "tech-store-123"
def test_subdomain_invalid_format(self):
"""Test subdomain with invalid characters raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="tech_store!",
name="Tech Store",
)
assert "subdomain" in str(exc_info.value).lower()
def test_name_required(self):
"""Test name is required."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="techstore",
)
assert "name" in str(exc_info.value).lower()
def test_name_min_length(self):
"""Test name minimum length (2)."""
with pytest.raises(ValidationError) as exc_info:
VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="techstore",
name="T",
)
assert "name" in str(exc_info.value).lower()
def test_optional_fields(self):
"""Test optional fields."""
vendor = VendorCreate(
company_id=1,
vendor_code="TECHSTORE",
subdomain="techstore",
name="Tech Store",
description="Best tech store",
letzshop_csv_url_fr="https://example.com/fr.csv",
contact_email="contact@techstore.com",
website="https://techstore.com",
)
assert vendor.description == "Best tech store"
assert vendor.letzshop_csv_url_fr == "https://example.com/fr.csv"
assert vendor.contact_email == "contact@techstore.com"
@pytest.mark.unit
@pytest.mark.schema
class TestVendorUpdateSchema:
"""Test VendorUpdate schema validation."""
def test_partial_update(self):
"""Test partial update with only some fields."""
update = VendorUpdate(name="New Tech Store")
assert update.name == "New Tech Store"
assert update.subdomain is None
assert update.is_active is None
def test_empty_update_is_valid(self):
"""Test empty update is valid."""
update = VendorUpdate()
assert update.model_dump(exclude_unset=True) == {}
def test_subdomain_normalized_to_lowercase(self):
"""Test subdomain is normalized to lowercase."""
update = VendorUpdate(subdomain="NewSubdomain")
assert update.subdomain == "newsubdomain"
def test_subdomain_stripped(self):
"""Test subdomain is stripped of whitespace."""
update = VendorUpdate(subdomain=" newsubdomain ")
assert update.subdomain == "newsubdomain"
def test_name_min_length(self):
"""Test name minimum length (2)."""
with pytest.raises(ValidationError):
VendorUpdate(name="X")
def test_is_active_update(self):
"""Test is_active can be updated."""
update = VendorUpdate(is_active=False)
assert update.is_active is False
def test_is_verified_update(self):
"""Test is_verified can be updated."""
update = VendorUpdate(is_verified=True)
assert update.is_verified is True
def test_reset_contact_to_company_flag(self):
"""Test reset_contact_to_company flag."""
update = VendorUpdate(reset_contact_to_company=True)
assert update.reset_contact_to_company is True
@pytest.mark.unit
@pytest.mark.schema
class TestVendorResponseSchema:
"""Test VendorResponse schema."""
def test_from_dict(self):
"""Test creating response from dict."""
from datetime import datetime
data = {
"id": 1,
"vendor_code": "TECHSTORE",
"subdomain": "techstore",
"name": "Tech Store",
"description": "Best tech store",
"company_id": 1,
"letzshop_csv_url_fr": None,
"letzshop_csv_url_en": None,
"letzshop_csv_url_de": None,
"is_active": True,
"is_verified": False,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
response = VendorResponse(**data)
assert response.id == 1
assert response.vendor_code == "TECHSTORE"
assert response.is_active is True
@pytest.mark.unit
@pytest.mark.schema
class TestVendorDetailResponseSchema:
"""Test VendorDetailResponse schema."""
def test_from_dict(self):
"""Test creating detail response from dict."""
from datetime import datetime
data = {
"id": 1,
"vendor_code": "TECHSTORE",
"subdomain": "techstore",
"name": "Tech Store",
"description": None,
"company_id": 1,
"letzshop_csv_url_fr": None,
"letzshop_csv_url_en": None,
"letzshop_csv_url_de": None,
"is_active": True,
"is_verified": True,
"created_at": datetime.now(),
"updated_at": datetime.now(),
# Additional detail fields
"company_name": "Tech Corp",
"owner_email": "owner@techcorp.com",
"owner_username": "owner",
"contact_email": "contact@techstore.com",
"contact_email_inherited": False,
}
response = VendorDetailResponse(**data)
assert response.company_name == "Tech Corp"
assert response.owner_email == "owner@techcorp.com"
assert response.contact_email_inherited is False
@pytest.mark.unit
@pytest.mark.schema
class TestVendorListResponseSchema:
"""Test VendorListResponse schema."""
def test_valid_list_response(self):
"""Test valid list response structure."""
response = VendorListResponse(
vendors=[],
total=0,
skip=0,
limit=10,
)
assert response.vendors == []
assert response.total == 0
@pytest.mark.unit
@pytest.mark.schema
class TestVendorSummarySchema:
"""Test VendorSummary schema."""
def test_from_dict(self):
"""Test creating summary from dict."""
data = {
"id": 1,
"vendor_code": "TECHSTORE",
"subdomain": "techstore",
"name": "Tech Store",
"company_id": 1,
"is_active": True,
}
summary = VendorSummary(**data)
assert summary.id == 1
assert summary.vendor_code == "TECHSTORE"
assert summary.is_active is True

View File

@@ -1,598 +0,0 @@
# tests/unit/models/test_database_models.py
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.customer import Customer, CustomerAddress
from models.database.inventory import Inventory
from models.database.marketplace_import_job import MarketplaceImportJob
from models.database.marketplace_product import MarketplaceProduct
from models.database.order import Order, OrderItem
from models.database.product import Product
from models.database.user import User
from models.database.vendor import Role, Vendor, VendorUser
@pytest.mark.unit
@pytest.mark.database
class TestUserModel:
"""Test User model"""
def test_user_creation(self, db):
"""Test User model creation and relationships"""
user = User(
email="db_test@example.com",
username="dbtest",
hashed_password="hashed_password_123",
role="user",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
assert user.id is not None
assert user.email == "db_test@example.com"
assert user.username == "dbtest"
assert user.role == "user"
assert user.is_active is True
assert user.created_at is not None
assert user.updated_at is not None
def test_user_email_uniqueness(self, db):
"""Test email unique constraint"""
user1 = User(
email="unique@example.com",
username="user1",
hashed_password="hash1",
)
db.add(user1)
db.commit()
# Duplicate email should raise error
with pytest.raises(IntegrityError):
user2 = User(
email="unique@example.com",
username="user2",
hashed_password="hash2",
)
db.add(user2)
db.commit()
@pytest.mark.unit
@pytest.mark.database
class TestVendorModel:
"""Test Vendor model"""
def test_vendor_creation_with_owner(self, db, test_user):
"""Test Vendor model with owner relationship"""
vendor = Vendor(
vendor_code="DBTEST",
subdomain="dbtest",
name="Database Test Vendor",
description="Testing vendor model",
owner_user_id=test_user.id,
contact_email="contact@dbtest.com",
contact_phone="+1234567890",
business_address="123 Test Street",
is_active=True,
is_verified=False,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
assert vendor.id is not None
assert vendor.vendor_code == "DBTEST"
assert vendor.subdomain == "dbtest"
assert vendor.name == "Database Test Vendor"
assert vendor.owner_user_id == test_user.id
assert vendor.owner.username == test_user.username
assert vendor.contact_email == "contact@dbtest.com"
assert vendor.is_active is True
assert vendor.is_verified is False
def test_vendor_with_letzshop_urls(self, db, test_user):
"""Test Vendor model with multi-language Letzshop URLs"""
vendor = Vendor(
vendor_code="MULTILANG",
subdomain="multilang",
name="Multi-Language Vendor",
owner_user_id=test_user.id,
letzshop_csv_url_fr="https://example.com/feed_fr.csv",
letzshop_csv_url_en="https://example.com/feed_en.csv",
letzshop_csv_url_de="https://example.com/feed_de.csv",
is_active=True,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
assert vendor.letzshop_csv_url_fr == "https://example.com/feed_fr.csv"
assert vendor.letzshop_csv_url_en == "https://example.com/feed_en.csv"
assert vendor.letzshop_csv_url_de == "https://example.com/feed_de.csv"
def test_vendor_code_uniqueness(self, db, test_user):
"""Test vendor_code unique constraint"""
vendor1 = Vendor(
vendor_code="UNIQUE",
subdomain="unique1",
name="Unique Vendor 1",
owner_user_id=test_user.id,
)
db.add(vendor1)
db.commit()
# Duplicate vendor_code should raise error
with pytest.raises(IntegrityError):
vendor2 = Vendor(
vendor_code="UNIQUE",
subdomain="unique2",
name="Unique Vendor 2",
owner_user_id=test_user.id,
)
db.add(vendor2)
db.commit()
def test_subdomain_uniqueness(self, db, test_user):
"""Test subdomain unique constraint"""
vendor1 = Vendor(
vendor_code="VENDOR1",
subdomain="testsubdomain",
name="Vendor 1",
owner_user_id=test_user.id,
)
db.add(vendor1)
db.commit()
# Duplicate subdomain should raise error
with pytest.raises(IntegrityError):
vendor2 = Vendor(
vendor_code="VENDOR2",
subdomain="testsubdomain",
name="Vendor 2",
owner_user_id=test_user.id,
)
db.add(vendor2)
db.commit()
@pytest.mark.unit
@pytest.mark.database
class TestTeamModels:
"""Test VendorUser and Role models"""
def test_role_creation(self, db, test_vendor):
"""Test Role model creation"""
role = Role(
vendor_id=test_vendor.id,
name="Manager",
permissions=["products.create", "orders.view"],
)
db.add(role)
db.commit()
db.refresh(role)
assert role.id is not None
assert role.vendor_id == test_vendor.id
assert role.name == "Manager"
assert "products.create" in role.permissions
assert "orders.view" in role.permissions
def test_vendor_user_creation(self, db, test_vendor, test_user):
"""Test VendorUser model for team management"""
# Create a role
role = Role(
vendor_id=test_vendor.id,
name="Manager",
permissions=["products.create", "orders.view"],
)
db.add(role)
db.commit()
# Create vendor user
vendor_user = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role.id,
is_active=True,
)
db.add(vendor_user)
db.commit()
db.refresh(vendor_user)
assert vendor_user.id is not None
assert vendor_user.vendor_id == test_vendor.id
assert vendor_user.user_id == test_user.id
assert vendor_user.role.name == "Manager"
assert "products.create" in vendor_user.role.permissions
def test_vendor_user_uniqueness(self, db, test_vendor, test_user):
"""Test vendor_user unique constraint (one user per vendor)"""
role = Role(
vendor_id=test_vendor.id,
name="Editor",
permissions=["products.view"],
)
db.add(role)
db.commit()
vendor_user1 = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role.id,
)
db.add(vendor_user1)
db.commit()
# Same user can't be added to same vendor twice
with pytest.raises(IntegrityError):
vendor_user2 = VendorUser(
vendor_id=test_vendor.id,
user_id=test_user.id,
role_id=role.id,
)
db.add(vendor_user2)
db.commit()
@pytest.mark.unit
@pytest.mark.database
class TestMarketplaceProductModel:
"""Test MarketplaceProduct model"""
def test_marketplace_product_creation(self, db, test_vendor):
"""Test MarketplaceProduct model creation with vendor_id"""
marketplace_product = MarketplaceProduct(
vendor_id=test_vendor.id,
marketplace_product_id="DB_TEST_001",
title="Database Test Product",
description="Testing product model",
price="25.99",
currency="USD",
brand="DBTest",
gtin="1234567890123",
availability="in stock",
marketplace="Letzshop",
)
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
assert marketplace_product.id is not None
assert marketplace_product.vendor_id == test_vendor.id
assert marketplace_product.marketplace_product_id == "DB_TEST_001"
assert marketplace_product.title == "Database Test Product"
assert marketplace_product.marketplace == "Letzshop"
assert marketplace_product.created_at is not None
def test_marketplace_product_id_uniqueness(self, db, test_vendor):
"""Test unique marketplace_product_id constraint"""
product1 = MarketplaceProduct(
vendor_id=test_vendor.id,
marketplace_product_id="UNIQUE_001",
title="Product 1",
marketplace="Letzshop",
)
db.add(product1)
db.commit()
# Duplicate marketplace_product_id should raise error
with pytest.raises(IntegrityError):
product2 = MarketplaceProduct(
vendor_id=test_vendor.id,
marketplace_product_id="UNIQUE_001",
title="Product 2",
marketplace="Letzshop",
)
db.add(product2)
db.commit()
@pytest.mark.unit
@pytest.mark.database
class TestProductModel:
"""Test Product (vendor catalog) model"""
def test_product_creation(self, db, test_vendor, test_marketplace_product):
"""Test Product model linking vendor catalog to marketplace product"""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
product_id="VENDOR_PROD_001",
price=89.99, # Vendor override price
currency="EUR",
availability="in stock",
is_featured=True,
is_active=True,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.id is not None
assert product.vendor_id == test_vendor.id
assert product.marketplace_product_id == test_marketplace_product.id
assert product.price == 89.99
assert product.is_featured is True
assert product.vendor.vendor_code == test_vendor.vendor_code
assert product.marketplace_product.title == test_marketplace_product.title
def test_product_unique_per_vendor(self, db, test_vendor, test_marketplace_product):
"""Test that same marketplace product can't be added twice to vendor catalog"""
product1 = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
is_active=True,
)
db.add(product1)
db.commit()
# Same marketplace product to same vendor should fail
with pytest.raises(IntegrityError):
product2 = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
is_active=True,
)
db.add(product2)
db.commit()
@pytest.mark.unit
@pytest.mark.database
class TestInventoryModel:
"""Test Inventory model"""
def test_inventory_creation_with_product(self, db, test_vendor, test_product):
"""Test Inventory model linked to product"""
inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=150,
reserved_quantity=10,
gtin=test_product.marketplace_product.gtin,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
assert inventory.id is not None
assert inventory.product_id == test_product.id
assert inventory.vendor_id == test_vendor.id
assert inventory.location == "WAREHOUSE_A"
assert inventory.quantity == 150
assert inventory.reserved_quantity == 10
assert inventory.available_quantity == 140 # 150 - 10
def test_inventory_unique_product_location(self, db, test_vendor, test_product):
"""Test unique constraint on product_id + location"""
inventory1 = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=100,
)
db.add(inventory1)
db.commit()
# Same product + location should fail
with pytest.raises(IntegrityError):
inventory2 = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=50,
)
db.add(inventory2)
db.commit()
@pytest.mark.unit
@pytest.mark.database
class TestMarketplaceImportJobModel:
"""Test MarketplaceImportJob model"""
def test_import_job_creation(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob model with relationships"""
import_job = MarketplaceImportJob(
vendor_id=test_vendor.id,
user_id=test_user.id,
marketplace="Letzshop",
source_url="https://example.com/feed.csv",
status="pending",
imported_count=0,
updated_count=0,
error_count=0,
total_processed=0,
)
db.add(import_job)
db.commit()
db.refresh(import_job)
assert import_job.id is not None
assert import_job.vendor_id == test_vendor.id
assert import_job.user_id == test_user.id
assert import_job.marketplace == "Letzshop"
assert import_job.source_url == "https://example.com/feed.csv"
assert import_job.status == "pending"
assert import_job.vendor.vendor_code == test_vendor.vendor_code
assert import_job.user.username == test_user.username
@pytest.mark.unit
@pytest.mark.database
class TestCustomerModel:
"""Test Customer model"""
def test_customer_creation(self, db, test_vendor):
"""Test Customer model with vendor isolation"""
customer = Customer(
vendor_id=test_vendor.id,
email="customer@example.com",
hashed_password="hashed_password",
first_name="John",
last_name="Doe",
customer_number="CUST001",
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
assert customer.id is not None
assert customer.vendor_id == test_vendor.id
assert customer.email == "customer@example.com"
assert customer.customer_number == "CUST001"
assert customer.first_name == "John"
assert customer.last_name == "Doe"
assert customer.vendor.vendor_code == test_vendor.vendor_code
def test_customer_email_unique_per_vendor(self, db, test_vendor):
"""Test email is unique within vendor scope only"""
customer1 = Customer(
vendor_id=test_vendor.id,
email="same@example.com",
hashed_password="hash1",
first_name="Customer",
last_name="One",
customer_number="CUST001",
)
db.add(customer1)
db.commit()
# Same email in same vendor should fail
with pytest.raises(IntegrityError):
customer2 = Customer(
vendor_id=test_vendor.id,
email="same@example.com",
hashed_password="hash2",
first_name="Customer",
last_name="Two",
customer_number="CUST002",
)
db.add(customer2)
db.commit()
def test_customer_address_creation(self, db, test_vendor, test_customer):
"""Test CustomerAddress model"""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
address_type="shipping",
first_name="John",
last_name="Doe",
address_line_1="123 Main St",
city="Luxembourg",
postal_code="L-1234",
country="Luxembourg",
is_default=True,
)
db.add(address)
db.commit()
db.refresh(address)
assert address.id is not None
assert address.vendor_id == test_vendor.id
assert address.customer_id == test_customer.id
assert address.address_type == "shipping"
assert address.is_default is True
@pytest.mark.unit
@pytest.mark.database
class TestOrderModel:
"""Test Order model"""
def test_order_creation(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test Order model with customer relationship"""
order = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="ORD-001",
status="pending",
subtotal=99.99,
total_amount=99.99,
currency="EUR",
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order)
db.commit()
db.refresh(order)
assert order.id is not None
assert order.vendor_id == test_vendor.id
assert order.customer_id == test_customer.id
assert order.order_number == "ORD-001"
assert order.status == "pending"
assert float(order.total_amount) == 99.99
def test_order_item_creation(self, db, test_order, test_product):
"""Test OrderItem model"""
order_item = OrderItem(
order_id=test_order.id,
product_id=test_product.id,
product_name=test_product.marketplace_product.title,
product_sku=test_product.product_id,
quantity=2,
unit_price=49.99,
total_price=99.98,
)
db.add(order_item)
db.commit()
db.refresh(order_item)
assert order_item.id is not None
assert order_item.order_id == test_order.id
assert order_item.product_id == test_product.id
assert order_item.quantity == 2
assert float(order_item.unit_price) == 49.99
assert float(order_item.total_price) == 99.98
def test_order_number_uniqueness(
self, db, test_vendor, test_customer, test_customer_address
):
"""Test order_number unique constraint"""
order1 = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="UNIQUE-ORD-001",
status="pending",
subtotal=50.00,
total_amount=50.00,
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order1)
db.commit()
# Duplicate order number should fail
with pytest.raises(IntegrityError):
order2 = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="UNIQUE-ORD-001",
status="pending",
subtotal=75.00,
total_amount=75.00,
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order2)
db.commit()