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

feat: add marketplace products admin UI with copy-to-vendor functionality

Browse Source 
- Add admin marketplace products page to browse imported products
- Add admin vendor products page to manage vendor catalog
- Add product detail pages for both marketplace and vendor products
- Implement copy-to-vendor API to copy marketplace products to vendor catalogs
- Add vendor product service with CRUD operations
- Update sidebar navigation with new product management links
- Add integration and unit tests for new endpoints and services

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-12-12 22:36:04 +01:00
parent 89c98cb645
commit 9c60989f1d
28 changed files with 4575 additions and 1414 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,10 +30,13 @@ class TestProductService:
marketplace="TestMarket",
)
product = self.service.create_product(db, product_data)
# 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.title == "Service Test MarketplaceProduct"
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
@@ -70,20 +73,19 @@ class TestProductService:
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"""
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
title="", # Empty title - allowed since translations are optional
price="19.99",
)
with pytest.raises(MarketplaceProductValidationException) as exc_info:
self.service.create_product(db, product_data)
product = 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"
# 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"""
@@ -135,7 +137,7 @@ class TestProductService:
product.marketplace_product_id
== test_marketplace_product.marketplace_product_id
)
assert product.title == test_marketplace_product.title
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"""
@@ -180,15 +182,17 @@ class TestProductService:
def test_update_product_success(self, db, test_marketplace_product):
"""Test successful product update"""
update_data = MarketplaceProductUpdate(
title="Updated MarketplaceProduct Title", price="39.99"
)
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
db,
test_marketplace_product.marketplace_product_id,
update_data,
title="Updated MarketplaceProduct Title",
)
assert updated_product.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
@@ -220,18 +224,17 @@ class TestProductService:
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"""
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="")
with pytest.raises(MarketplaceProductValidationException) as exc_info:
self.service.update_product(
db, test_marketplace_product.marketplace_product_id, update_data
)
updated_product = 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"
# 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"""
@@ -329,3 +332,172 @@ class TestProductService:
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,
)

126
tests/unit/services/test_vendor_product_service.py Normal file
View File

@@ -0,0 +1,126 @@
# tests/unit/services/test_vendor_product_service.py
"""
Unit tests for VendorProductService.
Tests the vendor product catalog service operations.
"""
import pytest
from app.exceptions import ProductNotFoundException
from app.services.vendor_product_service import VendorProductService
@pytest.mark.unit
@pytest.mark.products
class TestVendorProductService:
"""Tests for VendorProductService."""
def setup_method(self):
self.service = VendorProductService()
def test_get_products_success(self, db, test_product):
"""Test getting vendor products list."""
products, total = self.service.get_products(db)
assert total >= 1
assert len(products) >= 1
# Find our test product in results
found = False
for p in products:
if p["id"] == test_product.id:
found = True
assert p["vendor_id"] == test_product.vendor_id
assert p["marketplace_product_id"] == test_product.marketplace_product_id
break
assert found, "Test product not found in results"
def test_get_products_with_vendor_filter(self, db, test_product, test_vendor):
"""Test getting products filtered by vendor."""
products, total = self.service.get_products(db, vendor_id=test_vendor.id)
assert total >= 1
# All products should be from the filtered vendor
for p in products:
assert p["vendor_id"] == test_vendor.id
def test_get_products_with_active_filter(self, db, test_product):
"""Test getting products filtered by active status."""
products, total = self.service.get_products(db, is_active=True)
# All products should be active
for p in products:
assert p["is_active"] is True
def test_get_products_with_featured_filter(self, db, test_product):
"""Test getting products filtered by featured status."""
products, total = self.service.get_products(db, is_featured=False)
# All products should not be featured
for p in products:
assert p["is_featured"] is False
def test_get_products_pagination(self, db, test_product):
"""Test vendor products pagination."""
products, total = self.service.get_products(db, skip=0, limit=10)
assert total >= 1
assert len(products) <= 10
def test_get_product_stats_success(self, db, test_product):
"""Test getting vendor product statistics."""
stats = self.service.get_product_stats(db)
assert "total" in stats
assert "active" in stats
assert "inactive" in stats
assert "featured" in stats
assert "digital" in stats
assert "physical" in stats
assert "by_vendor" in stats
assert stats["total"] >= 1
def test_get_catalog_vendors_success(self, db, test_product, test_vendor):
"""Test getting list of vendors with products."""
vendors = self.service.get_catalog_vendors(db)
assert isinstance(vendors, list)
assert len(vendors) >= 1
# Check that test_vendor is in the list
vendor_ids = [v["id"] for v in vendors]
assert test_vendor.id in vendor_ids
def test_get_product_detail_success(self, db, test_product):
"""Test getting vendor product detail."""
product = self.service.get_product_detail(db, test_product.id)
assert product["id"] == test_product.id
assert product["vendor_id"] == test_product.vendor_id
assert product["marketplace_product_id"] == test_product.marketplace_product_id
assert "source_marketplace" in product
assert "source_vendor" in product
def test_get_product_detail_not_found(self, db):
"""Test getting non-existent product raises exception."""
with pytest.raises(ProductNotFoundException):
self.service.get_product_detail(db, 99999)
def test_remove_product_success(self, db, test_product):
"""Test removing product from vendor catalog."""
product_id = test_product.id
result = self.service.remove_product(db, product_id)
assert "message" in result
assert "removed" in result["message"].lower()
# Verify product is removed
with pytest.raises(ProductNotFoundException):
self.service.get_product_detail(db, product_id)
def test_remove_product_not_found(self, db):
"""Test removing non-existent product raises exception."""
with pytest.raises(ProductNotFoundException):
self.service.remove_product(db, 99999)