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

fix: correct tojson|safe usage in templates and update validator

Browse Source 
- 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>
This commit is contained in:
Samir Boulahtit
2025-12-13 22:59:51 +01:00
parent 94d268f330
commit 9920430b9e
123 changed files with 1408 additions and 840 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
tests/conftest.py
View File

@@ -18,6 +18,7 @@ IMPORTANT - Fixture Best Practices:
See docs/testing/testing-guide.md for comprehensive testing documentation.
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine

6
tests/fixtures/auth_fixtures.py vendored
View File

@@ -5,6 +5,7 @@ Authentication-related test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import uuid
import pytest
@@ -140,7 +141,10 @@ def vendor_user_headers(client, test_vendor_user, test_vendor_with_vendor_user):
"""
response = client.post(
"/api/v1/vendor/auth/login",
json={"email_or_username": test_vendor_user.username, "password": "vendorpass123"},
json={
"email_or_username": test_vendor_user.username,
"password": "vendorpass123",
},
)
assert response.status_code == 200, f"Vendor login failed: {response.text}"
token = response.json()["access_token"]

1
tests/fixtures/customer_fixtures.py vendored
View File

@@ -5,6 +5,7 @@ Customer-related test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import pytest
from models.database.customer import Customer, CustomerAddress

1
tests/fixtures/marketplace_import_job_fixtures.py vendored
View File

@@ -5,6 +5,7 @@ Marketplace import job test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import pytest
from models.database.marketplace_import_job import MarketplaceImportJob

5
tests/fixtures/marketplace_product_fixtures.py vendored
View File

@@ -8,12 +8,15 @@ we create the translation alongside the marketplace product.
Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import uuid
import pytest
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
from models.database.marketplace_product_translation import (
MarketplaceProductTranslation,
)
def _create_marketplace_product_with_translation(

1
tests/fixtures/vendor_fixtures.py vendored
View File

@@ -5,6 +5,7 @@ Vendor-related test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import uuid
import pytest

14
tests/integration/api/v1/admin/test_auth.py
View File

@@ -3,6 +3,7 @@
Tests the /api/v1/admin/auth/* endpoints.
"""
from datetime import UTC, datetime, timedelta
import pytest
@@ -46,7 +47,10 @@ class TestAdminAuthAPI:
"""Test login with wrong password."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_admin.username, "password": "wrongpassword"},
json={
"email_or_username": test_admin.username,
"password": "wrongpassword",
},
)
assert response.status_code == 401
@@ -73,7 +77,10 @@ class TestAdminAuthAPI:
try:
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_admin.username, "password": "adminpass123"},
json={
"email_or_username": test_admin.username,
"password": "adminpass123",
},
)
assert response.status_code == 403
@@ -153,7 +160,8 @@ class TestAdminAuthAPI:
)
response = client.get(
"/api/v1/admin/auth/me", headers={"Authorization": f"Bearer {expired_token}"}
"/api/v1/admin/auth/me",
headers={"Authorization": f"Bearer {expired_token}"},
)
assert response.status_code == 401

9
tests/integration/api/v1/admin/test_dashboard.py
View File

@@ -4,6 +4,7 @@ Integration tests for admin dashboard and statistics endpoints.
Tests the /api/v1/admin/dashboard/* endpoints.
"""
import pytest
@@ -34,7 +35,9 @@ class TestAdminDashboardAPI:
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
def test_get_comprehensive_stats(self, client, admin_headers, test_marketplace_product):
def test_get_comprehensive_stats(
self, client, admin_headers, test_marketplace_product
):
"""Test getting comprehensive statistics."""
response = client.get("/api/v1/admin/dashboard/stats", headers=admin_headers)
@@ -47,7 +50,9 @@ class TestAdminDashboardAPI:
assert "unique_vendors" in data
assert data["total_products"] >= 0
def test_get_marketplace_stats(self, client, admin_headers, test_marketplace_product):
def test_get_marketplace_stats(
self, client, admin_headers, test_marketplace_product
):
"""Test getting marketplace statistics."""
response = client.get(
"/api/v1/admin/dashboard/stats/marketplace", headers=admin_headers

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

@@ -9,8 +9,9 @@ Tests cover:
4. Order management for vendors
"""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.integration
@@ -20,13 +21,9 @@ from unittest.mock import patch, MagicMock
class TestAdminLetzshopVendorsAPI:
"""Test admin Letzshop vendor overview endpoints."""
def test_list_vendors_letzshop_status(
self, client, admin_headers, test_vendor
):
def test_list_vendors_letzshop_status(self, client, admin_headers, test_vendor):
"""Test listing vendors with Letzshop status."""
response = client.get(
"/api/v1/admin/letzshop/vendors", headers=admin_headers
)
response = client.get("/api/v1/admin/letzshop/vendors", headers=admin_headers)
assert response.status_code == 200
data = response.json()
@@ -41,12 +38,10 @@ class TestAdminLetzshopVendorsAPI:
break
# Vendor may not be found if inactive, that's ok
def test_list_vendors_configured_only(
self, client, db, admin_headers, test_vendor
):
def test_list_vendors_configured_only(self, client, db, admin_headers, test_vendor):
"""Test listing only configured vendors."""
from models.database.letzshop import VendorLetzshopCredentials
from app.utils.encryption import encrypt_value
from models.database.letzshop import VendorLetzshopCredentials
# Configure credentials for test vendor
credentials = VendorLetzshopCredentials(
@@ -87,9 +82,7 @@ class TestAdminLetzshopCredentialsAPI:
assert response.status_code == 404
def test_create_vendor_credentials(
self, client, admin_headers, test_vendor
):
def test_create_vendor_credentials(self, client, admin_headers, test_vendor):
"""Test creating credentials for a vendor."""
response = client.post(
f"/api/v1/admin/letzshop/vendors/{test_vendor.id}/credentials",
@@ -129,9 +122,7 @@ class TestAdminLetzshopCredentialsAPI:
data = response.json()
assert data["vendor_id"] == test_vendor.id
def test_update_vendor_credentials(
self, client, admin_headers, test_vendor
):
def test_update_vendor_credentials(self, client, admin_headers, test_vendor):
"""Test partial update of vendor credentials."""
# Create first
client.post(
@@ -151,9 +142,7 @@ class TestAdminLetzshopCredentialsAPI:
data = response.json()
assert data["auto_sync_enabled"] is True
def test_delete_vendor_credentials(
self, client, admin_headers, test_vendor
):
def test_delete_vendor_credentials(self, client, admin_headers, test_vendor):
"""Test deleting vendor credentials."""
# Create first
client.post(
@@ -218,9 +207,7 @@ class TestAdminLetzshopConnectionAPI:
assert data["success"] is True
@patch("app.services.letzshop.client_service.requests.Session.post")
def test_test_api_key_directly(
self, mock_post, client, admin_headers
):
def test_test_api_key_directly(self, mock_post, client, admin_headers):
"""Test any API key without associating with vendor."""
mock_response = MagicMock()
mock_response.status_code = 200
@@ -245,9 +232,7 @@ class TestAdminLetzshopConnectionAPI:
class TestAdminLetzshopOrdersAPI:
"""Test admin Letzshop order management endpoints."""
def test_list_vendor_orders_empty(
self, client, admin_headers, test_vendor
):
def test_list_vendor_orders_empty(self, client, admin_headers, test_vendor):
"""Test listing vendor orders when none exist."""
response = client.get(
f"/api/v1/admin/letzshop/vendors/{test_vendor.id}/orders",
@@ -259,9 +244,7 @@ class TestAdminLetzshopOrdersAPI:
assert data["orders"] == []
assert data["total"] == 0
def test_list_vendor_orders_with_data(
self, client, db, admin_headers, test_vendor
):
def test_list_vendor_orders_with_data(self, client, db, admin_headers, test_vendor):
"""Test listing vendor orders with data."""
from models.database.letzshop import LetzshopOrder
@@ -288,9 +271,7 @@ class TestAdminLetzshopOrdersAPI:
assert data["orders"][0]["customer_email"] == "admin-test@example.com"
@patch("app.services.letzshop.client_service.requests.Session.post")
def test_trigger_vendor_sync(
self, mock_post, client, admin_headers, test_vendor
):
def test_trigger_vendor_sync(self, mock_post, client, admin_headers, test_vendor):
"""Test triggering sync for a vendor."""
# Mock response
mock_response = MagicMock()
@@ -343,9 +324,7 @@ class TestAdminLetzshopOrdersAPI:
class TestAdminLetzshopAccessControl:
"""Test admin access control for Letzshop endpoints."""
def test_non_admin_cannot_access(
self, client, auth_headers, test_vendor
):
def test_non_admin_cannot_access(self, client, auth_headers, test_vendor):
"""Test that non-admin users cannot access admin endpoints."""
response = client.get(
"/api/v1/admin/letzshop/vendors",
@@ -368,9 +347,7 @@ class TestAdminLetzshopAccessControl:
class TestAdminLetzshopExportAPI:
"""Test admin Letzshop product export endpoints."""
def test_export_vendor_products_empty(
self, client, admin_headers, test_vendor
):
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",
@@ -387,11 +364,11 @@ class TestAdminLetzshopExportAPI:
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,
)
from models.database.product import Product
# Create marketplace product
mp = MarketplaceProduct(
@@ -441,11 +418,11 @@ class TestAdminLetzshopExportAPI:
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,
)
from models.database.product import Product
mp = MarketplaceProduct(
marketplace_product_id="EXPORT-FR-001",
@@ -487,11 +464,11 @@ class TestAdminLetzshopExportAPI:
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,
)
from models.database.product import Product
# Create inactive product
mp = MarketplaceProduct(

1
tests/integration/api/v1/admin/test_marketplace.py
View File

@@ -3,6 +3,7 @@
Tests the /api/v1/admin/marketplace-import-jobs/* endpoints.
"""
import pytest

5
tests/integration/api/v1/admin/test_products.py
View File

@@ -4,6 +4,7 @@ Integration tests for admin marketplace product catalog endpoints.
Tests the /api/v1/admin/products endpoints.
"""
import pytest
@@ -84,9 +85,7 @@ class TestAdminProductsAPI:
data = response.json()
assert data["total"] >= 1
def test_get_products_pagination(
self, client, admin_headers, multiple_products
):
def test_get_products_pagination(self, client, admin_headers, multiple_products):
"""Test admin products pagination."""
# Test first page
response = client.get(

1
tests/integration/api/v1/admin/test_users.py
View File

@@ -3,6 +3,7 @@
Tests the /api/v1/admin/users/* endpoints.
"""
import pytest

21
tests/integration/api/v1/admin/test_vendor_products.py
View File

@@ -4,6 +4,7 @@ Integration tests for admin vendor product catalog endpoints.
Tests the /api/v1/admin/vendor-products endpoints.
"""
import pytest
@@ -88,9 +89,7 @@ class TestAdminVendorProductsAPI:
for product in data["products"]:
assert product["is_featured"] is False
def test_get_vendor_products_pagination(
self, client, admin_headers, test_product
):
def test_get_vendor_products_pagination(self, client, admin_headers, test_product):
"""Test admin vendor products pagination."""
response = client.get(
"/api/v1/admin/vendor-products",
@@ -103,9 +102,7 @@ class TestAdminVendorProductsAPI:
assert data["skip"] == 0
assert data["limit"] == 10
def test_get_vendor_product_stats_admin(
self, client, admin_headers, test_product
):
def test_get_vendor_product_stats_admin(self, client, admin_headers, test_product):
"""Test admin getting vendor product statistics."""
response = client.get(
"/api/v1/admin/vendor-products/stats", headers=admin_headers
@@ -148,9 +145,7 @@ class TestAdminVendorProductsAPI:
vendor_ids = [v["id"] for v in data["vendors"]]
assert test_vendor.id in vendor_ids
def test_get_vendor_product_detail_admin(
self, client, admin_headers, test_product
):
def test_get_vendor_product_detail_admin(self, client, admin_headers, test_product):
"""Test admin getting vendor product detail."""
response = client.get(
f"/api/v1/admin/vendor-products/{test_product.id}",
@@ -175,9 +170,7 @@ class TestAdminVendorProductsAPI:
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
def test_remove_vendor_product_admin(
self, client, admin_headers, test_product, db
):
def test_remove_vendor_product_admin(self, client, admin_headers, test_product, db):
"""Test admin removing product from vendor catalog."""
product_id = test_product.id
@@ -209,9 +202,7 @@ class TestAdminVendorProductsAPI:
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
def test_remove_vendor_product_non_admin(
self, client, auth_headers, test_product
):
def test_remove_vendor_product_non_admin(self, client, auth_headers, test_product):
"""Test non-admin trying to remove product."""
response = client.delete(
f"/api/v1/admin/vendor-products/{test_product.id}",

1
tests/integration/api/v1/admin/test_vendors.py
View File

@@ -3,6 +3,7 @@
Tests the /api/v1/admin/vendors/* endpoints.
"""
import pytest

4
tests/integration/api/v1/vendor/test_dashboard.py vendored
View File

@@ -90,7 +90,9 @@ class TestVendorDashboardAPI:
db.commit()
# Get stats for vendor
response = client.get("/api/v1/vendor/dashboard/stats", headers=vendor_user_headers)
response = client.get(
"/api/v1/vendor/dashboard/stats", headers=vendor_user_headers
)
assert response.status_code == 200
data = response.json()

48
tests/integration/api/v1/vendor/test_inventory.py vendored
View File

@@ -4,6 +4,7 @@
Tests the /api/v1/vendor/inventory/* endpoints.
All endpoints require vendor JWT authentication.
"""
import pytest
@@ -14,11 +15,17 @@ class TestVendorInventoryAPI:
"""Test vendor inventory management endpoints at /api/v1/vendor/inventory/*."""
def test_set_inventory_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test setting inventory for a product."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -41,10 +48,16 @@ class TestVendorInventoryAPI:
assert data["quantity"] == 100
def test_adjust_inventory_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test adjusting inventory quantity."""
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -102,10 +115,16 @@ class TestVendorInventoryAPI:
assert "total" in data
def test_get_product_inventory_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test getting inventory for a specific product."""
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -128,10 +147,16 @@ class TestVendorInventoryAPI:
assert data["error_code"] == "INVALID_TOKEN"
def test_reserve_inventory_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test reserving inventory for an order."""
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -165,11 +190,15 @@ class TestVendorInventoryAPI:
assert response.status_code == 200
def test_update_inventory_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test updating inventory record."""
from models.database.product import Product
from models.database.inventory import Inventory
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
@@ -203,7 +232,12 @@ class TestVendorInventoryAPI:
assert response.status_code == 200
def test_delete_inventory_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test deleting inventory record."""
from models.database.product import Product

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

@@ -8,8 +8,9 @@ Tests cover:
3. Order listing
"""
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.integration
@@ -524,11 +525,11 @@ class TestVendorLetzshopExportAPI:
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,
)
from models.database.product import Product
# Create marketplace product
mp = MarketplaceProduct(
@@ -577,11 +578,11 @@ class TestVendorLetzshopExportAPI:
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,
)
from models.database.product import Product
mp = MarketplaceProduct(
marketplace_product_id="VENDOR-FR-001",
@@ -623,11 +624,11 @@ class TestVendorLetzshopExportAPI:
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,
)
from models.database.product import Product
mp = MarketplaceProduct(
marketplace_product_id="VENDOR-DE-001",
@@ -667,11 +668,11 @@ class TestVendorLetzshopExportAPI:
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,
)
from models.database.product import Product
mp = MarketplaceProduct(
marketplace_product_id="VENDOR-INACTIVE-001",

9
tests/integration/api/v1/vendor/test_marketplace.py vendored
View File

@@ -4,6 +4,7 @@
Tests the /api/v1/vendor/marketplace/* endpoints.
All endpoints require vendor JWT authentication.
"""
import pytest
@@ -78,8 +79,12 @@ class TestVendorMarketplaceImportJobAPI:
"""Test vendor marketplace import job management."""
def test_get_import_job_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user,
test_marketplace_import_job, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_marketplace_import_job,
db,
):
"""Test getting import job status by ID."""
# Ensure the import job belongs to the vendor

53
tests/integration/api/v1/vendor/test_products.py vendored
View File

@@ -4,6 +4,7 @@
Tests the /api/v1/vendor/products/* endpoints.
All endpoints require vendor JWT authentication.
"""
import pytest
@@ -40,11 +41,17 @@ class TestVendorProductsAPI:
assert data["marketplace_product"]["id"] == unique_product.id
def test_add_product_duplicate_returns_error(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test adding product that already exists returns error."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -65,9 +72,7 @@ class TestVendorProductsAPI:
data = response.json()
assert data["error_code"] in ["PRODUCT_ALREADY_EXISTS", "VALIDATION_ERROR"]
def test_add_nonexistent_product_returns_error(
self, client, vendor_user_headers
):
def test_add_nonexistent_product_returns_error(self, client, vendor_user_headers):
"""Test adding nonexistent marketplace product returns error."""
product_data = {
"marketplace_product_id": 99999,
@@ -126,11 +131,17 @@ class TestVendorProductsAPI:
assert data["error_code"] == "INVALID_TOKEN"
def test_get_product_detail_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test getting product details."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -154,11 +165,17 @@ class TestVendorProductsAPI:
assert response.status_code == 404
def test_update_product_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test updating product details."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -180,11 +197,17 @@ class TestVendorProductsAPI:
assert data["is_featured"] is True
def test_toggle_product_active(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test toggling product active status."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -200,11 +223,17 @@ class TestVendorProductsAPI:
assert "message" in data
def test_toggle_product_featured(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test toggling product featured status."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()
@@ -220,11 +249,17 @@ class TestVendorProductsAPI:
assert "message" in data
def test_delete_product_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, test_product, db
self,
client,
vendor_user_headers,
test_vendor_with_vendor_user,
test_product,
db,
):
"""Test removing product from catalog."""
# Ensure test_product belongs to the vendor
from models.database.product import Product
product = db.query(Product).filter(Product.id == test_product.id).first()
product.vendor_id = test_vendor_with_vendor_user.id
db.commit()

7
tests/integration/middleware/test_context_detection_flow.py
View File

@@ -127,7 +127,8 @@ class TestContextDetectionFlow:
with patch("app.core.config.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/api/test-admin-subdomain-context", headers={"host": "admin.platform.com"}
"/api/test-admin-subdomain-context",
headers={"host": "admin.platform.com"},
)
assert response.status_code == 200
@@ -289,7 +290,9 @@ class TestContextDetectionFlow:
with patch("app.core.config.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get("/shop/test-custom-domain-shop", headers={"host": "customdomain.com"})
response = client.get(
"/shop/test-custom-domain-shop", headers={"host": "customdomain.com"}
)
assert response.status_code == 200
data = response.json()

16
tests/integration/middleware/test_middleware_stack.py
View File

@@ -236,7 +236,8 @@ class TestMiddlewareStackIntegration:
@app.get("/api/test-execution-order")
async def test_execution_order(request: Request):
return {
"has_vendor": hasattr(request.state, "vendor") and request.state.vendor is not None,
"has_vendor": hasattr(request.state, "vendor")
and request.state.vendor is not None,
"has_clean_path": hasattr(request.state, "clean_path"),
"has_context_type": hasattr(request.state, "context_type"),
}
@@ -263,7 +264,8 @@ class TestMiddlewareStackIntegration:
@app.get("/api/test-theme-loading")
async def test_theme_loading(request: Request):
return {
"has_vendor": hasattr(request.state, "vendor") and request.state.vendor is not None,
"has_vendor": hasattr(request.state, "vendor")
and request.state.vendor is not None,
"has_theme": hasattr(request.state, "theme"),
"theme_primary_color": (
request.state.theme.get("primary_color")
@@ -308,7 +310,8 @@ class TestMiddlewareStackIntegration:
@app.get("/api/test-missing-vendor")
async def test_missing_vendor(request: Request):
return {
"has_vendor": hasattr(request.state, "vendor") and request.state.vendor is not None,
"has_vendor": hasattr(request.state, "vendor")
and request.state.vendor is not None,
"vendor": (
request.state.vendor if hasattr(request.state, "vendor") else None
),
@@ -339,7 +342,8 @@ class TestMiddlewareStackIntegration:
@app.get("/api/test-inactive-vendor")
async def test_inactive_vendor_endpoint(request: Request):
return {
"has_vendor": hasattr(request.state, "vendor") and request.state.vendor is not None,
"has_vendor": hasattr(request.state, "vendor")
and request.state.vendor is not None,
"vendor": (
request.state.vendor if hasattr(request.state, "vendor") else None
),
@@ -349,7 +353,9 @@ class TestMiddlewareStackIntegration:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/api/test-inactive-vendor",
headers={"host": f"{middleware_inactive_vendor.subdomain}.platform.com"},
headers={
"host": f"{middleware_inactive_vendor.subdomain}.platform.com"
},
)
assert response.status_code == 200

4
tests/integration/middleware/test_vendor_context_flow.py
View File

@@ -441,7 +441,9 @@ class TestVendorContextFlow:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-inactive-vendor-detection",
headers={"host": f"{middleware_inactive_vendor.subdomain}.platform.com"},
headers={
"host": f"{middleware_inactive_vendor.subdomain}.platform.com"
},
)
assert response.status_code == 200

1
tests/integration/security/test_authentication.py
View File

@@ -6,6 +6,7 @@ API Structure:
- /api/v1/admin/* - Admin endpoints (require admin token)
- /api/v1/vendor/* - Vendor endpoints (require vendor token with vendor_id claim)
"""
import pytest

9
tests/integration/security/test_authorization.py
View File

@@ -6,6 +6,7 @@ Tests role-based access control:
- Admin endpoints require admin role
- Vendor endpoints require vendor context (vendor_id in token)
"""
import pytest
@@ -29,7 +30,9 @@ class TestAuthorization:
for endpoint in admin_endpoints:
response = client.get(endpoint, headers=admin_headers)
assert response.status_code == 200, f"Admin should have access to {endpoint}"
assert response.status_code == 200, (
f"Admin should have access to {endpoint}"
)
def test_vendor_endpoint_requires_vendor_context(self, client, admin_headers):
"""Test that vendor endpoints require vendor context in token"""
@@ -38,9 +41,7 @@ class TestAuthorization:
# Should fail - admin token lacks vendor_id claim
assert response.status_code in [401, 403]
def test_vendor_owner_access_control(
self, client, admin_headers, test_vendor
):
def test_vendor_owner_access_control(self, client, admin_headers, test_vendor):
"""Test admin can access vendor by vendor code"""
response = client.get(
f"/api/v1/admin/vendors/{test_vendor.vendor_code}", headers=admin_headers

13
tests/integration/security/test_input_validation.py
View File

@@ -4,6 +4,7 @@ Input validation tests for the API.
Tests SQL injection prevention, parameter validation, and JSON validation.
"""
import pytest
@@ -27,14 +28,10 @@ class TestInputValidation:
def test_parameter_validation(self, client, admin_headers):
"""Test parameter validation for API endpoints"""
# Test invalid pagination parameters
response = client.get(
"/api/v1/admin/products?limit=-1", headers=admin_headers
)
response = client.get("/api/v1/admin/products?limit=-1", headers=admin_headers)
assert response.status_code == 422 # Validation error
response = client.get(
"/api/v1/admin/products?skip=-1", headers=admin_headers
)
response = client.get("/api/v1/admin/products?skip=-1", headers=admin_headers)
assert response.status_code == 422 # Validation error
def test_json_validation(self, client, admin_headers, test_company):
@@ -51,8 +48,6 @@ class TestInputValidation:
response = client.post(
"/api/v1/admin/vendors",
headers=admin_headers,
json={
"name": "Test Vendor"
}, # Missing required company_id, vendor_code
json={"name": "Test Vendor"}, # Missing required company_id, vendor_code
)
assert response.status_code == 422 # Validation error

5
tests/integration/workflows/test_integration.py
View File

@@ -7,6 +7,7 @@ Tests complete workflows using actual API endpoints:
- Admin product listing workflow
- Marketplace import workflow
"""
import uuid
import pytest
@@ -49,7 +50,9 @@ class TestIntegrationFlows:
vendors = response.json()
assert any(v["vendor_code"] == vendor["vendor_code"] for v in vendors)
def test_admin_product_listing_workflow(self, client, admin_headers, test_marketplace_product):
def test_admin_product_listing_workflow(
self, client, admin_headers, test_marketplace_product
):
"""Test admin product listing and search workflow"""
# 1. List all products
response = client.get("/api/v1/admin/products", headers=admin_headers)

9
tests/performance/test_api_performance.py
View File

@@ -4,12 +4,15 @@ Performance tests for API endpoints.
Note: MarketplaceProduct now stores title/description in translations table.
"""
import time
import pytest
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
from models.database.marketplace_product_translation import (
MarketplaceProductTranslation,
)
def create_product_with_translation(db, product_id: str, title: str, **kwargs):
@@ -54,9 +57,7 @@ class TestPerformance:
# Time the request
start_time = time.time()
response = client.get(
"/api/v1/admin/products?limit=100", headers=admin_headers
)
response = client.get("/api/v1/admin/products?limit=100", headers=admin_headers)
end_time = time.time()
assert response.status_code == 200

4
tests/system/test_error_handling.py
View File

@@ -88,9 +88,7 @@ class TestErrorHandling:
def test_marketplace_product_not_found(self, client, admin_headers):
"""Test accessing non-existent marketplace product"""
response = client.get(
"/api/v1/admin/products/999999", headers=admin_headers
)
response = client.get("/api/v1/admin/products/999999", headers=admin_headers)
assert response.status_code == 404
data = response.json()

21
tests/unit/middleware/test_vendor_context.py
View File

@@ -790,10 +790,11 @@ class TestVendorContextMiddleware:
call_next = AsyncMock(return_value=Mock())
with patch.object(
VendorContextManager, "is_admin_request", return_value=False
), patch.object(
VendorContextManager, "is_static_file_request", return_value=False
with (
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
):
await middleware.dispatch(request, call_next)
@@ -836,9 +837,7 @@ class TestVendorContextMiddleware:
mock_db = MagicMock()
with (
patch.object(
VendorContextManager, "is_admin_request", return_value=False
),
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
@@ -889,9 +888,7 @@ class TestVendorContextMiddleware:
mock_db = MagicMock()
with (
patch.object(
VendorContextManager, "is_admin_request", return_value=False
),
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
@@ -928,9 +925,7 @@ class TestVendorContextMiddleware:
call_next = AsyncMock(return_value=Mock())
with (
patch.object(
VendorContextManager, "is_admin_request", return_value=False
),
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),

1
tests/unit/models/database/test_customer.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_customer.py
"""Unit tests for Customer and CustomerAddress database models."""
import pytest
from models.database.customer import Customer, CustomerAddress

5
tests/unit/models/database/test_inventory.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_inventory.py
"""Unit tests for Inventory database model."""
import pytest
from sqlalchemy.exc import IntegrityError
@@ -56,7 +57,9 @@ class TestInventoryModel:
db.add(inventory2)
db.commit()
def test_inventory_same_product_different_location(self, db, test_vendor, test_product):
def test_inventory_same_product_different_location(
self, db, test_vendor, test_product
):
"""Test same product can have inventory in different locations."""
inventory1 = Inventory(
product_id=test_product.id,

9
tests/unit/models/database/test_marketplace_import_job.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_marketplace_import_job.py
"""Unit tests for MarketplaceImportJob database model."""
import pytest
from models.database.marketplace_import_job import MarketplaceImportJob
@@ -58,7 +59,13 @@ class TestMarketplaceImportJobModel:
def test_import_job_status_values(self, db, test_user, test_vendor):
"""Test MarketplaceImportJob with different status values."""
statuses = ["pending", "processing", "completed", "failed", "completed_with_errors"]
statuses = [
"pending",
"processing",
"completed",
"failed",
"completed_with_errors",
]
for i, status in enumerate(statuses):
import_job = MarketplaceImportJob(

5
tests/unit/models/database/test_marketplace_product.py
View File

@@ -1,10 +1,13 @@
# tests/unit/models/database/test_marketplace_product.py
"""Unit tests for MarketplaceProduct database model."""
import pytest
from sqlalchemy.exc import IntegrityError
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
from models.database.marketplace_product_translation import (
MarketplaceProductTranslation,
)
def _create_with_translation(db, marketplace_product_id, title, **kwargs):

14
tests/unit/models/database/test_order.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_order.py
"""Unit tests for Order and OrderItem database models."""
import pytest
from sqlalchemy.exc import IntegrityError
@@ -74,7 +75,14 @@ class TestOrderModel:
self, db, test_vendor, test_customer, test_customer_address
):
"""Test Order with different status values."""
statuses = ["pending", "confirmed", "processing", "shipped", "delivered", "cancelled"]
statuses = [
"pending",
"confirmed",
"processing",
"shipped",
"delivered",
"cancelled",
]
for i, status in enumerate(statuses):
order = Order(
@@ -93,9 +101,7 @@ class TestOrderModel:
assert order.status == status
def test_order_amounts(
self, db, test_vendor, test_customer, test_customer_address
):
def test_order_amounts(self, db, test_vendor, test_customer, test_customer_address):
"""Test Order amount fields."""
order = Order(
vendor_id=test_vendor.id,

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

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_product.py
"""Unit tests for Product (vendor catalog) database model."""
import pytest
from sqlalchemy.exc import IntegrityError
@@ -35,10 +36,9 @@ class TestProductModel:
assert product.is_featured is True
assert product.vendor.vendor_code == test_vendor.vendor_code
# Use get_title() method instead of .title attribute
assert (
product.marketplace_product.get_title("en")
== test_marketplace_product.get_title("en")
)
assert product.marketplace_product.get_title(
"en"
) == test_marketplace_product.get_title("en")
def test_product_unique_per_vendor(self, db, test_vendor, test_marketplace_product):
"""Test that same marketplace product can't be added twice to vendor catalog."""
@@ -75,7 +75,9 @@ class TestProductModel:
assert product.min_quantity == 1 # Default
assert product.display_order == 0 # Default
def test_product_vendor_override_fields(self, db, test_vendor, test_marketplace_product):
def test_product_vendor_override_fields(
self, db, test_vendor, test_marketplace_product
):
"""Test Product model vendor-specific override fields."""
product = Product(
vendor_id=test_vendor.id,
@@ -97,7 +99,9 @@ class TestProductModel:
assert product.currency == "USD"
assert product.availability == "limited"
def test_product_inventory_settings(self, db, test_vendor, test_marketplace_product):
def test_product_inventory_settings(
self, db, test_vendor, test_marketplace_product
):
"""Test Product model inventory settings."""
product = Product(
vendor_id=test_vendor.id,
@@ -126,7 +130,9 @@ class TestProductModel:
assert product.marketplace_product is not None
assert product.inventory_entries == [] # No inventory yet
def test_product_effective_properties(self, db, test_vendor, test_marketplace_product):
def test_product_effective_properties(
self, db, test_vendor, test_marketplace_product
):
"""Test Product effective properties with override pattern."""
# First, set some values on the marketplace product
test_marketplace_product.price_numeric = 100.00

5
tests/unit/models/database/test_team.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_team.py
"""Unit tests for VendorUser and Role database models."""
import pytest
from models.database.vendor import Role, Vendor, VendorUser
@@ -87,7 +88,9 @@ class TestVendorUserModel:
assert vendor_user.role.name == "Manager"
assert "products.create" in vendor_user.role.permissions
def test_vendor_user_multiple_vendors(self, db, test_vendor, test_user, other_company):
def test_vendor_user_multiple_vendors(
self, db, test_vendor, test_user, other_company
):
"""Test same user can be added to multiple vendors."""
# Create another vendor
other_vendor = Vendor(

1
tests/unit/models/database/test_user.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_user.py
"""Unit tests for User database model."""
import pytest
from sqlalchemy.exc import IntegrityError

1
tests/unit/models/database/test_vendor.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/database/test_vendor.py
"""Unit tests for Vendor database model."""
import pytest
from sqlalchemy.exc import IntegrityError

1
tests/unit/models/schema/test_auth.py
View File

@@ -1,5 +1,6 @@
# tests/unit/models/schema/test_auth.py
"""Unit tests for auth Pydantic schemas."""
import pytest
from pydantic import ValidationError

9
tests/unit/models/schema/test_customer.py
View File

@@ -1,16 +1,17 @@
# tests/unit/models/schema/test_customer.py
"""Unit tests for customer Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.customer import (
CustomerRegister,
CustomerUpdate,
CustomerResponse,
CustomerAddressCreate,
CustomerAddressUpdate,
CustomerAddressResponse,
CustomerAddressUpdate,
CustomerPreferencesUpdate,
CustomerRegister,
CustomerResponse,
CustomerUpdate,
)

8
tests/unit/models/schema/test_inventory.py
View File

@@ -1,16 +1,16 @@
# tests/unit/models/schema/test_inventory.py
"""Unit tests for inventory Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.inventory import (
InventoryBase,
InventoryCreate,
InventoryAdjust,
InventoryUpdate,
InventoryCreate,
InventoryLocationResponse,
InventoryReserve,
InventoryResponse,
InventoryLocationResponse,
InventoryUpdate,
ProductInventorySummary,
)

3
tests/unit/models/schema/test_marketplace_import_job.py
View File

@@ -1,12 +1,13 @@
# tests/unit/models/schema/test_marketplace_import_job.py
"""Unit tests for marketplace import job Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.marketplace_import_job import (
MarketplaceImportJobListResponse,
MarketplaceImportJobRequest,
MarketplaceImportJobResponse,
MarketplaceImportJobListResponse,
MarketplaceImportJobStatusUpdate,
)

20
tests/unit/models/schema/test_order.py
View File

@@ -1,18 +1,17 @@
# tests/unit/models/schema/test_order.py
"""Unit tests for order Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.order import (
OrderAddressCreate,
OrderCreate,
OrderItemCreate,
OrderItemResponse,
OrderAddressCreate,
OrderAddressResponse,
OrderCreate,
OrderUpdate,
OrderResponse,
OrderDetailResponse,
OrderListResponse,
OrderResponse,
OrderUpdate,
)
@@ -253,7 +252,14 @@ class TestOrderUpdateSchema:
def test_valid_status_values(self):
"""Test all valid status values."""
valid_statuses = ["pending", "processing", "shipped", "delivered", "cancelled", "refunded"]
valid_statuses = [
"pending",
"processing",
"shipped",
"delivered",
"cancelled",
"refunded",
]
for status in valid_statuses:
update = OrderUpdate(status=status)
assert update.status == status

6
tests/unit/models/schema/test_product.py
View File

@@ -1,14 +1,14 @@
# tests/unit/models/schema/test_product.py
"""Unit tests for product Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.product import (
ProductCreate,
ProductUpdate,
ProductResponse,
ProductDetailResponse,
ProductListResponse,
ProductResponse,
ProductUpdate,
)

5
tests/unit/models/schema/test_vendor.py
View File

@@ -1,15 +1,16 @@
# tests/unit/models/schema/test_vendor.py
"""Unit tests for vendor Pydantic schemas."""
import pytest
from pydantic import ValidationError
from models.schema.vendor import (
VendorCreate,
VendorUpdate,
VendorResponse,
VendorDetailResponse,
VendorListResponse,
VendorResponse,
VendorSummary,
VendorUpdate,
)

9
tests/unit/services/test_admin_service.py
View File

@@ -122,7 +122,9 @@ class TestAdminService:
from models.database.vendor import Vendor
# Re-query vendor to get fresh instance
vendor_to_unverify = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
vendor_to_unverify = (
db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
)
vendor_to_unverify.is_verified = False
db.commit()
@@ -186,7 +188,10 @@ class TestAdminService:
None,
)
assert test_job is not None
assert test_job.marketplace.lower() == test_marketplace_import_job.marketplace.lower()
assert (
test_job.marketplace.lower()
== test_marketplace_import_job.marketplace.lower()
)
assert test_job.status == test_marketplace_import_job.status
def test_get_marketplace_import_jobs_with_marketplace_filter(

1
tests/unit/services/test_auth_service.py
View File

@@ -1,5 +1,6 @@
# tests/unit/services/test_auth_service.py
"""Unit tests for AuthService - login and password hashing."""
import pytest
from app.exceptions.auth import (

52
tests/unit/services/test_inventory_service.py
View File

@@ -1,5 +1,6 @@
# tests/unit/services/test_inventory_service.py
"""Unit tests for InventoryService."""
import uuid
import pytest
@@ -352,23 +353,21 @@ class TestInventoryService:
self, db, test_inventory, test_product, test_vendor
):
"""Test getting product inventory summary."""
result = self.service.get_product_inventory(
db, test_vendor.id, test_product.id
)
result = self.service.get_product_inventory(db, test_vendor.id, test_product.id)
assert result.product_id == test_product.id
assert result.vendor_id == test_vendor.id
assert result.total_quantity >= test_inventory.quantity
assert len(result.locations) >= 1
def test_get_product_inventory_no_inventory(
self, db, test_product, test_vendor
):
def test_get_product_inventory_no_inventory(self, db, test_product, test_vendor):
"""Test getting inventory for product with no inventory entries."""
# Create a new product without inventory
from models.database.product import Product
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
from models.database.marketplace_product_translation import (
MarketplaceProductTranslation,
)
from models.database.product import Product
unique_id = str(uuid.uuid4())[:8]
mp = MarketplaceProduct(
@@ -412,9 +411,7 @@ class TestInventoryService:
# ==================== Get Vendor Inventory Tests ====================
def test_get_vendor_inventory_success(
self, db, test_inventory, test_vendor
):
def test_get_vendor_inventory_success(self, db, test_inventory, test_vendor):
"""Test getting all vendor inventory."""
result = self.service.get_vendor_inventory(db, test_vendor.id)
@@ -433,9 +430,7 @@ class TestInventoryService:
for inv in result:
assert test_inventory.location[:10].upper() in inv.location.upper()
def test_get_vendor_inventory_with_low_stock_filter(
self, db, test_vendor
):
def test_get_vendor_inventory_with_low_stock_filter(self, db, test_vendor):
"""Test getting vendor inventory filtered by low stock threshold."""
result = self.service.get_vendor_inventory(
db, test_vendor.id, low_stock_threshold=5
@@ -446,17 +441,13 @@ class TestInventoryService:
def test_get_vendor_inventory_pagination(self, db, test_vendor):
"""Test vendor inventory pagination."""
result = self.service.get_vendor_inventory(
db, test_vendor.id, skip=0, limit=10
)
result = self.service.get_vendor_inventory(db, test_vendor.id, skip=0, limit=10)
assert len(result) <= 10
# ==================== Update Inventory Tests ====================
def test_update_inventory_quantity(
self, db, test_inventory, test_vendor
):
def test_update_inventory_quantity(self, db, test_inventory, test_vendor):
"""Test updating inventory quantity."""
inventory_update = InventoryUpdate(quantity=500)
@@ -466,9 +457,7 @@ class TestInventoryService:
assert result.quantity == 500
def test_update_inventory_reserved_quantity(
self, db, test_inventory, test_vendor
):
def test_update_inventory_reserved_quantity(self, db, test_inventory, test_vendor):
"""Test updating inventory reserved quantity."""
inventory_update = InventoryUpdate(reserved_quantity=20)
@@ -478,9 +467,7 @@ class TestInventoryService:
assert result.reserved_quantity == 20
def test_update_inventory_location(
self, db, test_inventory, test_vendor
):
def test_update_inventory_location(self, db, test_inventory, test_vendor):
"""Test updating inventory location."""
unique_id = str(uuid.uuid4())[:8].upper()
new_location = f"NEW_LOCATION_{unique_id}"
@@ -499,9 +486,7 @@ class TestInventoryService:
with pytest.raises(InventoryNotFoundException):
self.service.update_inventory(db, test_vendor.id, 99999, inventory_update)
def test_update_inventory_wrong_vendor(
self, db, test_inventory, other_company
):
def test_update_inventory_wrong_vendor(self, db, test_inventory, other_company):
"""Test updating inventory from wrong vendor raises InventoryNotFoundException."""
from models.database.vendor import Vendor
@@ -525,9 +510,7 @@ class TestInventoryService:
# ==================== Delete Inventory Tests ====================
def test_delete_inventory_success(
self, db, test_inventory, test_vendor
):
def test_delete_inventory_success(self, db, test_inventory, test_vendor):
"""Test deleting inventory entry."""
inventory_id = test_inventory.id
@@ -544,9 +527,7 @@ class TestInventoryService:
with pytest.raises(InventoryNotFoundException):
self.service.delete_inventory(db, test_vendor.id, 99999)
def test_delete_inventory_wrong_vendor(
self, db, test_inventory, other_company
):
def test_delete_inventory_wrong_vendor(self, db, test_inventory, other_company):
"""Test deleting inventory from wrong vendor raises InventoryNotFoundException."""
from models.database.vendor import Vendor
@@ -563,4 +544,3 @@ class TestInventoryService:
with pytest.raises(InventoryNotFoundException):
self.service.delete_inventory(db, other_vendor.id, test_inventory.id)

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

@@ -8,26 +8,22 @@ Tests cover:
- GraphQL client (mocked)
"""
import pytest
from unittest.mock import MagicMock, patch
import pytest
from app.services.letzshop import (
LetzshopClient,
LetzshopClientError,
LetzshopAuthError,
LetzshopAPIError,
LetzshopCredentialsService,
CredentialsNotFoundError,
LetzshopAPIError,
LetzshopClient,
LetzshopCredentialsService,
)
from app.utils.encryption import (
EncryptionService,
EncryptionError,
encrypt_value,
decrypt_value,
EncryptionService,
mask_api_key,
)
# ============================================================================
# Encryption Tests
# ============================================================================
@@ -339,9 +335,7 @@ class TestLetzshopClient:
"""Test successful connection test."""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": {"__typename": "Query"}
}
mock_response.json.return_value = {"data": {"__typename": "Query"}}
mock_post.return_value = mock_response
client = LetzshopClient(api_key="test-key")
@@ -398,7 +392,7 @@ class TestLetzshopClient:
"inventoryUnits": [
{"id": "unit_1", "state": "confirmed"},
],
"errors": []
"errors": [],
}
}
}
@@ -420,12 +414,9 @@ class TestLetzshopClient:
"setShipmentTracking": {
"shipment": {
"id": "ship_1",
"tracking": {
"code": "1Z999AA1",
"provider": "ups"
}
"tracking": {"code": "1Z999AA1", "provider": "ups"},
},
"errors": []
"errors": [],
}
}
}
@@ -446,9 +437,7 @@ class TestLetzshopClient:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"errors": [
{"message": "Invalid shipment ID"}
]
"errors": [{"message": "Invalid shipment ID"}]
}
mock_post.return_value = mock_response

21
tests/unit/services/test_marketplace_service.py
View File

@@ -1,9 +1,9 @@
# tests/unit/services/test_marketplace_service.py
"""Unit tests for MarketplaceImportJobService."""
import uuid
import pytest
from unittest.mock import patch
from app.exceptions.base import ValidationException
from app.exceptions.marketplace_import_job import (
@@ -52,7 +52,9 @@ class TestMarketplaceImportJobService:
assert result.marketplace == "Letzshop" # Default
def test_create_import_job_database_error(self, db, test_vendor, test_user, monkeypatch):
def test_create_import_job_database_error(
self, db, test_vendor, test_user, monkeypatch
):
"""Test import job creation handles database errors."""
request = MarketplaceImportJobRequest(
source_url="https://example.com/products.csv",
@@ -119,6 +121,7 @@ class TestMarketplaceImportJobService:
def test_get_import_job_by_id_database_error(self, db, test_user, monkeypatch):
"""Test get import job handles database errors."""
def mock_query(*args):
raise Exception("Database query failed")
@@ -200,7 +203,10 @@ class TestMarketplaceImportJobService:
):
"""Test getting import jobs with marketplace filter."""
jobs = self.service.get_import_jobs(
db, test_vendor, test_user, marketplace=test_marketplace_import_job.marketplace
db,
test_vendor,
test_user,
marketplace=test_marketplace_import_job.marketplace,
)
assert len(jobs) >= 1
@@ -254,8 +260,11 @@ class TestMarketplaceImportJobService:
assert len(jobs) == 0
def test_get_import_jobs_database_error(self, db, test_vendor, test_user, monkeypatch):
def test_get_import_jobs_database_error(
self, db, test_vendor, test_user, monkeypatch
):
"""Test get import jobs handles database errors."""
def mock_query(*args):
raise Exception("Database query failed")
@@ -270,7 +279,9 @@ class TestMarketplaceImportJobService:
# ==================== convert_to_response_model Tests ====================
def test_convert_to_response_model(self, db, test_marketplace_import_job, test_vendor):
def test_convert_to_response_model(
self, db, test_marketplace_import_job, test_vendor
):
"""Test converting database model to response model."""
from models.database.marketplace_import_job import MarketplaceImportJob as MIJ

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

@@ -108,8 +108,7 @@ class TestProductService:
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
exc_info.value.details.get("marketplace_product_id") == existing_product_id
)
def test_create_product_invalid_price(self, db):
@@ -224,7 +223,9 @@ 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_preserves_existing(self, db, test_marketplace_product):
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="")
@@ -275,7 +276,9 @@ class TestProductService:
self, db, test_marketplace_product_with_inventory
):
"""Test getting inventory info for product with inventory."""
marketplace_product = test_marketplace_product_with_inventory["marketplace_product"]
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)
@@ -352,7 +355,10 @@ class TestMarketplaceProductServiceAdmin:
# Find our test product in results
found = False
for p in products:
if p["marketplace_product_id"] == test_marketplace_product.marketplace_product_id:
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
@@ -429,9 +435,7 @@ class TestMarketplaceProductServiceAdmin:
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
)
product = self.service.get_admin_product_detail(db, test_marketplace_product.id)
assert product["id"] == test_marketplace_product.id
assert (

26
tests/unit/services/test_stats_service.py
View File

@@ -1,19 +1,19 @@
# tests/unit/services/test_stats_service.py
"""Unit tests for StatsService following the application's testing patterns."""
import uuid
from unittest.mock import patch
import pytest
from sqlalchemy.exc import SQLAlchemyError
from unittest.mock import patch, MagicMock
from app.exceptions import AdminOperationException, VendorNotFoundException
from app.services.stats_service import StatsService
from models.database.inventory import Inventory
from models.database.marketplace_import_job import MarketplaceImportJob
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
from models.database.product import Product
from models.database.vendor import Vendor
from models.database.marketplace_product_translation import (
MarketplaceProductTranslation,
)
def create_marketplace_product_with_translation(
@@ -21,8 +21,7 @@ def create_marketplace_product_with_translation(
):
"""Helper to create a MarketplaceProduct with its translation."""
product = MarketplaceProduct(
marketplace_product_id=marketplace_product_id,
**kwargs
marketplace_product_id=marketplace_product_id, **kwargs
)
db.add(product)
db.flush() # Get the product ID
@@ -275,7 +274,10 @@ class TestStatsService:
with pytest.raises(AdminOperationException) as exc_info:
self.service.get_marketplace_breakdown_stats(db)
assert exc_info.value.details.get("operation") == "get_marketplace_breakdown_stats"
assert (
exc_info.value.details.get("operation")
== "get_marketplace_breakdown_stats"
)
# ==================== get_vendor_stats Tests ====================
@@ -371,9 +373,7 @@ class TestStatsService:
stats = self.service.get_vendor_statistics(db)
if stats["total_vendors"] > 0:
expected_rate = (
stats["verified_vendors"] / stats["total_vendors"] * 100
)
expected_rate = stats["verified_vendors"] / stats["total_vendors"] * 100
assert abs(stats["verification_rate"] - expected_rate) < 0.01
def test_get_vendor_statistics_database_error(self, db):
@@ -438,9 +438,7 @@ class TestStatsService:
stats = self.service.get_import_statistics(db)
if stats["total_imports"] > 0:
expected_rate = (
stats["completed_imports"] / stats["total_imports"] * 100
)
expected_rate = stats["completed_imports"] / stats["total_imports"] * 100
assert abs(stats["success_rate"] - expected_rate) < 0.01
else:
assert stats["success_rate"] == 0

5
tests/unit/services/test_vendor_product_service.py
View File

@@ -4,6 +4,7 @@ Unit tests for VendorProductService.
Tests the vendor product catalog service operations.
"""
import pytest
from app.exceptions import ProductNotFoundException
@@ -31,7 +32,9 @@ class TestVendorProductService:
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
assert (
p["marketplace_product_id"] == test_product.marketplace_product_id
)
break
assert found, "Test product not found in results"

18
tests/unit/services/test_vendor_service.py
View File

@@ -1,5 +1,6 @@
# tests/unit/services/test_vendor_service.py
"""Unit tests for VendorService following the application's exception patterns."""
import uuid
import pytest
@@ -85,7 +86,9 @@ class TestVendorService:
assert vendor.is_verified is True # Admin creates verified vendor
def test_create_vendor_duplicate_code(self, db, test_user, test_company, test_vendor):
def test_create_vendor_duplicate_code(
self, db, test_user, test_company, test_vendor
):
"""Test vendor creation fails with duplicate vendor code."""
vendor_data = VendorCreate(
company_id=test_company.id,
@@ -359,9 +362,11 @@ class TestVendorService:
# Re-query objects to avoid session issues
vendor = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
mp = db.query(MarketplaceProduct).filter(
MarketplaceProduct.id == unique_product.id
).first()
mp = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.id == unique_product.id)
.first()
)
product_data = ProductCreate(
marketplace_product_id=mp.id,
@@ -394,6 +399,7 @@ class TestVendorService:
"""Test adding product that's already in vendor fails."""
# Re-query to get fresh instances
from models.database.product import Product
vendor = db.query(Vendor).filter(Vendor.id == test_vendor.id).first()
product = db.query(Product).filter(Product.id == test_product.id).first()
@@ -469,7 +475,9 @@ class TestVendorServiceExceptionDetails:
def setup_method(self):
self.service = VendorService()
def test_exception_to_dict_structure(self, db, test_user, test_vendor, test_company):
def test_exception_to_dict_structure(
self, db, test_user, test_vendor, test_company
):
"""Test that exceptions can be properly serialized to dict for API responses."""
vendor_data = VendorCreate(
company_id=test_company.id,

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

@@ -10,7 +10,6 @@ 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