595 lines
19 KiB
Python
595 lines
19 KiB
Python
# tests/unit/models/test_database_models.py
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from models.database.marketplace_product import MarketplaceProduct
|
|
from models.database.vendor import Vendor, VendorUser, Role
|
|
from models.database.inventory import Inventory
|
|
from models.database.user import User
|
|
from models.database.marketplace_import_job import MarketplaceImportJob
|
|
from models.database.product import Product
|
|
from models.database.customer import Customer, CustomerAddress
|
|
from models.database.order import Order, OrderItem
|
|
|
|
|
|
@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()
|