Samir Boulahtit
9920430b9e
- Remove |safe from |tojson in HTML attributes (x-data) - quotes must become " for browsers to parse correctly - Update LANG-002 and LANG-003 architecture rules to document correct |tojson usage patterns: - HTML attributes: |tojson (no |safe) - Script blocks: |tojson|safe - Fix validator to warn when |tojson|safe is used in x-data (breaks HTML attribute parsing) - Improve code quality across services, APIs, and tests 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
508 lines
20 KiB
Python
508 lines
20 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",
|
|
)
|
|
|
|
# Title is passed as separate parameter for translation table
|
|
product = self.service.create_product(
|
|
db, product_data, title="Service Test MarketplaceProduct"
|
|
)
|
|
|
|
assert product.marketplace_product_id == "SVC001"
|
|
assert product.get_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_without_title(self, db):
|
|
"""Test product creation without title succeeds (title is optional, stored in translations)"""
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id="SVC003",
|
|
title="", # Empty title - allowed since translations are optional
|
|
price="19.99",
|
|
)
|
|
|
|
product = self.service.create_product(db, product_data)
|
|
|
|
# Product is created but title returns None since no translation
|
|
assert product.marketplace_product_id == "SVC003"
|
|
assert product.get_title() is None # No translation created for empty title
|
|
|
|
def test_create_product_already_exists(self, db, test_marketplace_product):
|
|
"""Test creating product with existing ID raises MarketplaceProductAlreadyExistsException"""
|
|
# Store the product ID before the exception (session may be invalid after)
|
|
existing_product_id = test_marketplace_product.marketplace_product_id
|
|
|
|
product_data = MarketplaceProductCreate(
|
|
marketplace_product_id=existing_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)
|
|
|
|
# Rollback to clear the session's invalid state from IntegrityError
|
|
db.rollback()
|
|
|
|
assert exc_info.value.error_code == "PRODUCT_ALREADY_EXISTS"
|
|
assert existing_product_id in str(exc_info.value)
|
|
assert exc_info.value.status_code == 409
|
|
assert (
|
|
exc_info.value.details.get("marketplace_product_id") == existing_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.get_title() == test_marketplace_product.get_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(price="39.99")
|
|
|
|
# Title is passed as separate parameter for translation table
|
|
updated_product = self.service.update_product(
|
|
db,
|
|
test_marketplace_product.marketplace_product_id,
|
|
update_data,
|
|
title="Updated MarketplaceProduct Title",
|
|
)
|
|
|
|
assert updated_product.get_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_preserves_existing(
|
|
self, db, test_marketplace_product
|
|
):
|
|
"""Test updating product with empty title preserves existing title in translation"""
|
|
original_title = test_marketplace_product.get_title()
|
|
update_data = MarketplaceProductUpdate(title="")
|
|
|
|
updated_product = self.service.update_product(
|
|
db, test_marketplace_product.marketplace_product_id, update_data
|
|
)
|
|
|
|
# Empty title update preserves existing translation title
|
|
assert updated_product.get_title() == original_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
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.products
|
|
class TestMarketplaceProductServiceAdmin:
|
|
"""Tests for admin-specific methods in MarketplaceProductService."""
|
|
|
|
def setup_method(self):
|
|
self.service = MarketplaceProductService()
|
|
|
|
def test_get_admin_products_success(self, db, test_marketplace_product):
|
|
"""Test getting admin products list."""
|
|
products, total = self.service.get_admin_products(db)
|
|
|
|
assert total >= 1
|
|
assert len(products) >= 1
|
|
|
|
# Find our test product in results
|
|
found = False
|
|
for p in products:
|
|
if (
|
|
p["marketplace_product_id"]
|
|
== test_marketplace_product.marketplace_product_id
|
|
):
|
|
found = True
|
|
assert p["id"] == test_marketplace_product.id
|
|
assert p["marketplace"] == test_marketplace_product.marketplace
|
|
break
|
|
|
|
assert found, "Test product not found in results"
|
|
|
|
def test_get_admin_products_with_search(self, db, test_marketplace_product):
|
|
"""Test getting admin products with search filter."""
|
|
products, total = self.service.get_admin_products(
|
|
db, search="Test MarketplaceProduct"
|
|
)
|
|
|
|
assert total >= 1
|
|
# Should find our test product
|
|
product_ids = [p["marketplace_product_id"] for p in products]
|
|
assert test_marketplace_product.marketplace_product_id in product_ids
|
|
|
|
def test_get_admin_products_with_marketplace_filter(
|
|
self, db, test_marketplace_product
|
|
):
|
|
"""Test getting admin products with marketplace filter."""
|
|
products, total = self.service.get_admin_products(
|
|
db, marketplace=test_marketplace_product.marketplace
|
|
)
|
|
|
|
assert total >= 1
|
|
# All products should be from the filtered marketplace
|
|
for p in products:
|
|
assert p["marketplace"] == test_marketplace_product.marketplace
|
|
|
|
def test_get_admin_products_pagination(self, db, multiple_products):
|
|
"""Test admin products pagination."""
|
|
# Get first 2
|
|
products, total = self.service.get_admin_products(db, skip=0, limit=2)
|
|
|
|
assert total >= 5 # We created 5 products
|
|
assert len(products) == 2
|
|
|
|
# Get next 2
|
|
products2, _ = self.service.get_admin_products(db, skip=2, limit=2)
|
|
assert len(products2) == 2
|
|
|
|
# Make sure they're different
|
|
ids1 = {p["id"] for p in products}
|
|
ids2 = {p["id"] for p in products2}
|
|
assert ids1.isdisjoint(ids2)
|
|
|
|
def test_get_admin_product_stats(self, db, test_marketplace_product):
|
|
"""Test getting admin product statistics."""
|
|
stats = self.service.get_admin_product_stats(db)
|
|
|
|
assert "total" in stats
|
|
assert "active" in stats
|
|
assert "inactive" in stats
|
|
assert "digital" in stats
|
|
assert "physical" in stats
|
|
assert "by_marketplace" in stats
|
|
assert stats["total"] >= 1
|
|
|
|
def test_get_marketplaces_list(self, db, test_marketplace_product):
|
|
"""Test getting list of marketplaces."""
|
|
marketplaces = self.service.get_marketplaces_list(db)
|
|
|
|
assert isinstance(marketplaces, list)
|
|
assert test_marketplace_product.marketplace in marketplaces
|
|
|
|
def test_get_source_vendors_list(self, db, test_marketplace_product):
|
|
"""Test getting list of source vendors."""
|
|
vendors = self.service.get_source_vendors_list(db)
|
|
|
|
assert isinstance(vendors, list)
|
|
assert test_marketplace_product.vendor_name in vendors
|
|
|
|
def test_get_admin_product_detail(self, db, test_marketplace_product):
|
|
"""Test getting admin product detail by ID."""
|
|
product = self.service.get_admin_product_detail(db, test_marketplace_product.id)
|
|
|
|
assert product["id"] == test_marketplace_product.id
|
|
assert (
|
|
product["marketplace_product_id"]
|
|
== test_marketplace_product.marketplace_product_id
|
|
)
|
|
assert product["marketplace"] == test_marketplace_product.marketplace
|
|
assert "translations" in product
|
|
|
|
def test_get_admin_product_detail_not_found(self, db):
|
|
"""Test getting non-existent product detail raises exception."""
|
|
with pytest.raises(MarketplaceProductNotFoundException):
|
|
self.service.get_admin_product_detail(db, 99999)
|
|
|
|
def test_copy_to_vendor_catalog_success(
|
|
self, db, test_marketplace_product, test_vendor
|
|
):
|
|
"""Test copying products to vendor catalog."""
|
|
result = self.service.copy_to_vendor_catalog(
|
|
db,
|
|
marketplace_product_ids=[test_marketplace_product.id],
|
|
vendor_id=test_vendor.id,
|
|
)
|
|
|
|
assert result["copied"] == 1
|
|
assert result["skipped"] == 0
|
|
assert result["failed"] == 0
|
|
|
|
def test_copy_to_vendor_catalog_skip_existing(
|
|
self, db, test_marketplace_product, test_vendor
|
|
):
|
|
"""Test copying products that already exist skips them."""
|
|
# First copy
|
|
result1 = self.service.copy_to_vendor_catalog(
|
|
db,
|
|
marketplace_product_ids=[test_marketplace_product.id],
|
|
vendor_id=test_vendor.id,
|
|
)
|
|
assert result1["copied"] == 1
|
|
|
|
# Second copy should skip
|
|
result2 = self.service.copy_to_vendor_catalog(
|
|
db,
|
|
marketplace_product_ids=[test_marketplace_product.id],
|
|
vendor_id=test_vendor.id,
|
|
skip_existing=True,
|
|
)
|
|
assert result2["copied"] == 0
|
|
assert result2["skipped"] == 1
|
|
|
|
def test_copy_to_vendor_catalog_invalid_vendor(self, db, test_marketplace_product):
|
|
"""Test copying to non-existent vendor raises exception."""
|
|
from app.exceptions import VendorNotFoundException
|
|
|
|
with pytest.raises(VendorNotFoundException):
|
|
self.service.copy_to_vendor_catalog(
|
|
db,
|
|
marketplace_product_ids=[test_marketplace_product.id],
|
|
vendor_id=99999,
|
|
)
|
|
|
|
def test_copy_to_vendor_catalog_invalid_products(self, db, test_vendor):
|
|
"""Test copying non-existent products raises exception."""
|
|
with pytest.raises(MarketplaceProductNotFoundException):
|
|
self.service.copy_to_vendor_catalog(
|
|
db,
|
|
marketplace_product_ids=[99999],
|
|
vendor_id=test_vendor.id,
|
|
)
|