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

feat: enhance Letzshop order import with EAN matching and stats

Browse Source 
- Add historical order import with pagination support
- Add customer_locale, shipping_country_iso, billing_country_iso columns
- Add gtin/gtin_type columns to Product table for EAN matching
- Fix order stats to count all orders server-side (not just visible page)
- Add GraphQL introspection script with tracking workaround tests
- Enrich inventory units with EAN, MPN, SKU, product name
- Add LetzshopOrderStats schema for proper status counts

Migrations:
- a9a86cef6cca: Add locale and country fields to letzshop_orders
- cb88bc9b5f86: Add gtin columns to products table

🤖 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-18 21:04:33 +01:00
parent 6d6c8b44d3
commit 0ab10128ae
17 changed files with 3451 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/integration/api/v1/admin/test_letzshop.py
View File

@@ -287,9 +287,9 @@ class TestAdminLetzshopOrdersAPI:
"id": "gid://letzshop/Order/111",
"number": "LS-ADMIN-001",
"email": "sync@example.com",
"totalPrice": {"amount": "200.00", "currency": "EUR"},
"total": "200.00",
},
"inventoryUnits": {"nodes": []},
"inventoryUnits": [],
}
]
}

10
tests/integration/api/v1/vendor/test_letzshop.py vendored
View File

@@ -344,13 +344,11 @@ class TestVendorLetzshopOrdersAPI:
"id": "gid://letzshop/Order/456",
"number": "LS-2025-001",
"email": "customer@example.com",
"totalPrice": {"amount": "99.99", "currency": "EUR"},
},
"inventoryUnits": {
"nodes": [
{"id": "unit_1", "state": "unconfirmed"},
]
"total": "99.99",
},
"inventoryUnits": [
{"id": "unit_1", "state": "unconfirmed"},
],
}
]
}

128
tests/unit/models/database/test_product.py
View File

@@ -227,3 +227,131 @@ class TestProductModel:
assert info["brand"] == "SourceBrand"
assert info["brand_overridden"] is False
assert info["brand_source"] == "SourceBrand"
@pytest.mark.unit
@pytest.mark.database
@pytest.mark.inventory
class TestProductInventoryProperties:
"""Test Product inventory properties including digital product handling."""
def test_physical_product_no_inventory_returns_zero(
self, db, test_vendor, test_marketplace_product
):
"""Test physical product with no inventory entries returns 0."""
# Ensure product is physical
test_marketplace_product.is_digital = False
db.commit()
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.is_digital is False
assert product.has_unlimited_inventory is False
assert product.total_inventory == 0
assert product.available_inventory == 0
def test_physical_product_with_inventory(
self, db, test_vendor, test_marketplace_product
):
"""Test physical product calculates inventory from entries."""
from models.database.inventory import Inventory
test_marketplace_product.is_digital = False
db.commit()
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
)
db.add(product)
db.commit()
db.refresh(product)
# Add inventory entries
inv1 = Inventory(
product_id=product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_A",
quantity=100,
reserved_quantity=10,
)
inv2 = Inventory(
product_id=product.id,
vendor_id=test_vendor.id,
location="WAREHOUSE_B",
quantity=50,
reserved_quantity=5,
)
db.add_all([inv1, inv2])
db.commit()
db.refresh(product)
assert product.has_unlimited_inventory is False
assert product.total_inventory == 150 # 100 + 50
assert product.available_inventory == 135 # (100-10) + (50-5)
def test_digital_product_has_unlimited_inventory(
self, db, test_vendor, test_marketplace_product
):
"""Test digital product returns unlimited inventory."""
test_marketplace_product.is_digital = True
db.commit()
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
)
db.add(product)
db.commit()
db.refresh(product)
assert product.is_digital is True
assert product.has_unlimited_inventory is True
assert product.total_inventory == Product.UNLIMITED_INVENTORY
assert product.available_inventory == Product.UNLIMITED_INVENTORY
def test_digital_product_ignores_inventory_entries(
self, db, test_vendor, test_marketplace_product
):
"""Test digital product returns unlimited even with inventory entries."""
from models.database.inventory import Inventory
test_marketplace_product.is_digital = True
db.commit()
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=test_marketplace_product.id,
)
db.add(product)
db.commit()
db.refresh(product)
# Add inventory entries (e.g., for license keys)
inv = Inventory(
product_id=product.id,
vendor_id=test_vendor.id,
location="DIGITAL_LICENSES",
quantity=10,
reserved_quantity=2,
)
db.add(inv)
db.commit()
db.refresh(product)
# Digital product should still return unlimited
assert product.has_unlimited_inventory is True
assert product.total_inventory == Product.UNLIMITED_INVENTORY
assert product.available_inventory == Product.UNLIMITED_INVENTORY
def test_unlimited_inventory_constant(self):
"""Test UNLIMITED_INVENTORY constant value."""
assert Product.UNLIMITED_INVENTORY == 999999
# Should be large enough to never cause "insufficient inventory"
assert Product.UNLIMITED_INVENTORY > 100000

300
tests/unit/services/test_letzshop_service.py
View File

@@ -447,3 +447,303 @@ class TestLetzshopClient:
client.get_shipments()
assert "Invalid shipment ID" in str(exc_info.value)
@patch("requests.Session.post")
def test_get_all_shipments_paginated(self, mock_post):
"""Test paginated shipment fetching."""
# First page response
page1_response = MagicMock()
page1_response.status_code = 200
page1_response.json.return_value = {
"data": {
"shipments": {
"pageInfo": {
"hasNextPage": True,
"endCursor": "cursor_1",
},
"nodes": [
{"id": "ship_1", "state": "confirmed"},
{"id": "ship_2", "state": "confirmed"},
],
}
}
}
# Second page response
page2_response = MagicMock()
page2_response.status_code = 200
page2_response.json.return_value = {
"data": {
"shipments": {
"pageInfo": {
"hasNextPage": False,
"endCursor": None,
},
"nodes": [
{"id": "ship_3", "state": "confirmed"},
],
}
}
}
mock_post.side_effect = [page1_response, page2_response]
client = LetzshopClient(api_key="test-key")
shipments = client.get_all_shipments_paginated(
state="confirmed",
page_size=2,
)
assert len(shipments) == 3
assert shipments[0]["id"] == "ship_1"
assert shipments[2]["id"] == "ship_3"
@patch("requests.Session.post")
def test_get_all_shipments_paginated_with_max_pages(self, mock_post):
"""Test paginated fetching respects max_pages limit."""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": {
"shipments": {
"pageInfo": {
"hasNextPage": True,
"endCursor": "cursor_1",
},
"nodes": [
{"id": "ship_1", "state": "confirmed"},
],
}
}
}
mock_post.return_value = mock_response
client = LetzshopClient(api_key="test-key")
shipments = client.get_all_shipments_paginated(
state="confirmed",
page_size=1,
max_pages=1, # Only fetch 1 page
)
assert len(shipments) == 1
assert mock_post.call_count == 1
@patch("requests.Session.post")
def test_get_all_shipments_paginated_with_callback(self, mock_post):
"""Test paginated fetching calls progress callback."""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": {
"shipments": {
"pageInfo": {"hasNextPage": False, "endCursor": None},
"nodes": [{"id": "ship_1"}],
}
}
}
mock_post.return_value = mock_response
callback_calls = []
def callback(page, total):
callback_calls.append((page, total))
client = LetzshopClient(api_key="test-key")
client.get_all_shipments_paginated(
state="confirmed",
progress_callback=callback,
)
assert len(callback_calls) == 1
assert callback_calls[0] == (1, 1)
# ============================================================================
# Order Service Tests
# ============================================================================
@pytest.mark.unit
@pytest.mark.letzshop
class TestLetzshopOrderService:
"""Test suite for Letzshop order service."""
def test_create_order_extracts_locale(self, db, test_vendor):
"""Test that create_order extracts customer locale."""
from app.services.letzshop.order_service import LetzshopOrderService
service = LetzshopOrderService(db)
shipment_data = {
"id": "ship_123",
"state": "confirmed",
"order": {
"id": "order_123",
"number": "R123456",
"email": "test@example.com",
"total": 29.99,
"locale": "fr",
"shipAddress": {
"firstName": "Jean",
"lastName": "Dupont",
"country": {"iso": "LU"},
},
"billAddress": {
"country": {"iso": "FR"},
},
},
"inventoryUnits": [],
}
order = service.create_order(test_vendor.id, shipment_data)
assert order.customer_locale == "fr"
assert order.shipping_country_iso == "LU"
assert order.billing_country_iso == "FR"
def test_create_order_extracts_ean(self, db, test_vendor):
"""Test that create_order extracts EAN from tradeId."""
from app.services.letzshop.order_service import LetzshopOrderService
service = LetzshopOrderService(db)
shipment_data = {
"id": "ship_123",
"state": "confirmed",
"order": {
"id": "order_123",
"number": "R123456",
"email": "test@example.com",
"total": 29.99,
"shipAddress": {},
},
"inventoryUnits": [
{
"id": "unit_1",
"state": "confirmed",
"variant": {
"id": "var_1",
"sku": "SKU123",
"mpn": "MPN456",
"price": 19.99,
"tradeId": {
"number": "0889698273022",
"parser": "gtin13",
},
"product": {
"name": {"en": "Test Product", "fr": "Produit Test"},
},
},
}
],
}
order = service.create_order(test_vendor.id, shipment_data)
assert len(order.inventory_units) == 1
unit = order.inventory_units[0]
assert unit["ean"] == "0889698273022"
assert unit["ean_type"] == "gtin13"
assert unit["sku"] == "SKU123"
assert unit["mpn"] == "MPN456"
assert unit["product_name"] == "Test Product"
assert unit["price"] == 19.99
def test_import_historical_shipments_deduplication(self, db, test_vendor):
"""Test that historical import deduplicates existing orders."""
from app.services.letzshop.order_service import LetzshopOrderService
service = LetzshopOrderService(db)
shipment_data = {
"id": "ship_existing",
"state": "confirmed",
"order": {
"id": "order_123",
"number": "R123456",
"email": "test@example.com",
"total": 29.99,
"shipAddress": {},
},
"inventoryUnits": [],
}
# Create first order
service.create_order(test_vendor.id, shipment_data)
db.commit()
# Import same shipment again
stats = service.import_historical_shipments(
vendor_id=test_vendor.id,
shipments=[shipment_data],
match_products=False,
)
assert stats["total"] == 1
assert stats["imported"] == 0
assert stats["skipped"] == 1
def test_import_historical_shipments_new_orders(self, db, test_vendor):
"""Test that historical import creates new orders."""
from app.services.letzshop.order_service import LetzshopOrderService
service = LetzshopOrderService(db)
shipments = [
{
"id": f"ship_{i}",
"state": "confirmed",
"order": {
"id": f"order_{i}",
"number": f"R{i}",
"email": f"customer{i}@example.com",
"total": 29.99,
"shipAddress": {},
},
"inventoryUnits": [],
}
for i in range(3)
]
stats = service.import_historical_shipments(
vendor_id=test_vendor.id,
shipments=shipments,
match_products=False,
)
assert stats["total"] == 3
assert stats["imported"] == 3
assert stats["skipped"] == 0
def test_get_historical_import_summary(self, db, test_vendor):
"""Test historical import summary statistics."""
from app.services.letzshop.order_service import LetzshopOrderService
service = LetzshopOrderService(db)
# Create some orders with different locales
for i, locale in enumerate(["fr", "fr", "de", "en"]):
shipment_data = {
"id": f"ship_{i}",
"state": "confirmed",
"order": {
"id": f"order_{i}",
"number": f"R{i}",
"email": f"customer{i}@example.com",
"total": 29.99,
"locale": locale,
"shipAddress": {"country": {"iso": "LU"}},
},
"inventoryUnits": [],
}
service.create_order(test_vendor.id, shipment_data)
db.commit()
summary = service.get_historical_import_summary(test_vendor.id)
assert summary["total_orders"] == 4
assert summary["unique_customers"] == 4
assert summary["orders_by_locale"]["fr"] == 2
assert summary["orders_by_locale"]["de"] == 1
assert summary["orders_by_locale"]["en"] == 1