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

major refactoring adding vendor and customer features

Browse Source
This commit is contained in:
Samir Boulahtit
2025-10-11 09:09:25 +02:00
parent f569995883
commit dd16198276
126 changed files with 15109 additions and 3747 deletions
Download Patch File Download Diff File Expand all files Collapse all files

585
tests/unit/models/test_database_models.py
View File

@@ -1,16 +1,24 @@
# 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
from models.database.stock import Stock
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 TestDatabaseModels:
def test_user_model(self, db):
class TestUserModel:
"""Test User model"""
def test_user_creation(self, db):
"""Test User model creation and relationships"""
user = User(
email="db_test@example.com",
@@ -26,52 +34,49 @@ class TestDatabaseModels:
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_product_model(self, db):
"""Test MarketplaceProduct model creation"""
marketplace_product = MarketplaceProduct(
marketplace_product_id="DB_TEST_001",
title="Database Test MarketplaceProduct",
description="Testing product model",
price="25.99",
currency="USD",
brand="DBTest",
gtin="1234567890123",
availability="in stock",
marketplace="TestDB",
vendor_name="DBTestVendor",
def test_user_email_uniqueness(self, db):
"""Test email unique constraint"""
user1 = User(
email="unique@example.com",
username="user1",
hashed_password="hash1",
)
db.add(marketplace_product)
db.add(user1)
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.created_at is not None
# 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_stock_model(self, db):
"""Test Stock model creation"""
stock = Stock(gtin="1234567890123", location="DB_WAREHOUSE", quantity=150)
db.add(stock)
db.commit()
db.refresh(stock)
@pytest.mark.unit
@pytest.mark.database
class TestVendorModel:
"""Test Vendor model"""
assert stock.id is not None
assert stock.gtin == "1234567890123"
assert stock.location == "DB_WAREHOUSE"
assert stock.quantity == 150
def test_vendor_model_with_owner(self, db, test_user):
def test_vendor_creation_with_owner(self, db, test_user):
"""Test Vendor model with owner relationship"""
vendor = Vendor(
vendor_code="DBTEST",
vendor_name="Database Test Vendor",
subdomain="dbtest",
name="Database Test Vendor",
description="Testing vendor model",
owner_id=test_user.id,
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,
)
@@ -82,18 +87,508 @@ class TestDatabaseModels:
assert vendor.id is not None
assert vendor.vendor_code == "DBTEST"
assert vendor.owner_id == test_user.id
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_database_constraints(self, db):
"""Test database constraints and unique indexes"""
# Test unique marketplace_product_id constraint
product1 = MarketplaceProduct(marketplace_product_id="UNIQUE_001", title="MarketplaceProduct 1")
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()
# This should raise an integrity error
with pytest.raises(Exception): # Could be IntegrityError or similar
product2 = MarketplaceProduct(marketplace_product_id="UNIQUE_001", title="MarketplaceProduct 2")
# 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()

6
tests/unit/services/test_admin_service.py
View File

@@ -180,7 +180,7 @@ class TestAdminService:
)
assert test_job is not None
assert test_job.marketplace == test_marketplace_import_job.marketplace
assert test_job.vendor_name == test_marketplace_import_job.vendor_name
assert test_job.vendor_name == test_marketplace_import_job.name
assert test_job.status == test_marketplace_import_job.status
def test_get_marketplace_import_jobs_with_marketplace_filter(self, db, test_marketplace_import_job):
@@ -196,12 +196,12 @@ class TestAdminService:
def test_get_marketplace_import_jobs_with_vendor_filter(self, db, test_marketplace_import_job):
"""Test filtering marketplace import jobs by vendor name"""
result = self.service.get_marketplace_import_jobs(
db, vendor_name=test_marketplace_import_job.vendor_name, skip=0, limit=10
db, vendor_name=test_marketplace_import_job.name, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert test_marketplace_import_job.vendor_name.lower() in job.vendor_name.lower()
assert test_marketplace_import_job.name.lower() in job.vendor_name.lower()
def test_get_marketplace_import_jobs_with_status_filter(self, db, test_marketplace_import_job):
"""Test filtering marketplace import jobs by status"""

505
tests/unit/services/test_inventory_service.py Normal file
View File

@@ -0,0 +1,505 @@
# tests/test_inventory_service.py
import uuid
import pytest
from app.services.inventory_service import InventoryService
from app.exceptions import (
InventoryNotFoundException,
InsufficientInventoryException,
InvalidInventoryOperationException,
InventoryValidationException,
NegativeInventoryException,
InvalidQuantityException,
ValidationException,
)
from models.schemas.inventory import InventoryAdd, InventoryCreate, InventoryUpdate
from models.database.marketplace_product import MarketplaceProduct
from models.database.inventory import Inventory
@pytest.mark.unit
@pytest.mark.inventory
class TestInventoryService:
def setup_method(self):
self.service = InventoryService()
def test_normalize_gtin_invalid(self):
"""Test GTIN normalization with invalid GTINs."""
# Completely invalid values that should return None
assert self.service._normalize_gtin("invalid") is None
assert self.service._normalize_gtin("abcdef") is None
assert self.service._normalize_gtin("") is None
assert self.service._normalize_gtin(None) is None
assert self.service._normalize_gtin(" ") is None # Only whitespace
assert self.service._normalize_gtin("!@#$%") is None # Only special characters
# Mixed invalid characters that become empty after filtering
assert self.service._normalize_gtin("abc-def-ghi") is None # No digits
def test_normalize_gtin_valid(self):
"""Test GTIN normalization with valid GTINs."""
# Test various valid GTIN formats - these should remain unchanged
assert self.service._normalize_gtin("1234567890123") == "1234567890123" # EAN-13
assert self.service._normalize_gtin("123456789012") == "123456789012" # UPC-A
assert self.service._normalize_gtin("12345678") == "12345678" # EAN-8
assert self.service._normalize_gtin("12345678901234") == "12345678901234" # GTIN-14
# Test with decimal points (should be removed)
assert self.service._normalize_gtin("1234567890123.0") == "1234567890123"
# Test with whitespace (should be trimmed)
assert self.service._normalize_gtin(" 1234567890123 ") == "1234567890123"
# Test short GTINs being padded
assert self.service._normalize_gtin("123") == "0000000000123" # Padded to EAN-13
assert self.service._normalize_gtin("12345") == "0000000012345" # Padded to EAN-13
# Test long GTINs being truncated
assert self.service._normalize_gtin("123456789012345") == "3456789012345" # Truncated to 13
def test_normalize_gtin_edge_cases(self):
"""Test GTIN normalization edge cases."""
# Test numeric inputs
assert self.service._normalize_gtin(1234567890123) == "1234567890123"
assert self.service._normalize_gtin(123) == "0000000000123"
# Test mixed valid/invalid characters
assert self.service._normalize_gtin("123-456-789-012") == "123456789012" # Dashes removed
assert self.service._normalize_gtin("123 456 789 012") == "123456789012" # Spaces removed
assert self.service._normalize_gtin("ABC123456789012DEF") == "123456789012" # Letters removed
def test_set_inventory_new_entry_success(self, db):
"""Test setting inventory for a new GTIN/location combination successfully."""
unique_id = str(uuid.uuid4())[:8]
inventory_data = InventoryCreate(
gtin="1234567890123", location=f"WAREHOUSE_A_{unique_id}", quantity=100
)
result = self.service.set_inventory(db, inventory_data)
assert result.gtin == "1234567890123"
assert result.location.upper() == f"WAREHOUSE_A_{unique_id}".upper()
assert result.quantity == 100
def test_set_inventory_existing_entry_success(self, db, test_inventory):
"""Test setting inventory for an existing GTIN/location combination successfully."""
inventory_data = InventoryCreate(
gtin=test_inventory.gtin,
location=test_inventory.location, # Use exact same location as test_inventory
quantity=200,
)
result = self.service.set_inventory(db, inventory_data)
assert result.gtin == test_inventory.gtin
assert result.location == test_inventory.location
assert result.quantity == 200 # Should replace the original quantity
def test_set_inventory_invalid_gtin_validation_error(self, db):
"""Test setting inventory with invalid GTIN returns InventoryValidationException."""
inventory_data = InventoryCreate(
gtin="invalid_gtin", location="WAREHOUSE_A", quantity=100
)
with pytest.raises(InventoryValidationException) as exc_info:
self.service.set_inventory(db, inventory_data)
assert exc_info.value.error_code == "INVENTORY_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
assert exc_info.value.details.get("field") == "gtin"
def test_set_inventory_invalid_quantity_error(self, db):
"""Test setting inventory with invalid quantity through service validation."""
# Test the service validation directly instead of going through Pydantic schema
# This bypasses the Pydantic validation to test service layer validation
# Create a mock inventory data object that bypasses Pydantic validation
class MockInventoryData:
def __init__(self, gtin, location, quantity):
self.gtin = gtin
self.location = location
self.quantity = quantity
mock_inventory_data = MockInventoryData("1234567890123", "WAREHOUSE_A", -10)
# Test the internal validation method directly
with pytest.raises(InvalidQuantityException) as exc_info:
self.service._validate_quantity(-10, allow_zero=True)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Quantity cannot be negative" in str(exc_info.value)
assert exc_info.value.details.get("quantity") == -10
def test_add_inventory_new_entry_success(self, db):
"""Test adding inventory for a new GTIN/location combination successfully."""
unique_id = str(uuid.uuid4())[:8]
inventory_data = InventoryAdd(
gtin="1234567890123", location=f"WAREHOUSE_B_{unique_id}", quantity=50
)
result = self.service.add_inventory(db, inventory_data)
assert result.gtin == "1234567890123"
assert result.location.upper() == f"WAREHOUSE_B_{unique_id}".upper()
assert result.quantity == 50
def test_add_inventory_existing_entry_success(self, db, test_inventory):
"""Test adding inventory to an existing GTIN/location combination successfully."""
original_quantity = test_inventory.quantity
inventory_data = InventoryAdd(
gtin=test_inventory.gtin,
location=test_inventory.location, # Use exact same location as test_inventory
quantity=25,
)
result = self.service.add_inventory(db, inventory_data)
assert result.gtin == test_inventory.gtin
assert result.location == test_inventory.location
assert result.quantity == original_quantity + 25
def test_add_inventory_invalid_gtin_validation_error(self, db):
"""Test adding inventory with invalid GTIN returns InventoryValidationException."""
inventory_data = InventoryAdd(gtin="invalid_gtin", location="WAREHOUSE_A", quantity=50)
with pytest.raises(InventoryValidationException) as exc_info:
self.service.add_inventory(db, inventory_data)
assert exc_info.value.error_code == "INVENTORY_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_add_inventory_invalid_quantity_error(self, db):
"""Test adding inventory with invalid quantity through service validation."""
# Test zero quantity which should fail for add_inventory (doesn't allow zero)
# This tests the service validation: allow_zero=False for add operations
with pytest.raises(InvalidQuantityException) as exc_info:
self.service._validate_quantity(0, allow_zero=False)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Quantity must be positive" in str(exc_info.value)
def test_remove_inventory_success(self, db, test_inventory):
"""Test removing inventory successfully."""
original_quantity = test_inventory.quantity
remove_quantity = min(10, original_quantity) # Ensure we don't remove more than available
inventory_data = InventoryAdd(
gtin=test_inventory.gtin,
location=test_inventory.location, # Use exact same location as test_inventory
quantity=remove_quantity,
)
result = self.service.remove_inventory(db, inventory_data)
assert result.gtin == test_inventory.gtin
assert result.location == test_inventory.location
assert result.quantity == original_quantity - remove_quantity
def test_remove_inventory_insufficient_inventory_error(self, db, test_inventory):
"""Test removing more inventory than available returns InsufficientInventoryException."""
inventory_data = InventoryAdd(
gtin=test_inventory.gtin,
location=test_inventory.location, # Use exact same location as test_inventory
quantity=test_inventory.quantity + 10, # More than available
)
with pytest.raises(InsufficientInventoryException) as exc_info:
self.service.remove_inventory(db, inventory_data)
assert exc_info.value.error_code == "INSUFFICIENT_INVENTORY"
assert exc_info.value.details["gtin"] == test_inventory.gtin
assert exc_info.value.details["location"] == test_inventory.location
assert exc_info.value.details["requested_quantity"] == test_inventory.quantity + 10
assert exc_info.value.details["available_quantity"] == test_inventory.quantity
def test_remove_inventory_nonexistent_entry_not_found(self, db):
"""Test removing inventory from non-existent GTIN/location returns InventoryNotFoundException."""
unique_id = str(uuid.uuid4())[:8]
inventory_data = InventoryAdd(
gtin="9999999999999", location=f"NONEXISTENT_{unique_id}", quantity=10
)
with pytest.raises(InventoryNotFoundException) as exc_info:
self.service.remove_inventory(db, inventory_data)
assert exc_info.value.error_code == "INVENTORY_NOT_FOUND"
assert "9999999999999" in str(exc_info.value)
assert exc_info.value.details["resource_type"] == "Inventory"
def test_remove_inventory_invalid_gtin_validation_error(self, db):
"""Test removing inventory with invalid GTIN returns InventoryValidationException."""
inventory_data = InventoryAdd(gtin="invalid_gtin", location="WAREHOUSE_A", quantity=10)
with pytest.raises(InventoryValidationException) as exc_info:
self.service.remove_inventory(db, inventory_data)
assert exc_info.value.error_code == "INVENTORY_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_remove_inventory_negative_result_error(self, db, test_inventory):
"""Test removing inventory that would result in negative quantity returns NegativeInventoryException."""
# This is handled by InsufficientInventoryException, but test the logic
inventory_data = InventoryAdd(
gtin=test_inventory.gtin,
location=test_inventory.location,
quantity=test_inventory.quantity + 1, # One more than available
)
with pytest.raises(InsufficientInventoryException) as exc_info:
self.service.remove_inventory(db, inventory_data)
# The service prevents negative inventory through InsufficientInventoryException
assert exc_info.value.error_code == "INSUFFICIENT_INVENTORY"
def test_get_inventory_by_gtin_success(self, db, test_inventory, test_marketplace_product):
"""Test getting inventory summary by GTIN successfully."""
result = self.service.get_inventory_by_gtin(db, test_inventory.gtin)
assert result.gtin == test_inventory.gtin
assert result.total_quantity == test_inventory.quantity
assert len(result.locations) == 1
assert result.locations[0].location == test_inventory.location
assert result.locations[0].quantity == test_inventory.quantity
assert result.product_title == test_marketplace_product.title
def test_get_inventory_by_gtin_multiple_locations_success(self, db, test_marketplace_product):
"""Test getting inventory summary with multiple locations successfully."""
unique_gtin = test_marketplace_product.gtin
unique_id = str(uuid.uuid4())[:8]
# Create multiple inventory entries for the same GTIN with unique locations
inventory1 = Inventory(gtin=unique_gtin, location=f"WAREHOUSE_A_{unique_id}", quantity=50)
inventory2 = Inventory(gtin=unique_gtin, location=f"WAREHOUSE_B_{unique_id}", quantity=30)
db.add(inventory1)
db.add(inventory2)
db.commit()
result = self.service.get_inventory_by_gtin(db, unique_gtin)
assert result.gtin == unique_gtin
assert result.total_quantity == 80
assert len(result.locations) == 2
def test_get_inventory_by_gtin_not_found_error(self, db):
"""Test getting inventory for non-existent GTIN returns InventoryNotFoundException."""
with pytest.raises(InventoryNotFoundException) as exc_info:
self.service.get_inventory_by_gtin(db, "9999999999999")
assert exc_info.value.error_code == "INVENTORY_NOT_FOUND"
assert "9999999999999" in str(exc_info.value)
assert exc_info.value.details["resource_type"] == "Inventory"
def test_get_inventory_by_gtin_invalid_gtin_validation_error(self, db):
"""Test getting inventory with invalid GTIN returns InventoryValidationException."""
with pytest.raises(InventoryValidationException) as exc_info:
self.service.get_inventory_by_gtin(db, "invalid_gtin")
assert exc_info.value.error_code == "INVENTORY_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_get_total_inventory_success(self, db, test_inventory, test_marketplace_product):
"""Test getting total inventory for a GTIN successfully."""
result = self.service.get_total_inventory(db, test_inventory.gtin)
assert result["gtin"] == test_inventory.gtin
assert result["total_quantity"] == test_inventory.quantity
assert result["product_title"] == test_marketplace_product.title
assert result["locations_count"] == 1
def test_get_total_inventory_invalid_gtin_validation_error(self, db):
"""Test getting total inventory with invalid GTIN returns InventoryValidationException."""
with pytest.raises(InventoryValidationException) as exc_info:
self.service.get_total_inventory(db, "invalid_gtin")
assert exc_info.value.error_code == "INVENTORY_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_get_total_inventory_not_found_error(self, db):
"""Test getting total inventory for non-existent GTIN returns InventoryNotFoundException."""
with pytest.raises(InventoryNotFoundException) as exc_info:
self.service.get_total_inventory(db, "9999999999999")
assert exc_info.value.error_code == "INVENTORY_NOT_FOUND"
def test_get_all_inventory_no_filters_success(self, db, test_inventory):
"""Test getting all inventory without filters successfully."""
result = self.service.get_all_inventory(db)
assert len(result) >= 1
assert any(inventory.gtin == test_inventory.gtin for inventory in result)
def test_get_all_inventory_with_location_filter_success(self, db, test_inventory):
"""Test getting all inventory with location filter successfully."""
result = self.service.get_all_inventory(db, location=test_inventory.location)
assert len(result) >= 1
# Check that all returned inventory match the filter (case insensitive)
for inventory in result:
assert test_inventory.location.upper() in inventory.location.upper()
def test_get_all_inventory_with_gtin_filter_success(self, db, test_inventory):
"""Test getting all inventory with GTIN filter successfully."""
result = self.service.get_all_inventory(db, gtin=test_inventory.gtin)
assert len(result) >= 1
assert all(inventory.gtin == test_inventory.gtin for inventory in result)
def test_get_all_inventory_with_pagination_success(self, db):
"""Test getting all inventory with pagination successfully."""
unique_prefix = str(uuid.uuid4())[:8]
# Create multiple inventory entries with unique GTINs and locations
for i in range(5):
inventory = Inventory(
gtin=f"1234567890{i:03d}", # Creates valid 13-digit GTINs
location=f"WAREHOUSE_{unique_prefix}_{i}",
quantity=10,
)
db.add(inventory)
db.commit()
result = self.service.get_all_inventory(db, skip=2, limit=2)
assert len(result) <= 2 # Should be at most 2, might be less if other records exist
def test_update_inventory_success(self, db, test_inventory):
"""Test updating inventory quantity successfully."""
inventory_update = InventoryUpdate(quantity=150)
result = self.service.update_inventory(db, test_inventory.id, inventory_update)
assert result.id == test_inventory.id
assert result.quantity == 150
def test_update_inventory_not_found_error(self, db):
"""Test updating non-existent inventory entry returns InventoryNotFoundException."""
inventory_update = InventoryUpdate(quantity=150)
with pytest.raises(InventoryNotFoundException) as exc_info:
self.service.update_inventory(db, 99999, inventory_update)
assert exc_info.value.error_code == "INVENTORY_NOT_FOUND"
assert "99999" in str(exc_info.value)
def test_update_inventory_invalid_quantity_error(self, db, test_inventory):
"""Test updating inventory with invalid quantity returns InvalidQuantityException."""
inventory_update = InventoryUpdate(quantity=-10)
with pytest.raises(InvalidQuantityException) as exc_info:
self.service.update_inventory(db, test_inventory.id, inventory_update)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Quantity cannot be negative" in str(exc_info.value)
def test_delete_inventory_success(self, db, test_inventory):
"""Test deleting inventory entry successfully."""
inventory_id = test_inventory.id
result = self.service.delete_inventory(db, inventory_id)
assert result is True
# Verify the inventory is actually deleted
deleted_inventory = db.query(Inventory).filter(Inventory.id == inventory_id).first()
assert deleted_inventory is None
def test_delete_inventory_not_found_error(self, db):
"""Test deleting non-existent inventory entry returns InventoryNotFoundException."""
with pytest.raises(InventoryNotFoundException) as exc_info:
self.service.delete_inventory(db, 99999)
assert exc_info.value.error_code == "INVENTORY_NOT_FOUND"
assert "99999" in str(exc_info.value)
def test_get_low_inventory_items_success(self, db, test_inventory, test_marketplace_product):
"""Test getting low inventory items successfully."""
# Set inventory to a low value
test_inventory.quantity = 5
db.commit()
result = self.service.get_low_inventory_items(db, threshold=10)
assert len(result) >= 1
low_inventory_item = next((item for item in result if item["gtin"] == test_inventory.gtin), None)
assert low_inventory_item is not None
assert low_inventory_item["current_quantity"] == 5
assert low_inventory_item["location"] == test_inventory.location
assert low_inventory_item["product_title"] == test_marketplace_product.title
def test_get_low_inventory_items_invalid_threshold_error(self, db):
"""Test getting low inventory items with invalid threshold returns InvalidQuantityException."""
with pytest.raises(InvalidQuantityException) as exc_info:
self.service.get_low_inventory_items(db, threshold=-5)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Threshold must be non-negative" in str(exc_info.value)
def test_get_inventory_summary_by_location_success(self, db, test_inventory):
"""Test getting inventory summary by location successfully."""
result = self.service.get_inventory_summary_by_location(db, test_inventory.location)
assert result["location"] == test_inventory.location.upper() # Service normalizes to uppercase
assert result["total_items"] >= 1
assert result["total_quantity"] >= test_inventory.quantity
assert result["unique_gtins"] >= 1
def test_get_inventory_summary_by_location_empty_result(self, db):
"""Test getting inventory summary for location with no inventory."""
unique_id = str(uuid.uuid4())[:8]
result = self.service.get_inventory_summary_by_location(db, f"EMPTY_LOCATION_{unique_id}")
assert result["total_items"] == 0
assert result["total_quantity"] == 0
assert result["unique_gtins"] == 0
def test_validate_quantity_edge_cases(self, db):
"""Test quantity validation with edge cases."""
# Test zero quantity with allow_zero=True (should succeed)
inventory_data = InventoryCreate(gtin="1234567890123", location="WAREHOUSE_A", quantity=0)
result = self.service.set_inventory(db, inventory_data)
assert result.quantity == 0
# Test zero quantity with add_inventory (should fail - doesn't allow zero)
inventory_data_add = InventoryAdd(gtin="1234567890123", location="WAREHOUSE_B", quantity=0)
with pytest.raises(InvalidQuantityException):
self.service.add_inventory(db, inventory_data_add)
def test_exception_structure_consistency(self, db):
"""Test that all exceptions follow the consistent LetzShopException structure."""
# Test with a known error case
with pytest.raises(InventoryNotFoundException) as exc_info:
self.service.get_inventory_by_gtin(db, "9999999999999")
exception = exc_info.value
# Verify exception structure matches LetzShopException.to_dict()
assert hasattr(exception, 'error_code')
assert hasattr(exception, 'message')
assert hasattr(exception, 'status_code')
assert hasattr(exception, 'details')
assert isinstance(exception.error_code, str)
assert isinstance(exception.message, str)
assert isinstance(exception.status_code, int)
assert isinstance(exception.details, dict)
@pytest.fixture
def test_product_with_inventory(db, test_inventory):
"""Create a test product that corresponds to the test inventory."""
product = MarketplaceProduct(
marketplace_product_id="MP_TEST_001",
title="Inventory Test MarketplaceProduct",
gtin=test_inventory.gtin,
price="29.99",
brand="TestBrand",
marketplace="Letzshop",
)
db.add(product)
db.commit()
db.refresh(product)
return product

14
tests/unit/services/test_marketplace_service.py
View File

@@ -28,13 +28,13 @@ class TestMarketplaceService:
def test_validate_vendor_access_success(self, db, test_vendor, test_user):
"""Test successful vendor access validation"""
# Set the vendor owner to the test user
test_vendor.owner_id = test_user.id
test_vendor.owner_user_id = test_user.id
db.commit()
result = self.service.validate_vendor_access(db, test_vendor.vendor_code, test_user)
assert result.vendor_code == test_vendor.vendor_code
assert result.owner_id == test_user.id
assert result.owner_user_id == test_user.id
def test_validate_vendor_access_admin_can_access_any_vendor(
self, db, test_vendor, test_admin
@@ -59,7 +59,7 @@ class TestMarketplaceService:
):
"""Test vendor access validation when user doesn't own the vendor """
# Set the vendor owner to a different user
test_vendor.owner_id = other_user.id
test_vendor.owner_user_id = other_user.id
db.commit()
with pytest.raises(UnauthorizedVendorAccessException) as exc_info:
@@ -73,7 +73,7 @@ class TestMarketplaceService:
def test_create_import_job_success(self, db, test_vendor, test_user):
"""Test successful creation of import job"""
# Set the vendor owner to the test user
test_vendor.owner_id = test_user.id
test_vendor.owner_user_id = test_user.id
db.commit()
request = MarketplaceImportJobRequest(
@@ -90,7 +90,7 @@ class TestMarketplaceService:
assert result.user_id == test_user.id
assert result.status == "pending"
assert result.source_url == "https://example.com/products.csv"
assert result.vendor_name == test_vendor.vendor_name
assert result.vendor_name == test_vendor.name
def test_create_import_job_invalid_vendor(self, db, test_user):
"""Test import job creation with invalid vendor """
@@ -111,7 +111,7 @@ class TestMarketplaceService:
def test_create_import_job_unauthorized_access(self, db, test_vendor, test_user, other_user):
"""Test import job creation with unauthorized vendor access"""
# Set the vendor owner to a different user
test_vendor.owner_id = other_user.id
test_vendor.owner_user_id = other_user.id
db.commit()
request = MarketplaceImportJobRequest(
@@ -436,7 +436,7 @@ class TestMarketplaceService:
# Test edge cases and error scenarios
def test_validate_vendor_access_case_insensitive(self, db, test_vendor, test_user):
"""Test vendor access validation is case insensitive"""
test_vendor.owner_id = test_user.id
test_vendor.owner_user_id = test_user.id
db.commit()
# Test with lowercase vendor code

24
tests/unit/services/test_product_service.py
View File

@@ -229,23 +229,23 @@ class TestProductService:
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
assert "NONEXISTENT" in str(exc_info.value)
def test_get_stock_info_success(self, db, test_marketplace_product_with_stock):
"""Test getting stock info for product with stock"""
def test_get_inventory_info_success(self, db, test_marketplace_product_with_inventory):
"""Test getting inventory info for product with inventory"""
# Extract the product from the dictionary
marketplace_product = test_marketplace_product_with_stock['marketplace_product']
marketplace_product = test_marketplace_product_with_inventory['marketplace_product']
stock_info = self.service.get_stock_info(db, marketplace_product.gtin)
inventory_info = self.service.get_inventory_info(db, marketplace_product.gtin)
assert stock_info is not None
assert stock_info.gtin == marketplace_product.gtin
assert stock_info.total_quantity > 0
assert len(stock_info.locations) > 0
assert inventory_info is not None
assert inventory_info.gtin == marketplace_product.gtin
assert inventory_info.total_quantity > 0
assert len(inventory_info.locations) > 0
def test_get_stock_info_no_stock(self, db, test_marketplace_product):
"""Test getting stock info for product without stock"""
stock_info = self.service.get_stock_info(db, test_marketplace_product.gtin or "1234567890123")
def test_get_inventory_info_no_inventory(self, db, test_marketplace_product):
"""Test getting inventory info for product without inventory"""
inventory_info = self.service.get_inventory_info(db, test_marketplace_product.gtin or "1234567890123")
assert stock_info is None
assert inventory_info is None
def test_product_exists_true(self, db, test_marketplace_product):
"""Test product_exists returns True for existing product"""

38
tests/unit/services/test_stats_service.py
View File

@@ -3,7 +3,7 @@ import pytest
from app.services.stats_service import StatsService
from models.database.marketplace_product import MarketplaceProduct
from models.database.stock import Stock
from models.database.inventory import Inventory
@pytest.mark.unit
@@ -15,7 +15,7 @@ class TestStatsService:
"""Setup method following the same pattern as other service tests"""
self.service = StatsService()
def test_get_comprehensive_stats_basic(self, db, test_marketplace_product, test_stock):
def test_get_comprehensive_stats_basic(self, db, test_marketplace_product, test_inventory):
"""Test getting comprehensive stats with basic data"""
stats = self.service.get_comprehensive_stats(db)
@@ -24,12 +24,12 @@ class TestStatsService:
assert "unique_categories" in stats
assert "unique_marketplaces" in stats
assert "unique_vendors" in stats
assert "total_stock_entries" in stats
assert "total_inventory_entries" in stats
assert "total_inventory_quantity" in stats
assert stats["total_products"] >= 1
assert stats["total_stock_entries"] >= 1
assert stats["total_inventory_quantity"] >= 10 # test_stock has quantity 10
assert stats["total_inventory_entries"] >= 1
assert stats["total_inventory_quantity"] >= 10 # test_inventory has quantity 10
def test_get_comprehensive_stats_multiple_products(self, db, test_marketplace_product):
"""Test comprehensive stats with multiple products across different dimensions"""
@@ -341,33 +341,33 @@ class TestStatsService:
assert count >= 2 # At least VendorA and VendorB, plus test_marketplace_product vendor
assert isinstance(count, int)
def test_get_stock_statistics(self, db, test_stock):
"""Test getting stock statistics"""
# Add additional stock entries
additional_stocks = [
Stock(
def test_get_inventory_statistics(self, db, test_inventory):
"""Test getting inventory statistics"""
# Add additional inventory entries
additional_inventory = [
Inventory(
gtin="1234567890124",
location="LOCATION2",
quantity=25,
reserved_quantity=5,
vendor_id=test_stock.vendor_id,
vendor_id=test_inventory.vendor_id,
),
Stock(
Inventory(
gtin="1234567890125",
location="LOCATION3",
quantity=0, # Out of stock
quantity=0, # Out of inventory
reserved_quantity=0,
vendor_id=test_stock.vendor_id,
vendor_id=test_inventory.vendor_id,
),
]
db.add_all(additional_stocks)
db.add_all(additional_inventory)
db.commit()
stats = self.service.get_stock_statistics(db)
stats = self.service.get_inventory_statistics(db)
assert "total_stock_entries" in stats
assert "total_inventory_entries" in stats
assert "total_inventory_quantity" in stats
assert stats["total_stock_entries"] >= 3 # test_stock + 2 additional
assert stats["total_inventory_entries"] >= 3 # test_inventory + 2 additional
assert stats["total_inventory_quantity"] >= 35 # 10 + 25 + 0 = 35
def test_get_brands_by_marketplace(self, db):
@@ -495,7 +495,7 @@ class TestStatsService:
assert stats["unique_categories"] == 0
assert stats["unique_marketplaces"] == 0
assert stats["unique_vendors"] == 0
assert stats["total_stock_entries"] == 0
assert stats["total_inventory_entries"] == 0
assert stats["total_inventory_quantity"] == 0
def test_marketplace_breakdown_empty_database(self, db):

505
tests/unit/services/test_stock_service.py
View File

@@ -1,505 +0,0 @@
# tests/test_stock_service.py
import uuid
import pytest
from app.services.stock_service import StockService
from app.exceptions import (
StockNotFoundException,
InsufficientStockException,
InvalidStockOperationException,
StockValidationException,
NegativeStockException,
InvalidQuantityException,
ValidationException,
)
from models.schemas.stock import StockAdd, StockCreate, StockUpdate
from models.database.marketplace_product import MarketplaceProduct
from models.database.stock import Stock
@pytest.mark.unit
@pytest.mark.stock
class TestStockService:
def setup_method(self):
self.service = StockService()
def test_normalize_gtin_invalid(self):
"""Test GTIN normalization with invalid GTINs."""
# Completely invalid values that should return None
assert self.service._normalize_gtin("invalid") is None
assert self.service._normalize_gtin("abcdef") is None
assert self.service._normalize_gtin("") is None
assert self.service._normalize_gtin(None) is None
assert self.service._normalize_gtin(" ") is None # Only whitespace
assert self.service._normalize_gtin("!@#$%") is None # Only special characters
# Mixed invalid characters that become empty after filtering
assert self.service._normalize_gtin("abc-def-ghi") is None # No digits
def test_normalize_gtin_valid(self):
"""Test GTIN normalization with valid GTINs."""
# Test various valid GTIN formats - these should remain unchanged
assert self.service._normalize_gtin("1234567890123") == "1234567890123" # EAN-13
assert self.service._normalize_gtin("123456789012") == "123456789012" # UPC-A
assert self.service._normalize_gtin("12345678") == "12345678" # EAN-8
assert self.service._normalize_gtin("12345678901234") == "12345678901234" # GTIN-14
# Test with decimal points (should be removed)
assert self.service._normalize_gtin("1234567890123.0") == "1234567890123"
# Test with whitespace (should be trimmed)
assert self.service._normalize_gtin(" 1234567890123 ") == "1234567890123"
# Test short GTINs being padded
assert self.service._normalize_gtin("123") == "0000000000123" # Padded to EAN-13
assert self.service._normalize_gtin("12345") == "0000000012345" # Padded to EAN-13
# Test long GTINs being truncated
assert self.service._normalize_gtin("123456789012345") == "3456789012345" # Truncated to 13
def test_normalize_gtin_edge_cases(self):
"""Test GTIN normalization edge cases."""
# Test numeric inputs
assert self.service._normalize_gtin(1234567890123) == "1234567890123"
assert self.service._normalize_gtin(123) == "0000000000123"
# Test mixed valid/invalid characters
assert self.service._normalize_gtin("123-456-789-012") == "123456789012" # Dashes removed
assert self.service._normalize_gtin("123 456 789 012") == "123456789012" # Spaces removed
assert self.service._normalize_gtin("ABC123456789012DEF") == "123456789012" # Letters removed
def test_set_stock_new_entry_success(self, db):
"""Test setting stock for a new GTIN/location combination successfully."""
unique_id = str(uuid.uuid4())[:8]
stock_data = StockCreate(
gtin="1234567890123", location=f"WAREHOUSE_A_{unique_id}", quantity=100
)
result = self.service.set_stock(db, stock_data)
assert result.gtin == "1234567890123"
assert result.location.upper() == f"WAREHOUSE_A_{unique_id}".upper()
assert result.quantity == 100
def test_set_stock_existing_entry_success(self, db, test_stock):
"""Test setting stock for an existing GTIN/location combination successfully."""
stock_data = StockCreate(
gtin=test_stock.gtin,
location=test_stock.location, # Use exact same location as test_stock
quantity=200,
)
result = self.service.set_stock(db, stock_data)
assert result.gtin == test_stock.gtin
assert result.location == test_stock.location
assert result.quantity == 200 # Should replace the original quantity
def test_set_stock_invalid_gtin_validation_error(self, db):
"""Test setting stock with invalid GTIN returns StockValidationException."""
stock_data = StockCreate(
gtin="invalid_gtin", location="WAREHOUSE_A", quantity=100
)
with pytest.raises(StockValidationException) as exc_info:
self.service.set_stock(db, stock_data)
assert exc_info.value.error_code == "STOCK_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
assert exc_info.value.details.get("field") == "gtin"
def test_set_stock_invalid_quantity_error(self, db):
"""Test setting stock with invalid quantity through service validation."""
# Test the service validation directly instead of going through Pydantic schema
# This bypasses the Pydantic validation to test service layer validation
# Create a mock stock data object that bypasses Pydantic validation
class MockStockData:
def __init__(self, gtin, location, quantity):
self.gtin = gtin
self.location = location
self.quantity = quantity
mock_stock_data = MockStockData("1234567890123", "WAREHOUSE_A", -10)
# Test the internal validation method directly
with pytest.raises(InvalidQuantityException) as exc_info:
self.service._validate_quantity(-10, allow_zero=True)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Quantity cannot be negative" in str(exc_info.value)
assert exc_info.value.details.get("quantity") == -10
def test_add_stock_new_entry_success(self, db):
"""Test adding stock for a new GTIN/location combination successfully."""
unique_id = str(uuid.uuid4())[:8]
stock_data = StockAdd(
gtin="1234567890123", location=f"WAREHOUSE_B_{unique_id}", quantity=50
)
result = self.service.add_stock(db, stock_data)
assert result.gtin == "1234567890123"
assert result.location.upper() == f"WAREHOUSE_B_{unique_id}".upper()
assert result.quantity == 50
def test_add_stock_existing_entry_success(self, db, test_stock):
"""Test adding stock to an existing GTIN/location combination successfully."""
original_quantity = test_stock.quantity
stock_data = StockAdd(
gtin=test_stock.gtin,
location=test_stock.location, # Use exact same location as test_stock
quantity=25,
)
result = self.service.add_stock(db, stock_data)
assert result.gtin == test_stock.gtin
assert result.location == test_stock.location
assert result.quantity == original_quantity + 25
def test_add_stock_invalid_gtin_validation_error(self, db):
"""Test adding stock with invalid GTIN returns StockValidationException."""
stock_data = StockAdd(gtin="invalid_gtin", location="WAREHOUSE_A", quantity=50)
with pytest.raises(StockValidationException) as exc_info:
self.service.add_stock(db, stock_data)
assert exc_info.value.error_code == "STOCK_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_add_stock_invalid_quantity_error(self, db):
"""Test adding stock with invalid quantity through service validation."""
# Test zero quantity which should fail for add_stock (doesn't allow zero)
# This tests the service validation: allow_zero=False for add operations
with pytest.raises(InvalidQuantityException) as exc_info:
self.service._validate_quantity(0, allow_zero=False)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Quantity must be positive" in str(exc_info.value)
def test_remove_stock_success(self, db, test_stock):
"""Test removing stock successfully."""
original_quantity = test_stock.quantity
remove_quantity = min(10, original_quantity) # Ensure we don't remove more than available
stock_data = StockAdd(
gtin=test_stock.gtin,
location=test_stock.location, # Use exact same location as test_stock
quantity=remove_quantity,
)
result = self.service.remove_stock(db, stock_data)
assert result.gtin == test_stock.gtin
assert result.location == test_stock.location
assert result.quantity == original_quantity - remove_quantity
def test_remove_stock_insufficient_stock_error(self, db, test_stock):
"""Test removing more stock than available returns InsufficientStockException."""
stock_data = StockAdd(
gtin=test_stock.gtin,
location=test_stock.location, # Use exact same location as test_stock
quantity=test_stock.quantity + 10, # More than available
)
with pytest.raises(InsufficientStockException) as exc_info:
self.service.remove_stock(db, stock_data)
assert exc_info.value.error_code == "INSUFFICIENT_STOCK"
assert exc_info.value.details["gtin"] == test_stock.gtin
assert exc_info.value.details["location"] == test_stock.location
assert exc_info.value.details["requested_quantity"] == test_stock.quantity + 10
assert exc_info.value.details["available_quantity"] == test_stock.quantity
def test_remove_stock_nonexistent_entry_not_found(self, db):
"""Test removing stock from non-existent GTIN/location returns StockNotFoundException."""
unique_id = str(uuid.uuid4())[:8]
stock_data = StockAdd(
gtin="9999999999999", location=f"NONEXISTENT_{unique_id}", quantity=10
)
with pytest.raises(StockNotFoundException) as exc_info:
self.service.remove_stock(db, stock_data)
assert exc_info.value.error_code == "STOCK_NOT_FOUND"
assert "9999999999999" in str(exc_info.value)
assert exc_info.value.details["resource_type"] == "Stock"
def test_remove_stock_invalid_gtin_validation_error(self, db):
"""Test removing stock with invalid GTIN returns StockValidationException."""
stock_data = StockAdd(gtin="invalid_gtin", location="WAREHOUSE_A", quantity=10)
with pytest.raises(StockValidationException) as exc_info:
self.service.remove_stock(db, stock_data)
assert exc_info.value.error_code == "STOCK_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_remove_stock_negative_result_error(self, db, test_stock):
"""Test removing stock that would result in negative quantity returns NegativeStockException."""
# This is handled by InsufficientStockException, but test the logic
stock_data = StockAdd(
gtin=test_stock.gtin,
location=test_stock.location,
quantity=test_stock.quantity + 1, # One more than available
)
with pytest.raises(InsufficientStockException) as exc_info:
self.service.remove_stock(db, stock_data)
# The service prevents negative stock through InsufficientStockException
assert exc_info.value.error_code == "INSUFFICIENT_STOCK"
def test_get_stock_by_gtin_success(self, db, test_stock, test_marketplace_product):
"""Test getting stock summary by GTIN successfully."""
result = self.service.get_stock_by_gtin(db, test_stock.gtin)
assert result.gtin == test_stock.gtin
assert result.total_quantity == test_stock.quantity
assert len(result.locations) == 1
assert result.locations[0].location == test_stock.location
assert result.locations[0].quantity == test_stock.quantity
assert result.product_title == test_marketplace_product.title
def test_get_stock_by_gtin_multiple_locations_success(self, db, test_marketplace_product):
"""Test getting stock summary with multiple locations successfully."""
unique_gtin = test_marketplace_product.gtin
unique_id = str(uuid.uuid4())[:8]
# Create multiple stock entries for the same GTIN with unique locations
stock1 = Stock(gtin=unique_gtin, location=f"WAREHOUSE_A_{unique_id}", quantity=50)
stock2 = Stock(gtin=unique_gtin, location=f"WAREHOUSE_B_{unique_id}", quantity=30)
db.add(stock1)
db.add(stock2)
db.commit()
result = self.service.get_stock_by_gtin(db, unique_gtin)
assert result.gtin == unique_gtin
assert result.total_quantity == 80
assert len(result.locations) == 2
def test_get_stock_by_gtin_not_found_error(self, db):
"""Test getting stock for non-existent GTIN returns StockNotFoundException."""
with pytest.raises(StockNotFoundException) as exc_info:
self.service.get_stock_by_gtin(db, "9999999999999")
assert exc_info.value.error_code == "STOCK_NOT_FOUND"
assert "9999999999999" in str(exc_info.value)
assert exc_info.value.details["resource_type"] == "Stock"
def test_get_stock_by_gtin_invalid_gtin_validation_error(self, db):
"""Test getting stock with invalid GTIN returns StockValidationException."""
with pytest.raises(StockValidationException) as exc_info:
self.service.get_stock_by_gtin(db, "invalid_gtin")
assert exc_info.value.error_code == "STOCK_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_get_total_stock_success(self, db, test_stock, test_marketplace_product):
"""Test getting total stock for a GTIN successfully."""
result = self.service.get_total_stock(db, test_stock.gtin)
assert result["gtin"] == test_stock.gtin
assert result["total_quantity"] == test_stock.quantity
assert result["product_title"] == test_marketplace_product.title
assert result["locations_count"] == 1
def test_get_total_stock_invalid_gtin_validation_error(self, db):
"""Test getting total stock with invalid GTIN returns StockValidationException."""
with pytest.raises(StockValidationException) as exc_info:
self.service.get_total_stock(db, "invalid_gtin")
assert exc_info.value.error_code == "STOCK_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_get_total_stock_not_found_error(self, db):
"""Test getting total stock for non-existent GTIN returns StockNotFoundException."""
with pytest.raises(StockNotFoundException) as exc_info:
self.service.get_total_stock(db, "9999999999999")
assert exc_info.value.error_code == "STOCK_NOT_FOUND"
def test_get_all_stock_no_filters_success(self, db, test_stock):
"""Test getting all stock without filters successfully."""
result = self.service.get_all_stock(db)
assert len(result) >= 1
assert any(stock.gtin == test_stock.gtin for stock in result)
def test_get_all_stock_with_location_filter_success(self, db, test_stock):
"""Test getting all stock with location filter successfully."""
result = self.service.get_all_stock(db, location=test_stock.location)
assert len(result) >= 1
# Check that all returned stocks match the filter (case insensitive)
for stock in result:
assert test_stock.location.upper() in stock.location.upper()
def test_get_all_stock_with_gtin_filter_success(self, db, test_stock):
"""Test getting all stock with GTIN filter successfully."""
result = self.service.get_all_stock(db, gtin=test_stock.gtin)
assert len(result) >= 1
assert all(stock.gtin == test_stock.gtin for stock in result)
def test_get_all_stock_with_pagination_success(self, db):
"""Test getting all stock with pagination successfully."""
unique_prefix = str(uuid.uuid4())[:8]
# Create multiple stock entries with unique GTINs and locations
for i in range(5):
stock = Stock(
gtin=f"1234567890{i:03d}", # Creates valid 13-digit GTINs
location=f"WAREHOUSE_{unique_prefix}_{i}",
quantity=10,
)
db.add(stock)
db.commit()
result = self.service.get_all_stock(db, skip=2, limit=2)
assert len(result) <= 2 # Should be at most 2, might be less if other records exist
def test_update_stock_success(self, db, test_stock):
"""Test updating stock quantity successfully."""
stock_update = StockUpdate(quantity=150)
result = self.service.update_stock(db, test_stock.id, stock_update)
assert result.id == test_stock.id
assert result.quantity == 150
def test_update_stock_not_found_error(self, db):
"""Test updating non-existent stock entry returns StockNotFoundException."""
stock_update = StockUpdate(quantity=150)
with pytest.raises(StockNotFoundException) as exc_info:
self.service.update_stock(db, 99999, stock_update)
assert exc_info.value.error_code == "STOCK_NOT_FOUND"
assert "99999" in str(exc_info.value)
def test_update_stock_invalid_quantity_error(self, db, test_stock):
"""Test updating stock with invalid quantity returns InvalidQuantityException."""
stock_update = StockUpdate(quantity=-10)
with pytest.raises(InvalidQuantityException) as exc_info:
self.service.update_stock(db, test_stock.id, stock_update)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Quantity cannot be negative" in str(exc_info.value)
def test_delete_stock_success(self, db, test_stock):
"""Test deleting stock entry successfully."""
stock_id = test_stock.id
result = self.service.delete_stock(db, stock_id)
assert result is True
# Verify the stock is actually deleted
deleted_stock = db.query(Stock).filter(Stock.id == stock_id).first()
assert deleted_stock is None
def test_delete_stock_not_found_error(self, db):
"""Test deleting non-existent stock entry returns StockNotFoundException."""
with pytest.raises(StockNotFoundException) as exc_info:
self.service.delete_stock(db, 99999)
assert exc_info.value.error_code == "STOCK_NOT_FOUND"
assert "99999" in str(exc_info.value)
def test_get_low_stock_items_success(self, db, test_stock, test_marketplace_product):
"""Test getting low stock items successfully."""
# Set stock to a low value
test_stock.quantity = 5
db.commit()
result = self.service.get_low_stock_items(db, threshold=10)
assert len(result) >= 1
low_stock_item = next((item for item in result if item["gtin"] == test_stock.gtin), None)
assert low_stock_item is not None
assert low_stock_item["current_quantity"] == 5
assert low_stock_item["location"] == test_stock.location
assert low_stock_item["product_title"] == test_marketplace_product.title
def test_get_low_stock_items_invalid_threshold_error(self, db):
"""Test getting low stock items with invalid threshold returns InvalidQuantityException."""
with pytest.raises(InvalidQuantityException) as exc_info:
self.service.get_low_stock_items(db, threshold=-5)
assert exc_info.value.error_code == "INVALID_QUANTITY"
assert "Threshold must be non-negative" in str(exc_info.value)
def test_get_stock_summary_by_location_success(self, db, test_stock):
"""Test getting stock summary by location successfully."""
result = self.service.get_stock_summary_by_location(db, test_stock.location)
assert result["location"] == test_stock.location.upper() # Service normalizes to uppercase
assert result["total_items"] >= 1
assert result["total_quantity"] >= test_stock.quantity
assert result["unique_gtins"] >= 1
def test_get_stock_summary_by_location_empty_result(self, db):
"""Test getting stock summary for location with no stock."""
unique_id = str(uuid.uuid4())[:8]
result = self.service.get_stock_summary_by_location(db, f"EMPTY_LOCATION_{unique_id}")
assert result["total_items"] == 0
assert result["total_quantity"] == 0
assert result["unique_gtins"] == 0
def test_validate_quantity_edge_cases(self, db):
"""Test quantity validation with edge cases."""
# Test zero quantity with allow_zero=True (should succeed)
stock_data = StockCreate(gtin="1234567890123", location="WAREHOUSE_A", quantity=0)
result = self.service.set_stock(db, stock_data)
assert result.quantity == 0
# Test zero quantity with add_stock (should fail - doesn't allow zero)
stock_data_add = StockAdd(gtin="1234567890123", location="WAREHOUSE_B", quantity=0)
with pytest.raises(InvalidQuantityException):
self.service.add_stock(db, stock_data_add)
def test_exception_structure_consistency(self, db):
"""Test that all exceptions follow the consistent LetzShopException structure."""
# Test with a known error case
with pytest.raises(StockNotFoundException) as exc_info:
self.service.get_stock_by_gtin(db, "9999999999999")
exception = exc_info.value
# Verify exception structure matches LetzShopException.to_dict()
assert hasattr(exception, 'error_code')
assert hasattr(exception, 'message')
assert hasattr(exception, 'status_code')
assert hasattr(exception, 'details')
assert isinstance(exception.error_code, str)
assert isinstance(exception.message, str)
assert isinstance(exception.status_code, int)
assert isinstance(exception.details, dict)
@pytest.fixture
def test_product_with_stock(db, test_stock):
"""Create a test product that corresponds to the test stock."""
product = MarketplaceProduct(
marketplace_product_id="STOCK_TEST_001",
title="Stock Test MarketplaceProduct",
gtin=test_stock.gtin,
price="29.99",
brand="TestBrand",
marketplace="Letzshop",
)
db.add(product)
db.commit()
db.refresh(product)
return product

6
tests/unit/services/test_vendor_service.py
View File

@@ -37,7 +37,7 @@ class TestVendorService:
assert vendor is not None
assert vendor.vendor_code == "NEWVENDOR"
assert vendor.owner_id == test_user.id
assert vendor.owner_user_id == test_user.id
assert vendor.is_verified is False # Regular user creates unverified vendor
def test_create_vendor_admin_auto_verify(self, db, test_admin, vendor_factory):
@@ -51,7 +51,7 @@ class TestVendorService:
def test_create_vendor_duplicate_code(self, db, test_user, test_vendor):
"""Test vendor creation fails with duplicate vendor code"""
vendor_data = VendorCreate(
vendor_code=test_vendor.vendor_code, vendor_name=test_vendor.vendor_name
vendor_code=test_vendor.vendor_code, vendor_name=test_vendor.name
)
with pytest.raises(VendorAlreadyExistsException) as exc_info:
@@ -84,7 +84,7 @@ class TestVendorService:
exception = exc_info.value
assert exception.error_code == "INVALID_VENDOR_DATA"
assert exception.details["field"] == "vendor_name"
assert exception.details["field"] == "name"
def test_create_vendor_invalid_code_format(self, db, test_user):
"""Test vendor creation fails with invalid vendor code format"""

2
tests/unit/utils/test_csv_processor.py
View File

@@ -116,7 +116,7 @@ TEST002,Test MarketplaceProduct 2,15.99,TestMarket"""
"title": ["MarketplaceProduct 1", "MarketplaceProduct 2"],
"price": ["10.99", "15.99"],
"marketplace": ["TestMarket", "TestMarket"],
"vendor_name": ["TestVendor", "TestVendor"],
"name": ["TestVendor", "TestVendor"],
}
)
mock_parse.return_value = mock_df