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

test: update tests for multi-language translation support

Browse Source 
- Update marketplace_product_fixtures to create translations
- Update test_marketplace_product.py for translation-based titles
- Update test_product.py for effective property tests
- Update test_order.py to use get_title() method
- Add comprehensive CSV processor tests for translations
- Update stats service tests for new flat response structure
- Fix product schema tests with required marketplace_product_id field
- Add helper function create_marketplace_product_with_translation()

🤖 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-11 17:29:50 +01:00
parent 22c4937779
commit 2ecc2a9785
8 changed files with 790 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,4 +1,6 @@
# tests/test_csv_processor.py
# tests/unit/utils/test_csv_processor.py
"""Unit tests for CSV processor with translation support."""
from unittest.mock import Mock, patch
import pandas as pd
@@ -7,6 +9,8 @@ import requests
import requests.exceptions
from app.utils.csv_processor import CSVProcessor
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
@pytest.mark.unit
@@ -102,6 +106,74 @@ TEST002,Test MarketplaceProduct 2,15.99,TestMarket"""
assert df.iloc[0]["marketplace_product_id"] == "TEST001"
assert df.iloc[1]["price"] == 15.99
def test_normalize_columns_google_shopping(self):
"""Test column normalization for Google Shopping feed format"""
csv_content = """g:id,g:title,g:description,g:price,g:brand,g:product_type
TEST001,Product 1,Description 1,19.99 EUR,Brand1,Category1"""
df = self.processor.parse_csv(csv_content)
df = self.processor.normalize_columns(df)
assert "marketplace_product_id" in df.columns
assert "title" in df.columns
assert "description" in df.columns
assert "product_type_raw" in df.columns # Renamed from product_type
assert df.iloc[0]["marketplace_product_id"] == "TEST001"
assert df.iloc[0]["title"] == "Product 1"
def test_extract_translation_data(self):
"""Test extraction of translation fields from product data"""
product_data = {
"marketplace_product_id": "TEST001",
"title": "Test Product",
"description": "Test Description",
"short_description": "Short desc",
"price": "19.99",
"brand": "TestBrand",
}
translation_data = self.processor._extract_translation_data(product_data)
# Translation fields should be extracted
assert translation_data["title"] == "Test Product"
assert translation_data["description"] == "Test Description"
assert translation_data["short_description"] == "Short desc"
# Product data should no longer have translation fields
assert "title" not in product_data
assert "description" not in product_data
assert "short_description" not in product_data
# Non-translation fields should remain
assert product_data["marketplace_product_id"] == "TEST001"
assert product_data["price"] == "19.99"
assert product_data["brand"] == "TestBrand"
def test_parse_price_to_numeric(self):
"""Test price string to numeric conversion"""
assert self.processor._parse_price_to_numeric("19.99 EUR") == 19.99
assert self.processor._parse_price_to_numeric("19,99 EUR") == 19.99
assert self.processor._parse_price_to_numeric("$29.99") == 29.99
assert self.processor._parse_price_to_numeric("100") == 100.0
assert self.processor._parse_price_to_numeric(None) is None
assert self.processor._parse_price_to_numeric("") is None
def test_clean_row_data_with_prices(self):
"""Test row data cleaning with price parsing"""
row_data = {
"marketplace_product_id": "TEST001",
"title": "Test Product",
"price": "19.99 EUR",
"sale_price": "14.99 EUR",
"gtin": "1234567890123",
}
cleaned = self.processor._clean_row_data(row_data)
assert cleaned["price_numeric"] == 19.99
assert cleaned["sale_price_numeric"] == 14.99
assert cleaned["currency"] == "EUR"
@pytest.mark.asyncio
async def test_process_marketplace_csv_from_url(self, db):
"""Test complete marketplace CSV processing"""
@@ -117,15 +189,275 @@ TEST002,Test MarketplaceProduct 2,15.99,TestMarket"""
"title": ["MarketplaceProduct 1", "MarketplaceProduct 2"],
"price": ["10.99", "15.99"],
"marketplace": ["TestMarket", "TestMarket"],
"name": ["TestVendor", "TestVendor"],
"vendor_name": ["TestVendor", "TestVendor"],
}
)
mock_parse.return_value = mock_df
result = await self.processor.process_marketplace_csv_from_url(
"http://example.com/test.csv", "TestMarket", "TestVendor", 1000, db
"http://example.com/test.csv",
"TestMarket",
"TestVendor",
1000,
db,
language="en",
)
assert "imported" in result
assert "updated" in result
assert "total_processed" in result
assert "language" in result
assert result["language"] == "en"
@pytest.mark.asyncio
async def test_process_batch_creates_translations(self, db):
"""Test that batch processing creates translation records"""
# Clean up any existing test data
existing = (
db.query(MarketplaceProduct)
.filter(
MarketplaceProduct.marketplace_product_id.in_(
["TRANS_TEST_001", "TRANS_TEST_002"]
)
)
.all()
)
for p in existing:
db.delete(p)
db.commit()
# Create test DataFrame
batch_df = pd.DataFrame(
{
"marketplace_product_id": ["TRANS_TEST_001", "TRANS_TEST_002"],
"title": ["Product One", "Product Two"],
"description": ["Description One", "Description Two"],
"price": ["19.99 EUR", "29.99 EUR"],
"brand": ["Brand1", "Brand2"],
}
)
result = await self.processor._process_marketplace_batch(
batch_df,
"TestMarket",
"TestVendor",
db,
batch_num=1,
language="en",
source_file="test.csv",
)
assert result["imported"] == 2
assert result["errors"] == 0
# Verify products were created
products = (
db.query(MarketplaceProduct)
.filter(
MarketplaceProduct.marketplace_product_id.in_(
["TRANS_TEST_001", "TRANS_TEST_002"]
)
)
.all()
)
assert len(products) == 2
# Verify translations were created
for product in products:
assert len(product.translations) == 1
translation = product.translations[0]
assert translation.language == "en"
assert translation.title is not None
assert translation.source_file == "test.csv"
# Verify get_title method works
product1 = next(
p for p in products if p.marketplace_product_id == "TRANS_TEST_001"
)
assert product1.get_title("en") == "Product One"
assert product1.get_description("en") == "Description One"
# Clean up
for p in products:
db.delete(p)
db.commit()
@pytest.mark.asyncio
async def test_process_batch_updates_existing_translations(self, db):
"""Test that batch processing updates existing translation records"""
# Clean up any existing test data
existing = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == "UPDATE_TEST_001")
.first()
)
if existing:
db.delete(existing)
db.commit()
# Create initial product with translation
batch_df = pd.DataFrame(
{
"marketplace_product_id": ["UPDATE_TEST_001"],
"title": ["Original Title"],
"description": ["Original Description"],
"price": ["19.99 EUR"],
}
)
await self.processor._process_marketplace_batch(
batch_df, "TestMarket", "TestVendor", db, 1, language="en"
)
# Update with new data
update_df = pd.DataFrame(
{
"marketplace_product_id": ["UPDATE_TEST_001"],
"title": ["Updated Title"],
"description": ["Updated Description"],
"price": ["24.99 EUR"],
}
)
result = await self.processor._process_marketplace_batch(
update_df, "TestMarket", "TestVendor", db, 1, language="en"
)
assert result["updated"] == 1
assert result["imported"] == 0
# Verify translation was updated
product = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == "UPDATE_TEST_001")
.first()
)
assert product.get_title("en") == "Updated Title"
assert product.get_description("en") == "Updated Description"
# Clean up
db.delete(product)
db.commit()
@pytest.mark.asyncio
async def test_process_batch_multi_language(self, db):
"""Test importing same product in multiple languages"""
# Clean up
existing = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == "MULTI_LANG_001")
.first()
)
if existing:
db.delete(existing)
db.commit()
# Import English version
en_df = pd.DataFrame(
{
"marketplace_product_id": ["MULTI_LANG_001"],
"title": ["English Title"],
"description": ["English Description"],
"price": ["19.99 EUR"],
"brand": ["TestBrand"],
}
)
await self.processor._process_marketplace_batch(
en_df, "TestMarket", "TestVendor", db, 1, language="en"
)
# Import French version (same product, different language)
fr_df = pd.DataFrame(
{
"marketplace_product_id": ["MULTI_LANG_001"],
"title": ["Titre Français"],
"description": ["Description Française"],
"price": ["19.99 EUR"],
"brand": ["TestBrand"],
}
)
result = await self.processor._process_marketplace_batch(
fr_df, "TestMarket", "TestVendor", db, 1, language="fr"
)
assert result["updated"] == 1 # Product existed, so it's an update
# Verify both translations exist
product = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == "MULTI_LANG_001")
.first()
)
assert len(product.translations) == 2
# Verify each language
assert product.get_title("en") == "English Title"
assert product.get_title("fr") == "Titre Français"
assert product.get_description("en") == "English Description"
assert product.get_description("fr") == "Description Française"
# Test fallback to English for unknown language
assert product.get_title("de") == "English Title"
# Clean up
db.delete(product)
db.commit()
@pytest.mark.asyncio
async def test_process_batch_skips_missing_title(self, db):
"""Test that rows without title are skipped"""
batch_df = pd.DataFrame(
{
"marketplace_product_id": ["NO_TITLE_001", "HAS_TITLE_001"],
"title": [None, "Has Title"],
"price": ["19.99", "29.99"],
}
)
result = await self.processor._process_marketplace_batch(
batch_df, "TestMarket", "TestVendor", db, 1, language="en"
)
assert result["imported"] == 1
assert result["errors"] == 1 # Missing title is an error
# Clean up
product = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == "HAS_TITLE_001")
.first()
)
if product:
db.delete(product)
db.commit()
@pytest.mark.asyncio
async def test_process_batch_skips_missing_product_id(self, db):
"""Test that rows without marketplace_product_id are skipped"""
batch_df = pd.DataFrame(
{
"marketplace_product_id": [None, "HAS_ID_001"],
"title": ["No ID Product", "Has ID Product"],
"price": ["19.99", "29.99"],
}
)
result = await self.processor._process_marketplace_batch(
batch_df, "TestMarket", "TestVendor", db, 1, language="en"
)
assert result["imported"] == 1
assert result["errors"] == 1 # Missing ID is an error
# Clean up
product = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == "HAS_ID_001")
.first()
)
if product:
db.delete(product)
db.commit()