feat: add Letzshop CSV product export for admin and vendor
- Create letzshop_export_service.py with CSV generation logic
- Add admin endpoint: GET /api/v1/admin/vendors/{vendor}/export/letzshop
- Add vendor endpoint: GET /api/v1/vendor/letzshop/export
- Support language selection (en, fr, de) and include_inactive filter
- Generate Google Shopping compatible tab-delimited CSV format
- Add comprehensive integration tests for both endpoints
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -308,3 +308,56 @@ def delete_vendor(
|
||||
message = admin_service.delete_vendor(db, vendor.id)
|
||||
db.commit()
|
||||
return {"message": message}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# LETZSHOP EXPORT
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.get("/{vendor_identifier}/export/letzshop")
|
||||
def export_vendor_products_letzshop(
|
||||
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
|
||||
language: str = Query("en", description="Language for title/description (en, fr, de)"),
|
||||
include_inactive: bool = Query(False, description="Include inactive products"),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_admin_api),
|
||||
):
|
||||
"""
|
||||
Export vendor products in Letzshop CSV format (Admin only).
|
||||
|
||||
Generates a Google Shopping compatible CSV file for Letzshop marketplace.
|
||||
The file uses tab-separated values and includes all required Letzshop fields.
|
||||
|
||||
**Supported languages:** en, fr, de
|
||||
|
||||
**CSV Format:**
|
||||
- Delimiter: Tab (\\t)
|
||||
- Encoding: UTF-8
|
||||
- Fields: id, title, description, price, availability, image_link, etc.
|
||||
|
||||
Returns:
|
||||
CSV file as attachment (vendor_code_letzshop_export.csv)
|
||||
"""
|
||||
from fastapi.responses import Response
|
||||
|
||||
from app.services.letzshop_export_service import letzshop_export_service
|
||||
|
||||
vendor = vendor_service.get_vendor_by_identifier(db, vendor_identifier)
|
||||
|
||||
csv_content = letzshop_export_service.export_vendor_products(
|
||||
db=db,
|
||||
vendor_id=vendor.id,
|
||||
language=language,
|
||||
include_inactive=include_inactive,
|
||||
)
|
||||
|
||||
filename = f"{vendor.vendor_code.lower()}_letzshop_export.csv"
|
||||
|
||||
return Response(
|
||||
content=csv_content,
|
||||
media_type="text/csv; charset=utf-8",
|
||||
headers={
|
||||
"Content-Disposition": f'attachment; filename="{filename}"',
|
||||
},
|
||||
)
|
||||
|
||||
54
app/api/v1/vendor/letzshop.py
vendored
54
app/api/v1/vendor/letzshop.py
vendored
@@ -690,3 +690,57 @@ def list_fulfillment_queue(
|
||||
skip=skip,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Product Export
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.get("/export")
|
||||
def export_products_letzshop(
|
||||
language: str = Query("en", description="Language for title/description (en, fr, de)"),
|
||||
include_inactive: bool = Query(False, description="Include inactive products"),
|
||||
current_user: User = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Export vendor products in Letzshop CSV format.
|
||||
|
||||
Generates a Google Shopping compatible CSV file for Letzshop marketplace.
|
||||
The file uses tab-separated values and includes all required Letzshop fields.
|
||||
|
||||
**Supported languages:** en, fr, de
|
||||
|
||||
**CSV Format:**
|
||||
- Delimiter: Tab (\\t)
|
||||
- Encoding: UTF-8
|
||||
- Fields: id, title, description, price, availability, image_link, etc.
|
||||
|
||||
Returns:
|
||||
CSV file as attachment (vendor_code_letzshop_export.csv)
|
||||
"""
|
||||
from fastapi.responses import Response
|
||||
|
||||
from app.services.letzshop_export_service import letzshop_export_service
|
||||
from app.services.vendor_service import vendor_service
|
||||
|
||||
vendor_id = current_user.token_vendor_id
|
||||
vendor = vendor_service.get_vendor_by_id(db, vendor_id)
|
||||
|
||||
csv_content = letzshop_export_service.export_vendor_products(
|
||||
db=db,
|
||||
vendor_id=vendor_id,
|
||||
language=language,
|
||||
include_inactive=include_inactive,
|
||||
)
|
||||
|
||||
filename = f"{vendor.vendor_code.lower()}_letzshop_export.csv"
|
||||
|
||||
return Response(
|
||||
content=csv_content,
|
||||
media_type="text/csv; charset=utf-8",
|
||||
headers={
|
||||
"Content-Disposition": f'attachment; filename="{filename}"',
|
||||
},
|
||||
)
|
||||
|
||||
280
app/services/letzshop_export_service.py
Normal file
280
app/services/letzshop_export_service.py
Normal file
@@ -0,0 +1,280 @@
|
||||
# app/services/letzshop_export_service.py
|
||||
"""
|
||||
Service for exporting products to Letzshop CSV format.
|
||||
|
||||
Generates Google Shopping compatible CSV files for Letzshop marketplace.
|
||||
"""
|
||||
import csv
|
||||
import io
|
||||
import logging
|
||||
from typing import BinaryIO
|
||||
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.product import Product
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Letzshop CSV columns in order
|
||||
LETZSHOP_CSV_COLUMNS = [
|
||||
"id",
|
||||
"title",
|
||||
"description",
|
||||
"link",
|
||||
"image_link",
|
||||
"additional_image_link",
|
||||
"availability",
|
||||
"price",
|
||||
"sale_price",
|
||||
"brand",
|
||||
"gtin",
|
||||
"mpn",
|
||||
"google_product_category",
|
||||
"product_type",
|
||||
"condition",
|
||||
"adult",
|
||||
"multipack",
|
||||
"is_bundle",
|
||||
"age_group",
|
||||
"color",
|
||||
"gender",
|
||||
"material",
|
||||
"pattern",
|
||||
"size",
|
||||
"size_type",
|
||||
"size_system",
|
||||
"item_group_id",
|
||||
"custom_label_0",
|
||||
"custom_label_1",
|
||||
"custom_label_2",
|
||||
"custom_label_3",
|
||||
"custom_label_4",
|
||||
"identifier_exists",
|
||||
"unit_pricing_measure",
|
||||
"unit_pricing_base_measure",
|
||||
"shipping",
|
||||
"atalanda:tax_rate",
|
||||
"atalanda:quantity",
|
||||
"atalanda:boost_sort",
|
||||
"atalanda:delivery_method",
|
||||
]
|
||||
|
||||
|
||||
class LetzshopExportService:
|
||||
"""Service for exporting products to Letzshop CSV format."""
|
||||
|
||||
def __init__(self, default_tax_rate: float = 17.0):
|
||||
"""
|
||||
Initialize the export service.
|
||||
|
||||
Args:
|
||||
default_tax_rate: Default VAT rate for Luxembourg (17%)
|
||||
"""
|
||||
self.default_tax_rate = default_tax_rate
|
||||
|
||||
def export_vendor_products(
|
||||
self,
|
||||
db: Session,
|
||||
vendor_id: int,
|
||||
language: str = "en",
|
||||
include_inactive: bool = False,
|
||||
) -> str:
|
||||
"""
|
||||
Export all products for a vendor in Letzshop CSV format.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
vendor_id: Vendor ID to export products for
|
||||
language: Language for title/description (en, fr, de)
|
||||
include_inactive: Whether to include inactive products
|
||||
|
||||
Returns:
|
||||
CSV string content
|
||||
"""
|
||||
# Query products for this vendor with their marketplace product data
|
||||
query = (
|
||||
db.query(Product)
|
||||
.filter(Product.vendor_id == vendor_id)
|
||||
.options(
|
||||
joinedload(Product.marketplace_product).joinedload(
|
||||
MarketplaceProduct.translations
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
if not include_inactive:
|
||||
query = query.filter(Product.is_active == True)
|
||||
|
||||
products = query.all()
|
||||
|
||||
logger.info(
|
||||
f"Exporting {len(products)} products for vendor {vendor_id} in {language}"
|
||||
)
|
||||
|
||||
return self._generate_csv(products, language)
|
||||
|
||||
def export_marketplace_products(
|
||||
self,
|
||||
db: Session,
|
||||
marketplace: str = "Letzshop",
|
||||
language: str = "en",
|
||||
limit: int | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
Export marketplace products directly (admin use).
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
marketplace: Filter by marketplace source
|
||||
language: Language for title/description
|
||||
limit: Optional limit on number of products
|
||||
|
||||
Returns:
|
||||
CSV string content
|
||||
"""
|
||||
query = (
|
||||
db.query(MarketplaceProduct)
|
||||
.filter(MarketplaceProduct.is_active == True)
|
||||
.options(joinedload(MarketplaceProduct.translations))
|
||||
)
|
||||
|
||||
if marketplace:
|
||||
query = query.filter(MarketplaceProduct.marketplace.ilike(f"%{marketplace}%"))
|
||||
|
||||
if limit:
|
||||
query = query.limit(limit)
|
||||
|
||||
products = query.all()
|
||||
|
||||
logger.info(
|
||||
f"Exporting {len(products)} marketplace products for {marketplace} in {language}"
|
||||
)
|
||||
|
||||
return self._generate_csv_from_marketplace_products(products, language)
|
||||
|
||||
def _generate_csv(self, products: list[Product], language: str) -> str:
|
||||
"""Generate CSV from vendor Product objects."""
|
||||
output = io.StringIO()
|
||||
writer = csv.DictWriter(
|
||||
output,
|
||||
fieldnames=LETZSHOP_CSV_COLUMNS,
|
||||
delimiter="\t",
|
||||
quoting=csv.QUOTE_MINIMAL,
|
||||
)
|
||||
writer.writeheader()
|
||||
|
||||
for product in products:
|
||||
if product.marketplace_product:
|
||||
row = self._product_to_row(product, language)
|
||||
writer.writerow(row)
|
||||
|
||||
return output.getvalue()
|
||||
|
||||
def _generate_csv_from_marketplace_products(
|
||||
self, products: list[MarketplaceProduct], language: str
|
||||
) -> str:
|
||||
"""Generate CSV from MarketplaceProduct objects directly."""
|
||||
output = io.StringIO()
|
||||
writer = csv.DictWriter(
|
||||
output,
|
||||
fieldnames=LETZSHOP_CSV_COLUMNS,
|
||||
delimiter="\t",
|
||||
quoting=csv.QUOTE_MINIMAL,
|
||||
)
|
||||
writer.writeheader()
|
||||
|
||||
for mp in products:
|
||||
row = self._marketplace_product_to_row(mp, language)
|
||||
writer.writerow(row)
|
||||
|
||||
return output.getvalue()
|
||||
|
||||
def _product_to_row(self, product: Product, language: str) -> dict:
|
||||
"""Convert a Product (with MarketplaceProduct) to a CSV row."""
|
||||
mp = product.marketplace_product
|
||||
return self._marketplace_product_to_row(mp, language, vendor_sku=product.vendor_sku)
|
||||
|
||||
def _marketplace_product_to_row(
|
||||
self,
|
||||
mp: MarketplaceProduct,
|
||||
language: str,
|
||||
vendor_sku: str | None = None,
|
||||
) -> dict:
|
||||
"""Convert a MarketplaceProduct to a CSV row dict."""
|
||||
# Get localized title and description
|
||||
title = mp.get_title(language) or ""
|
||||
description = mp.get_description(language) or ""
|
||||
|
||||
# Format price with currency
|
||||
price = ""
|
||||
if mp.price_numeric:
|
||||
price = f"{mp.price_numeric:.2f} {mp.currency or 'EUR'}"
|
||||
elif mp.price:
|
||||
price = mp.price
|
||||
|
||||
# Format sale price
|
||||
sale_price = ""
|
||||
if mp.sale_price_numeric:
|
||||
sale_price = f"{mp.sale_price_numeric:.2f} {mp.currency or 'EUR'}"
|
||||
elif mp.sale_price:
|
||||
sale_price = mp.sale_price
|
||||
|
||||
# Additional images - join with comma if multiple
|
||||
additional_images = ""
|
||||
if mp.additional_images:
|
||||
additional_images = ",".join(mp.additional_images)
|
||||
elif mp.additional_image_link:
|
||||
additional_images = mp.additional_image_link
|
||||
|
||||
# Determine identifier_exists
|
||||
identifier_exists = mp.identifier_exists
|
||||
if not identifier_exists:
|
||||
identifier_exists = "yes" if (mp.gtin or mp.mpn) else "no"
|
||||
|
||||
return {
|
||||
"id": vendor_sku or mp.marketplace_product_id,
|
||||
"title": title,
|
||||
"description": description,
|
||||
"link": mp.link or mp.source_url or "",
|
||||
"image_link": mp.image_link or "",
|
||||
"additional_image_link": additional_images,
|
||||
"availability": mp.availability or "in stock",
|
||||
"price": price,
|
||||
"sale_price": sale_price,
|
||||
"brand": mp.brand or "",
|
||||
"gtin": mp.gtin or "",
|
||||
"mpn": mp.mpn or "",
|
||||
"google_product_category": mp.google_product_category or "",
|
||||
"product_type": mp.product_type_raw or "",
|
||||
"condition": mp.condition or "new",
|
||||
"adult": mp.adult or "no",
|
||||
"multipack": str(mp.multipack) if mp.multipack else "",
|
||||
"is_bundle": mp.is_bundle or "no",
|
||||
"age_group": mp.age_group or "",
|
||||
"color": mp.color or "",
|
||||
"gender": mp.gender or "",
|
||||
"material": mp.material or "",
|
||||
"pattern": mp.pattern or "",
|
||||
"size": mp.size or "",
|
||||
"size_type": mp.size_type or "",
|
||||
"size_system": mp.size_system or "",
|
||||
"item_group_id": mp.item_group_id or "",
|
||||
"custom_label_0": mp.custom_label_0 or "",
|
||||
"custom_label_1": mp.custom_label_1 or "",
|
||||
"custom_label_2": mp.custom_label_2 or "",
|
||||
"custom_label_3": mp.custom_label_3 or "",
|
||||
"custom_label_4": mp.custom_label_4 or "",
|
||||
"identifier_exists": identifier_exists,
|
||||
"unit_pricing_measure": mp.unit_pricing_measure or "",
|
||||
"unit_pricing_base_measure": mp.unit_pricing_base_measure or "",
|
||||
"shipping": mp.shipping or "",
|
||||
"atalanda:tax_rate": str(self.default_tax_rate),
|
||||
"atalanda:quantity": "", # Would need inventory data
|
||||
"atalanda:boost_sort": "",
|
||||
"atalanda:delivery_method": "",
|
||||
}
|
||||
|
||||
|
||||
# Singleton instance
|
||||
letzshop_export_service = LetzshopExportService()
|
||||
@@ -359,3 +359,197 @@ class TestAdminLetzshopAccessControl:
|
||||
response = client.get("/api/v1/admin/letzshop/vendors")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.admin
|
||||
@pytest.mark.letzshop
|
||||
class TestAdminLetzshopExportAPI:
|
||||
"""Test admin Letzshop product export endpoints."""
|
||||
|
||||
def test_export_vendor_products_empty(
|
||||
self, client, admin_headers, test_vendor
|
||||
):
|
||||
"""Test exporting products when vendor has no products."""
|
||||
response = client.get(
|
||||
f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop",
|
||||
headers=admin_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "text/csv; charset=utf-8"
|
||||
# Should have header row at minimum
|
||||
content = response.text
|
||||
assert "id\ttitle\tdescription" in content
|
||||
|
||||
def test_export_vendor_products_with_data(
|
||||
self, client, db, admin_headers, test_vendor
|
||||
):
|
||||
"""Test exporting products with actual data."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
# Create marketplace product
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="EXPORT-TEST-001",
|
||||
price="29.99",
|
||||
price_numeric=29.99,
|
||||
currency="EUR",
|
||||
brand="TestBrand",
|
||||
availability="in stock",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
# Add translation
|
||||
translation = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="en",
|
||||
title="Export Test Product",
|
||||
description="A product for testing exports",
|
||||
)
|
||||
db.add(translation)
|
||||
db.flush()
|
||||
|
||||
# Create product linked to vendor
|
||||
product = Product(
|
||||
vendor_id=test_vendor.id,
|
||||
vendor_sku="EXP-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop",
|
||||
headers=admin_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "EXP-001" in content
|
||||
assert "Export Test Product" in content
|
||||
assert "29.99 EUR" in content
|
||||
|
||||
def test_export_vendor_products_french(
|
||||
self, client, db, admin_headers, test_vendor
|
||||
):
|
||||
"""Test exporting products in French."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="EXPORT-FR-001",
|
||||
price="19.99",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
# Add French translation
|
||||
translation_fr = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="fr",
|
||||
title="Produit Test Export",
|
||||
description="Un produit pour tester les exportations",
|
||||
)
|
||||
db.add(translation_fr)
|
||||
db.flush()
|
||||
|
||||
product = Product(
|
||||
vendor_id=test_vendor.id,
|
||||
vendor_sku="EXP-FR-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop?language=fr",
|
||||
headers=admin_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "Produit Test Export" in content
|
||||
|
||||
def test_export_vendor_products_include_inactive(
|
||||
self, client, db, admin_headers, test_vendor
|
||||
):
|
||||
"""Test exporting including inactive products."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
# Create inactive product
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="EXPORT-INACTIVE-001",
|
||||
price="5.99",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
translation = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="en",
|
||||
title="Inactive Product",
|
||||
)
|
||||
db.add(translation)
|
||||
db.flush()
|
||||
|
||||
product = Product(
|
||||
vendor_id=test_vendor.id,
|
||||
vendor_sku="INACTIVE-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=False, # Inactive
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
# Without include_inactive
|
||||
response = client.get(
|
||||
f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop",
|
||||
headers=admin_headers,
|
||||
)
|
||||
assert "INACTIVE-001" not in response.text
|
||||
|
||||
# With include_inactive
|
||||
response = client.get(
|
||||
f"/api/v1/admin/vendors/{test_vendor.id}/export/letzshop?include_inactive=true",
|
||||
headers=admin_headers,
|
||||
)
|
||||
assert "INACTIVE-001" in response.text
|
||||
|
||||
def test_export_vendor_products_by_vendor_code(
|
||||
self, client, admin_headers, test_vendor
|
||||
):
|
||||
"""Test exporting products using vendor_code identifier."""
|
||||
response = client.get(
|
||||
f"/api/v1/admin/vendors/{test_vendor.vendor_code}/export/letzshop",
|
||||
headers=admin_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "text/csv; charset=utf-8"
|
||||
|
||||
def test_export_vendor_not_found(self, client, admin_headers):
|
||||
"""Test exporting for non-existent vendor."""
|
||||
response = client.get(
|
||||
"/api/v1/admin/vendors/999999/export/letzshop",
|
||||
headers=admin_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
231
tests/integration/api/v1/vendor/test_letzshop.py
vendored
231
tests/integration/api/v1/vendor/test_letzshop.py
vendored
@@ -497,3 +497,234 @@ class TestVendorLetzshopFulfillmentAPI:
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["tracking_number"] == "1Z999AA1"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.api
|
||||
@pytest.mark.vendor
|
||||
@pytest.mark.letzshop
|
||||
class TestVendorLetzshopExportAPI:
|
||||
"""Test vendor Letzshop product export endpoints."""
|
||||
|
||||
def test_export_products_empty(
|
||||
self, client, vendor_user_headers, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test exporting products when vendor has no products."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export", headers=vendor_user_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "text/csv; charset=utf-8"
|
||||
# Should have header row at minimum
|
||||
content = response.text
|
||||
assert "id\ttitle\tdescription" in content
|
||||
|
||||
def test_export_products_with_data(
|
||||
self, client, db, vendor_user_headers, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test exporting products with actual data."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
# Create marketplace product
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="VENDOR-EXPORT-001",
|
||||
price="49.99",
|
||||
price_numeric=49.99,
|
||||
currency="EUR",
|
||||
brand="VendorBrand",
|
||||
availability="in stock",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
# Add translation
|
||||
translation = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="en",
|
||||
title="Vendor Export Product",
|
||||
description="A product for testing vendor exports",
|
||||
)
|
||||
db.add(translation)
|
||||
db.flush()
|
||||
|
||||
# Create product linked to vendor
|
||||
product = Product(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
vendor_sku="VEXP-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export", headers=vendor_user_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "VEXP-001" in content
|
||||
assert "Vendor Export Product" in content
|
||||
assert "49.99 EUR" in content
|
||||
|
||||
def test_export_products_french_language(
|
||||
self, client, db, vendor_user_headers, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test exporting products in French."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="VENDOR-FR-001",
|
||||
price="39.99",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
# Add French translation
|
||||
translation_fr = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="fr",
|
||||
title="Produit Vendeur Export",
|
||||
description="Un produit pour tester",
|
||||
)
|
||||
db.add(translation_fr)
|
||||
db.flush()
|
||||
|
||||
product = Product(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
vendor_sku="VFR-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export?language=fr",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "Produit Vendeur Export" in content
|
||||
|
||||
def test_export_products_german_language(
|
||||
self, client, db, vendor_user_headers, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test exporting products in German."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="VENDOR-DE-001",
|
||||
price="59.99",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
translation_de = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="de",
|
||||
title="Deutsches Exportprodukt",
|
||||
)
|
||||
db.add(translation_de)
|
||||
db.flush()
|
||||
|
||||
product = Product(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
vendor_sku="VDE-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export?language=de",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
content = response.text
|
||||
assert "Deutsches Exportprodukt" in content
|
||||
|
||||
def test_export_products_include_inactive(
|
||||
self, client, db, vendor_user_headers, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test exporting including inactive products."""
|
||||
from models.database.product import Product
|
||||
from models.database.marketplace_product import MarketplaceProduct
|
||||
from models.database.marketplace_product_translation import (
|
||||
MarketplaceProductTranslation,
|
||||
)
|
||||
|
||||
mp = MarketplaceProduct(
|
||||
marketplace_product_id="VENDOR-INACTIVE-001",
|
||||
price="9.99",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(mp)
|
||||
db.flush()
|
||||
|
||||
translation = MarketplaceProductTranslation(
|
||||
marketplace_product_id=mp.id,
|
||||
language="en",
|
||||
title="Inactive Vendor Product",
|
||||
)
|
||||
db.add(translation)
|
||||
db.flush()
|
||||
|
||||
product = Product(
|
||||
vendor_id=test_vendor_with_vendor_user.id,
|
||||
vendor_sku="VINACTIVE-001",
|
||||
marketplace_product_id=mp.id,
|
||||
is_active=False, # Inactive
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
|
||||
# Without include_inactive
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
assert "VINACTIVE-001" not in response.text
|
||||
|
||||
# With include_inactive
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export?include_inactive=true",
|
||||
headers=vendor_user_headers,
|
||||
)
|
||||
assert "VINACTIVE-001" in response.text
|
||||
|
||||
def test_export_unauthenticated(self, client):
|
||||
"""Test that unauthenticated requests are rejected."""
|
||||
response = client.get("/api/v1/vendor/letzshop/export")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_export_filename_uses_vendor_code(
|
||||
self, client, vendor_user_headers, test_vendor_with_vendor_user
|
||||
):
|
||||
"""Test that export filename uses vendor code."""
|
||||
response = client.get(
|
||||
"/api/v1/vendor/letzshop/export", headers=vendor_user_headers
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
content_disposition = response.headers.get("content-disposition", "")
|
||||
assert "letzshop_export.csv" in content_disposition
|
||||
|
||||
Reference in New Issue
Block a user