Samir Boulahtit
120d8196fe
Updates to work with refactored fixtures (no expunge): - Re-query entities when modifying state in tests - Remove assertions on expunged object properties Auth service tests: - Update to use email_or_username field instead of username Admin service tests: - Fix statistics test to use stats_service module - Remove vendor_name filter test (field removed) - Update import job assertions Inventory/Marketplace/Stats/Vendor service tests: - Refactor to work with attached session objects - Update assertions for changed response formats - Improve test isolation and cleanup 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
326 lines
14 KiB
Python
326 lines
14 KiB
Python
# tests/test_product_service.py
|
|
import pytest
|
|
|
|
from app.exceptions import (
|
|
InvalidMarketplaceProductDataException,
|
|
MarketplaceProductAlreadyExistsException,
|
|
MarketplaceProductNotFoundException,
|
|
MarketplaceProductValidationException,
|
|
)
|
|
from app.services.marketplace_product_service import MarketplaceProductService
|
|
from models.schema.marketplace_product import (
|
|
MarketplaceProductCreate,
|
|
MarketplaceProductUpdate,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.products
|
|
class TestProductService:
|
|
def setup_method(self):
|
|
self.service = MarketplaceProductService()
|
|
|
|
def test_create_product_success(self, db):
|
|
"""Test successful product creation with valid data"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id="SVC001",
|
|
title="Service Test MarketplaceProduct",
|
|
gtin="1234567890123",
|
|
price="19.99",
|
|
marketplace="TestMarket",
|
|
)
|
|
|
|
product = self.service.create_product(db, product_data)
|
|
|
|
assert product.marketplace_product_id == "SVC001"
|
|
assert product.title == "Service Test MarketplaceProduct"
|
|
assert product.gtin == "1234567890123"
|
|
assert product.marketplace == "TestMarket"
|
|
assert product.price == "19.99" # Price is stored as string after processing
|
|
|
|
def test_create_product_invalid_gtin(self, db):
|
|
"""Test product creation with invalid GTIN raises InvalidMarketplaceProductDataException"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id="SVC002",
|
|
title="Service Test MarketplaceProduct",
|
|
gtin="invalid_gtin",
|
|
price="19.99",
|
|
)
|
|
|
|
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
|
|
self.service.create_product(db, product_data)
|
|
|
|
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
|
|
assert "Invalid GTIN format" in str(exc_info.value)
|
|
assert exc_info.value.status_code == 422
|
|
assert exc_info.value.details.get("field") == "gtin"
|
|
|
|
def test_create_product_missing_product_id(self, db):
|
|
"""Test product creation without marketplace_product_id raises MarketplaceProductValidationException"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id="", # Empty product ID
|
|
title="Service Test MarketplaceProduct",
|
|
price="19.99",
|
|
)
|
|
|
|
with pytest.raises(MarketplaceProductValidationException) as exc_info:
|
|
self.service.create_product(db, product_data)
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED"
|
|
assert "MarketplaceProduct ID is required" in str(exc_info.value)
|
|
assert exc_info.value.details.get("field") == "marketplace_product_id"
|
|
|
|
def test_create_product_missing_title(self, db):
|
|
"""Test product creation without title raises MarketplaceProductValidationException"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id="SVC003",
|
|
title="", # Empty title
|
|
price="19.99",
|
|
)
|
|
|
|
with pytest.raises(MarketplaceProductValidationException) as exc_info:
|
|
self.service.create_product(db, product_data)
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED"
|
|
assert "MarketplaceProduct title is required" in str(exc_info.value)
|
|
assert exc_info.value.details.get("field") == "title"
|
|
|
|
def test_create_product_already_exists(self, db, test_marketplace_product):
|
|
"""Test creating product with existing ID raises MarketplaceProductAlreadyExistsException"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id=test_marketplace_product.marketplace_product_id, # Use existing product ID
|
|
title="Duplicate MarketplaceProduct",
|
|
price="29.99",
|
|
)
|
|
|
|
with pytest.raises(MarketplaceProductAlreadyExistsException) as exc_info:
|
|
self.service.create_product(db, product_data)
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_ALREADY_EXISTS"
|
|
assert test_marketplace_product.marketplace_product_id in str(exc_info.value)
|
|
assert exc_info.value.status_code == 409
|
|
assert (
|
|
exc_info.value.details.get("marketplace_product_id")
|
|
== test_marketplace_product.marketplace_product_id
|
|
)
|
|
|
|
def test_create_product_invalid_price(self, db):
|
|
"""Test product creation with invalid price raises InvalidMarketplaceProductDataException"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id="SVC004",
|
|
title="Service Test MarketplaceProduct",
|
|
price="invalid_price",
|
|
)
|
|
|
|
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
|
|
self.service.create_product(db, product_data)
|
|
|
|
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
|
|
assert "Invalid price format" in str(exc_info.value)
|
|
assert exc_info.value.details.get("field") == "price"
|
|
|
|
def test_get_product_by_id_or_raise_success(self, db, test_marketplace_product):
|
|
"""Test successful product retrieval by ID"""
|
|
product = self.service.get_product_by_id_or_raise(
|
|
db, test_marketplace_product.marketplace_product_id
|
|
)
|
|
|
|
assert (
|
|
product.marketplace_product_id
|
|
== test_marketplace_product.marketplace_product_id
|
|
)
|
|
assert product.title == test_marketplace_product.title
|
|
|
|
def test_get_product_by_id_or_raise_not_found(self, db):
|
|
"""Test product retrieval with non-existent ID raises MarketplaceProductNotFoundException"""
|
|
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
|
|
self.service.get_product_by_id_or_raise(db, "NONEXISTENT")
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
|
|
assert "NONEXISTENT" in str(exc_info.value)
|
|
assert exc_info.value.status_code == 404
|
|
assert exc_info.value.details.get("resource_type") == "MarketplaceProduct"
|
|
assert exc_info.value.details.get("identifier") == "NONEXISTENT"
|
|
|
|
def test_get_products_with_filters_success(self, db, test_marketplace_product):
|
|
"""Test getting products with various filters"""
|
|
products, total = self.service.get_products_with_filters(
|
|
db, brand=test_marketplace_product.brand
|
|
)
|
|
|
|
assert total == 1
|
|
assert len(products) == 1
|
|
assert products[0].brand == test_marketplace_product.brand
|
|
|
|
def test_get_products_with_search(self, db, test_marketplace_product):
|
|
"""Test getting products with search term"""
|
|
products, total = self.service.get_products_with_filters(
|
|
db, search="Test MarketplaceProduct"
|
|
)
|
|
|
|
assert total >= 1
|
|
assert len(products) >= 1
|
|
# Verify search worked by checking that title contains search term
|
|
found_product = next(
|
|
(
|
|
p
|
|
for p in products
|
|
if p.marketplace_product_id
|
|
== test_marketplace_product.marketplace_product_id
|
|
),
|
|
None,
|
|
)
|
|
assert found_product is not None
|
|
|
|
def test_update_product_success(self, db, test_marketplace_product):
|
|
"""Test successful product update"""
|
|
update_data = MarketplaceProductUpdate(
|
|
title="Updated MarketplaceProduct Title", price="39.99"
|
|
)
|
|
|
|
updated_product = self.service.update_product(
|
|
db, test_marketplace_product.marketplace_product_id, update_data
|
|
)
|
|
|
|
assert updated_product.title == "Updated MarketplaceProduct Title"
|
|
assert (
|
|
updated_product.price == "39.99"
|
|
) # Price is stored as string after processing
|
|
assert (
|
|
updated_product.marketplace_product_id
|
|
== test_marketplace_product.marketplace_product_id
|
|
) # ID unchanged
|
|
|
|
def test_update_product_not_found(self, db):
|
|
"""Test updating non-existent product raises MarketplaceProductNotFoundException"""
|
|
update_data = MarketplaceProductUpdate(title="Updated Title")
|
|
|
|
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
|
|
self.service.update_product(db, "NONEXISTENT", update_data)
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
|
|
assert "NONEXISTENT" in str(exc_info.value)
|
|
|
|
def test_update_product_invalid_gtin(self, db, test_marketplace_product):
|
|
"""Test updating product with invalid GTIN raises InvalidMarketplaceProductDataException"""
|
|
update_data = MarketplaceProductUpdate(gtin="invalid_gtin")
|
|
|
|
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
|
|
self.service.update_product(
|
|
db, test_marketplace_product.marketplace_product_id, update_data
|
|
)
|
|
|
|
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
|
|
assert "Invalid GTIN format" in str(exc_info.value)
|
|
assert exc_info.value.details.get("field") == "gtin"
|
|
|
|
def test_update_product_empty_title(self, db, test_marketplace_product):
|
|
"""Test updating product with empty title raises MarketplaceProductValidationException"""
|
|
update_data = MarketplaceProductUpdate(title="")
|
|
|
|
with pytest.raises(MarketplaceProductValidationException) as exc_info:
|
|
self.service.update_product(
|
|
db, test_marketplace_product.marketplace_product_id, update_data
|
|
)
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED"
|
|
assert "MarketplaceProduct title cannot be empty" in str(exc_info.value)
|
|
assert exc_info.value.details.get("field") == "title"
|
|
|
|
def test_update_product_invalid_price(self, db, test_marketplace_product):
|
|
"""Test updating product with invalid price raises InvalidMarketplaceProductDataException"""
|
|
update_data = MarketplaceProductUpdate(price="invalid_price")
|
|
|
|
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
|
|
self.service.update_product(
|
|
db, test_marketplace_product.marketplace_product_id, update_data
|
|
)
|
|
|
|
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
|
|
assert "Invalid price format" in str(exc_info.value)
|
|
assert exc_info.value.details.get("field") == "price"
|
|
|
|
def test_delete_product_success(self, db, test_marketplace_product):
|
|
"""Test successful product deletion"""
|
|
result = self.service.delete_product(
|
|
db, test_marketplace_product.marketplace_product_id
|
|
)
|
|
|
|
assert result is True
|
|
|
|
# Verify product is deleted
|
|
deleted_product = self.service.get_product_by_id(
|
|
db, test_marketplace_product.marketplace_product_id
|
|
)
|
|
assert deleted_product is None
|
|
|
|
def test_delete_product_not_found(self, db):
|
|
"""Test deleting non-existent product raises MarketplaceProductNotFoundException"""
|
|
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
|
|
self.service.delete_product(db, "NONEXISTENT")
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
|
|
assert "NONEXISTENT" in str(exc_info.value)
|
|
|
|
def test_get_inventory_info_success(
|
|
self, db, test_marketplace_product_with_inventory
|
|
):
|
|
"""Test getting inventory info for product with inventory."""
|
|
marketplace_product = test_marketplace_product_with_inventory["marketplace_product"]
|
|
inventory = test_marketplace_product_with_inventory["inventory"]
|
|
|
|
inventory_info = self.service.get_inventory_info(db, marketplace_product.gtin)
|
|
|
|
assert inventory_info is not None
|
|
assert inventory_info.total_quantity == inventory.quantity
|
|
assert len(inventory_info.locations) >= 1
|
|
|
|
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 inventory_info is None
|
|
|
|
def test_product_exists_true(self, db, test_marketplace_product):
|
|
"""Test product_exists returns True for existing product"""
|
|
exists = self.service.product_exists(
|
|
db, test_marketplace_product.marketplace_product_id
|
|
)
|
|
assert exists is True
|
|
|
|
def test_product_exists_false(self, db):
|
|
"""Test product_exists returns False for non-existent product"""
|
|
exists = self.service.product_exists(db, "NONEXISTENT")
|
|
assert exists is False
|
|
|
|
def test_generate_csv_export_success(self, db, test_marketplace_product):
|
|
"""Test CSV export generation"""
|
|
csv_generator = self.service.generate_csv_export(db)
|
|
|
|
# Convert generator to list to test content
|
|
csv_lines = list(csv_generator)
|
|
|
|
assert len(csv_lines) > 1 # Header + at least one data row
|
|
assert csv_lines[0].startswith(
|
|
"marketplace_product_id,title,description"
|
|
) # Check header
|
|
|
|
# Check that test product appears in CSV
|
|
csv_content = "".join(csv_lines)
|
|
assert test_marketplace_product.marketplace_product_id in csv_content
|
|
|
|
def test_generate_csv_export_with_filters(self, db, test_marketplace_product):
|
|
"""Test CSV export with marketplace filter"""
|
|
csv_generator = self.service.generate_csv_export(
|
|
db, marketplace=test_marketplace_product.marketplace
|
|
)
|
|
|
|
csv_lines = list(csv_generator)
|
|
assert len(csv_lines) >= 1 # At least header
|
|
|
|
if len(csv_lines) > 1: # If there's data
|
|
csv_content = "".join(csv_lines)
|
|
assert test_marketplace_product.marketplace in csv_content
|