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

marketplace refactoring

Browse Source
This commit is contained in:
Samir Boulahtit
2025-10-04 13:38:10 +02:00
parent 32be301d83
commit c971674ec2
68 changed files with 1102 additions and 1128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/unit/middleware/test_middleware.py
View File

@@ -41,7 +41,8 @@ class TestRateLimiter:
# Next request should be blocked
assert limiter.allow_request(client_id, max_requests, 3600) is False
@pytest.mark.unit
@pytest.mark.auth # for auth manager tests
class TestAuthManager:
def test_password_hashing_and_verification(self):
"""Test password hashing and verification"""

26
tests/unit/models/test_database_models.py
View File

@@ -1,7 +1,7 @@
# tests/unit/models/test_database_models.py
import pytest
from models.database.product import Product
from models.database.marketplace_product import MarketplaceProduct
from models.database.shop import Shop
from models.database.stock import Stock
from models.database.user import User
@@ -30,10 +30,10 @@ class TestDatabaseModels:
assert user.updated_at is not None
def test_product_model(self, db):
"""Test Product model creation"""
product = Product(
product_id="DB_TEST_001",
title="Database Test Product",
"""Test MarketplaceProduct model creation"""
marketplace_product = MarketplaceProduct(
marketplace_product_id="DB_TEST_001",
title="Database Test MarketplaceProduct",
description="Testing product model",
price="25.99",
currency="USD",
@@ -44,13 +44,13 @@ class TestDatabaseModels:
shop_name="DBTestShop",
)
db.add(product)
db.add(marketplace_product)
db.commit()
db.refresh(product)
db.refresh(marketplace_product)
assert product.id is not None
assert product.product_id == "DB_TEST_001"
assert product.created_at is not None
assert marketplace_product.id is not None
assert marketplace_product.marketplace_product_id == "DB_TEST_001"
assert marketplace_product.created_at is not None
def test_stock_model(self, db):
"""Test Stock model creation"""
@@ -87,13 +87,13 @@ class TestDatabaseModels:
def test_database_constraints(self, db):
"""Test database constraints and unique indexes"""
# Test unique product_id constraint
product1 = Product(product_id="UNIQUE_001", title="Product 1")
# Test unique marketplace_product_id constraint
product1 = MarketplaceProduct(marketplace_product_id="UNIQUE_001", title="MarketplaceProduct 1")
db.add(product1)
db.commit()
# This should raise an integrity error
with pytest.raises(Exception): # Could be IntegrityError or similar
product2 = Product(product_id="UNIQUE_001", title="Product 2")
product2 = MarketplaceProduct(marketplace_product_id="UNIQUE_001", title="MarketplaceProduct 2")
db.add(product2)
db.commit()

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

@@ -10,7 +10,7 @@ from app.exceptions import (
AdminOperationException,
)
from app.services.admin_service import AdminService
from models.database.marketplace import MarketplaceImportJob
from models.database.marketplace_import_job import MarketplaceImportJob
from models.database.shop import Shop
@@ -169,51 +169,51 @@ class TestAdminService:
assert exception.error_code == "SHOP_NOT_FOUND"
# Marketplace Import Jobs Tests
def test_get_marketplace_import_jobs_no_filters(self, db, test_marketplace_job):
def test_get_marketplace_import_jobs_no_filters(self, db, test_marketplace_import_job):
"""Test getting marketplace import jobs without filters"""
result = self.service.get_marketplace_import_jobs(db, skip=0, limit=10)
assert len(result) >= 1
# Find our test job in the results
test_job = next(
(job for job in result if job.job_id == test_marketplace_job.id), None
(job for job in result if job.job_id == test_marketplace_import_job.id), None
)
assert test_job is not None
assert test_job.marketplace == test_marketplace_job.marketplace
assert test_job.shop_name == test_marketplace_job.shop_name
assert test_job.status == test_marketplace_job.status
assert test_job.marketplace == test_marketplace_import_job.marketplace
assert test_job.shop_name == test_marketplace_import_job.shop_name
assert test_job.status == test_marketplace_import_job.status
def test_get_marketplace_import_jobs_with_marketplace_filter(self, db, test_marketplace_job):
def test_get_marketplace_import_jobs_with_marketplace_filter(self, db, test_marketplace_import_job):
"""Test filtering marketplace import jobs by marketplace"""
result = self.service.get_marketplace_import_jobs(
db, marketplace=test_marketplace_job.marketplace, skip=0, limit=10
db, marketplace=test_marketplace_import_job.marketplace, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert test_marketplace_job.marketplace.lower() in job.marketplace.lower()
assert test_marketplace_import_job.marketplace.lower() in job.marketplace.lower()
def test_get_marketplace_import_jobs_with_shop_filter(self, db, test_marketplace_job):
def test_get_marketplace_import_jobs_with_shop_filter(self, db, test_marketplace_import_job):
"""Test filtering marketplace import jobs by shop name"""
result = self.service.get_marketplace_import_jobs(
db, shop_name=test_marketplace_job.shop_name, skip=0, limit=10
db, shop_name=test_marketplace_import_job.shop_name, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert test_marketplace_job.shop_name.lower() in job.shop_name.lower()
assert test_marketplace_import_job.shop_name.lower() in job.shop_name.lower()
def test_get_marketplace_import_jobs_with_status_filter(self, db, test_marketplace_job):
def test_get_marketplace_import_jobs_with_status_filter(self, db, test_marketplace_import_job):
"""Test filtering marketplace import jobs by status"""
result = self.service.get_marketplace_import_jobs(
db, status=test_marketplace_job.status, skip=0, limit=10
db, status=test_marketplace_import_job.status, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert job.status == test_marketplace_job.status
assert job.status == test_marketplace_import_job.status
def test_get_marketplace_import_jobs_pagination(self, db, test_marketplace_job):
def test_get_marketplace_import_jobs_pagination(self, db, test_marketplace_import_job):
"""Test marketplace import jobs pagination"""
result_page1 = self.service.get_marketplace_import_jobs(db, skip=0, limit=1)
result_page2 = self.service.get_marketplace_import_jobs(db, skip=1, limit=1)

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

@@ -4,7 +4,7 @@ from datetime import datetime
import pytest
from app.exceptions.marketplace import (
from app.exceptions.marketplace_import_job import (
ImportJobNotFoundException,
ImportJobNotOwnedException,
ImportJobCannotBeCancelledException,
@@ -12,9 +12,9 @@ from app.exceptions.marketplace import (
)
from app.exceptions.shop import ShopNotFoundException, UnauthorizedShopAccessException
from app.exceptions.base import ValidationException
from app.services.marketplace_service import MarketplaceService
from models.schemas.marketplace import MarketplaceImportRequest
from models.database.marketplace import MarketplaceImportJob
from app.services.marketplace_import_job_service import MarketplaceImportJobService
from models.schemas.marketplace_import_job import MarketplaceImportJobRequest
from models.database.marketplace_import_job import MarketplaceImportJob
from models.database.shop import Shop
from models.database.user import User
@@ -23,7 +23,7 @@ from models.database.user import User
@pytest.mark.marketplace
class TestMarketplaceService:
def setup_method(self):
self.service = MarketplaceService()
self.service = MarketplaceImportJobService()
def test_validate_shop_access_success(self, db, test_shop, test_user):
"""Test successful shop access validation"""
@@ -76,7 +76,7 @@ class TestMarketplaceService:
test_shop.owner_id = test_user.id
db.commit()
request = MarketplaceImportRequest(
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code=test_shop.shop_code,
@@ -94,7 +94,7 @@ class TestMarketplaceService:
def test_create_import_job_invalid_shop(self, db, test_user):
"""Test import job creation with invalid shop"""
request = MarketplaceImportRequest(
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code="INVALID_SHOP",
@@ -114,7 +114,7 @@ class TestMarketplaceService:
test_shop.owner_id = other_user.id
db.commit()
request = MarketplaceImportRequest(
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code=test_shop.shop_code,
@@ -127,24 +127,24 @@ class TestMarketplaceService:
exception = exc_info.value
assert exception.error_code == "UNAUTHORIZED_SHOP_ACCESS"
def test_get_import_job_by_id_success(self, db, test_marketplace_job, test_user):
def test_get_import_job_by_id_success(self, db, test_marketplace_import_job, test_user):
"""Test getting import job by ID for job owner"""
result = self.service.get_import_job_by_id(
db, test_marketplace_job.id, test_user
db, test_marketplace_import_job.id, test_user
)
assert result.id == test_marketplace_job.id
assert result.id == test_marketplace_import_job.id
assert result.user_id == test_user.id
def test_get_import_job_by_id_admin_access(
self, db, test_marketplace_job, test_admin
self, db, test_marketplace_import_job, test_admin
):
"""Test that admin can access any import job"""
result = self.service.get_import_job_by_id(
db, test_marketplace_job.id, test_admin
db, test_marketplace_import_job.id, test_admin
)
assert result.id == test_marketplace_job.id
assert result.id == test_marketplace_import_job.id
def test_get_import_job_by_id_not_found(self, db, test_user):
"""Test getting non-existent import job"""
@@ -157,42 +157,42 @@ class TestMarketplaceService:
assert "99999" in exception.message
def test_get_import_job_by_id_access_denied(
self, db, test_marketplace_job, other_user
self, db, test_marketplace_import_job, other_user
):
"""Test access denied when user doesn't own the job"""
with pytest.raises(ImportJobNotOwnedException) as exc_info:
self.service.get_import_job_by_id(db, test_marketplace_job.id, other_user)
self.service.get_import_job_by_id(db, test_marketplace_import_job.id, other_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_OWNED"
assert exception.status_code == 403
assert str(test_marketplace_job.id) in exception.message
assert str(test_marketplace_import_job.id) in exception.message
def test_get_import_jobs_user_filter(self, db, test_marketplace_job, test_user):
def test_get_import_jobs_user_filter(self, db, test_marketplace_import_job, test_user):
"""Test getting import jobs filtered by user"""
jobs = self.service.get_import_jobs(db, test_user)
assert len(jobs) >= 1
assert any(job.id == test_marketplace_job.id for job in jobs)
assert test_marketplace_job.user_id == test_user.id
assert any(job.id == test_marketplace_import_job.id for job in jobs)
assert test_marketplace_import_job.user_id == test_user.id
def test_get_import_jobs_admin_sees_all(self, db, test_marketplace_job, test_admin):
def test_get_import_jobs_admin_sees_all(self, db, test_marketplace_import_job, test_admin):
"""Test that admin sees all import jobs"""
jobs = self.service.get_import_jobs(db, test_admin)
assert len(jobs) >= 1
assert any(job.id == test_marketplace_job.id for job in jobs)
assert any(job.id == test_marketplace_import_job.id for job in jobs)
def test_get_import_jobs_with_marketplace_filter(
self, db, test_marketplace_job, test_user
self, db, test_marketplace_import_job, test_user
):
"""Test getting import jobs with marketplace filter"""
jobs = self.service.get_import_jobs(
db, test_user, marketplace=test_marketplace_job.marketplace
db, test_user, marketplace=test_marketplace_import_job.marketplace
)
assert len(jobs) >= 1
assert any(job.marketplace == test_marketplace_job.marketplace for job in jobs)
assert any(job.marketplace == test_marketplace_import_job.marketplace for job in jobs)
def test_get_import_jobs_with_pagination(self, db, test_user, test_shop):
"""Test getting import jobs with pagination"""
@@ -228,11 +228,11 @@ class TestMarketplaceService:
assert exception.error_code == "VALIDATION_ERROR"
assert "Failed to retrieve import jobs" in exception.message
def test_update_job_status_success(self, db, test_marketplace_job):
def test_update_job_status_success(self, db, test_marketplace_import_job):
"""Test updating job status"""
result = self.service.update_job_status(
db,
test_marketplace_job.id,
test_marketplace_import_job.id,
"completed",
imported_count=100,
total_processed=100,
@@ -260,7 +260,7 @@ class TestMarketplaceService:
assert exception.error_code == "VALIDATION_ERROR"
assert "Failed to update job status" in exception.message
def test_get_job_stats_user(self, db, test_marketplace_job, test_user):
def test_get_job_stats_user(self, db, test_marketplace_import_job, test_user):
"""Test getting job statistics for user"""
stats = self.service.get_job_stats(db, test_user)
@@ -271,7 +271,7 @@ class TestMarketplaceService:
assert "failed_jobs" in stats
assert isinstance(stats["total_jobs"], int)
def test_get_job_stats_admin(self, db, test_marketplace_job, test_admin):
def test_get_job_stats_admin(self, db, test_marketplace_import_job, test_admin):
"""Test getting job statistics for admin"""
stats = self.service.get_job_stats(db, test_admin)
@@ -287,14 +287,14 @@ class TestMarketplaceService:
assert exception.error_code == "VALIDATION_ERROR"
assert "Failed to retrieve job statistics" in exception.message
def test_convert_to_response_model(self, test_marketplace_job):
def test_convert_to_response_model(self, test_marketplace_import_job):
"""Test converting database model to response model"""
response = self.service.convert_to_response_model(test_marketplace_job)
response = self.service.convert_to_response_model(test_marketplace_import_job)
assert response.job_id == test_marketplace_job.id
assert response.status == test_marketplace_job.status
assert response.marketplace == test_marketplace_job.marketplace
assert response.imported == (test_marketplace_job.imported_count or 0)
assert response.job_id == test_marketplace_import_job.id
assert response.status == test_marketplace_import_job.status
assert response.marketplace == test_marketplace_import_job.marketplace
assert response.imported == (test_marketplace_import_job.imported_count or 0)
def test_cancel_import_job_success(self, db, test_user, test_shop):
"""Test cancelling a pending import job"""
@@ -330,24 +330,24 @@ class TestMarketplaceService:
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_FOUND"
def test_cancel_import_job_access_denied(self, db, test_marketplace_job, other_user):
def test_cancel_import_job_access_denied(self, db, test_marketplace_import_job, other_user):
"""Test cancelling import job without access"""
with pytest.raises(ImportJobNotOwnedException) as exc_info:
self.service.cancel_import_job(db, test_marketplace_job.id, other_user)
self.service.cancel_import_job(db, test_marketplace_import_job.id, other_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_OWNED"
def test_cancel_import_job_invalid_status(
self, db, test_marketplace_job, test_user
self, db, test_marketplace_import_job, test_user
):
"""Test cancelling a job that can't be cancelled"""
# Set job status to completed
test_marketplace_job.status = "completed"
test_marketplace_import_job.status = "completed"
db.commit()
with pytest.raises(ImportJobCannotBeCancelledException) as exc_info:
self.service.cancel_import_job(db, test_marketplace_job.id, test_user)
self.service.cancel_import_job(db, test_marketplace_import_job.id, test_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_CANNOT_BE_CANCELLED"
@@ -396,10 +396,10 @@ class TestMarketplaceService:
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_FOUND"
def test_delete_import_job_access_denied(self, db, test_marketplace_job, other_user):
def test_delete_import_job_access_denied(self, db, test_marketplace_import_job, other_user):
"""Test deleting import job without access"""
with pytest.raises(ImportJobNotOwnedException) as exc_info:
self.service.delete_import_job(db, test_marketplace_job.id, other_user)
self.service.delete_import_job(db, test_marketplace_import_job.id, other_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_OWNED"
@@ -449,7 +449,7 @@ class TestMarketplaceService:
def test_create_import_job_database_error(self, db_with_error, test_user):
"""Test import job creation handles database errors"""
request = MarketplaceImportRequest(
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code="TEST_SHOP",

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

@@ -1,29 +1,29 @@
# tests/test_product_service.py
import pytest
from app.services.product_service import ProductService
from app.services.marketplace_product_service import MarketplaceProductService
from app.exceptions import (
ProductNotFoundException,
ProductAlreadyExistsException,
InvalidProductDataException,
ProductValidationException,
MarketplaceProductNotFoundException,
MarketplaceProductAlreadyExistsException,
InvalidMarketplaceProductDataException,
MarketplaceProductValidationException,
ValidationException,
)
from models.schemas.product import ProductCreate, ProductUpdate
from models.database.product import Product
from models.schemas.marketplace_product import MarketplaceProductCreate, MarketplaceProductUpdate
from models.database.marketplace_product import MarketplaceProduct
@pytest.mark.unit
@pytest.mark.products
class TestProductService:
def setup_method(self):
self.service = ProductService()
self.service = MarketplaceProductService()
def test_create_product_success(self, db):
"""Test successful product creation with valid data"""
product_data = ProductCreate(
product_id="SVC001",
title="Service Test Product",
product_data = MarketplaceProductCreate(
marketplace_product_id="SVC001",
title="Service Test MarketplaceProduct",
gtin="1234567890123",
price="19.99",
marketplace="TestMarket",
@@ -31,22 +31,22 @@ class TestProductService:
product = self.service.create_product(db, product_data)
assert product.product_id == "SVC001"
assert product.title == "Service Test Product"
assert product.marketplace_product_id == "SVC001"
assert product.title == "Service Test MarketplaceProduct"
assert product.gtin == "1234567890123"
assert product.marketplace == "TestMarket"
assert product.price == "19.99" # Price is stored as string after processing
def test_create_product_invalid_gtin(self, db):
"""Test product creation with invalid GTIN raises InvalidProductDataException"""
product_data = ProductCreate(
product_id="SVC002",
title="Service Test Product",
"""Test product creation with invalid GTIN raises InvalidMarketplaceProductDataException"""
product_data = MarketplaceProductCreate(
marketplace_product_id="SVC002",
title="Service Test MarketplaceProduct",
gtin="invalid_gtin",
price="19.99",
)
with pytest.raises(InvalidProductDataException) as exc_info:
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
self.service.create_product(db, product_data)
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
@@ -55,201 +55,201 @@ class TestProductService:
assert exc_info.value.details.get("field") == "gtin"
def test_create_product_missing_product_id(self, db):
"""Test product creation without product_id raises ProductValidationException"""
product_data = ProductCreate(
product_id="", # Empty product ID
title="Service Test Product",
"""Test product creation without marketplace_product_id raises MarketplaceProductValidationException"""
product_data = MarketplaceProductCreate(
marketplace_product_id="", # Empty product ID
title="Service Test MarketplaceProduct",
price="19.99",
)
with pytest.raises(ProductValidationException) as exc_info:
with pytest.raises(MarketplaceProductValidationException) as exc_info:
self.service.create_product(db, product_data)
assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED"
assert "Product ID is required" in str(exc_info.value)
assert exc_info.value.details.get("field") == "product_id"
assert "MarketplaceProduct ID is required" in str(exc_info.value)
assert exc_info.value.details.get("field") == "marketplace_product_id"
def test_create_product_missing_title(self, db):
"""Test product creation without title raises ProductValidationException"""
product_data = ProductCreate(
product_id="SVC003",
"""Test product creation without title raises MarketplaceProductValidationException"""
product_data = MarketplaceProductCreate(
marketplace_product_id="SVC003",
title="", # Empty title
price="19.99",
)
with pytest.raises(ProductValidationException) as exc_info:
with pytest.raises(MarketplaceProductValidationException) as exc_info:
self.service.create_product(db, product_data)
assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED"
assert "Product title is required" in str(exc_info.value)
assert "MarketplaceProduct title is required" in str(exc_info.value)
assert exc_info.value.details.get("field") == "title"
def test_create_product_already_exists(self, db, test_product):
"""Test creating product with existing ID raises ProductAlreadyExistsException"""
product_data = ProductCreate(
product_id=test_product.product_id, # Use existing product ID
title="Duplicate Product",
def test_create_product_already_exists(self, db, test_marketplace_product):
"""Test creating product with existing ID raises MarketplaceProductAlreadyExistsException"""
product_data = MarketplaceProductCreate(
marketplace_product_id=test_marketplace_product.marketplace_product_id, # Use existing product ID
title="Duplicate MarketplaceProduct",
price="29.99",
)
with pytest.raises(ProductAlreadyExistsException) as exc_info:
with pytest.raises(MarketplaceProductAlreadyExistsException) as exc_info:
self.service.create_product(db, product_data)
assert exc_info.value.error_code == "PRODUCT_ALREADY_EXISTS"
assert test_product.product_id in str(exc_info.value)
assert test_marketplace_product.marketplace_product_id in str(exc_info.value)
assert exc_info.value.status_code == 409
assert exc_info.value.details.get("product_id") == test_product.product_id
assert exc_info.value.details.get("marketplace_product_id") == test_marketplace_product.marketplace_product_id
def test_create_product_invalid_price(self, db):
"""Test product creation with invalid price raises InvalidProductDataException"""
product_data = ProductCreate(
product_id="SVC004",
title="Service Test Product",
"""Test product creation with invalid price raises InvalidMarketplaceProductDataException"""
product_data = MarketplaceProductCreate(
marketplace_product_id="SVC004",
title="Service Test MarketplaceProduct",
price="invalid_price",
)
with pytest.raises(InvalidProductDataException) as exc_info:
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
self.service.create_product(db, product_data)
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
assert "Invalid price format" in str(exc_info.value)
assert exc_info.value.details.get("field") == "price"
def test_get_product_by_id_or_raise_success(self, db, test_product):
def test_get_product_by_id_or_raise_success(self, db, test_marketplace_product):
"""Test successful product retrieval by ID"""
product = self.service.get_product_by_id_or_raise(db, test_product.product_id)
product = self.service.get_product_by_id_or_raise(db, test_marketplace_product.marketplace_product_id)
assert product.product_id == test_product.product_id
assert product.title == test_product.title
assert product.marketplace_product_id == test_marketplace_product.marketplace_product_id
assert product.title == test_marketplace_product.title
def test_get_product_by_id_or_raise_not_found(self, db):
"""Test product retrieval with non-existent ID raises ProductNotFoundException"""
with pytest.raises(ProductNotFoundException) as exc_info:
"""Test product retrieval with non-existent ID raises MarketplaceProductNotFoundException"""
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
self.service.get_product_by_id_or_raise(db, "NONEXISTENT")
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
assert "NONEXISTENT" in str(exc_info.value)
assert exc_info.value.status_code == 404
assert exc_info.value.details.get("resource_type") == "Product"
assert exc_info.value.details.get("resource_type") == "MarketplaceProduct"
assert exc_info.value.details.get("identifier") == "NONEXISTENT"
def test_get_products_with_filters_success(self, db, test_product):
def test_get_products_with_filters_success(self, db, test_marketplace_product):
"""Test getting products with various filters"""
products, total = self.service.get_products_with_filters(
db, brand=test_product.brand
db, brand=test_marketplace_product.brand
)
assert total == 1
assert len(products) == 1
assert products[0].brand == test_product.brand
assert products[0].brand == test_marketplace_product.brand
def test_get_products_with_search(self, db, test_product):
def test_get_products_with_search(self, db, test_marketplace_product):
"""Test getting products with search term"""
products, total = self.service.get_products_with_filters(
db, search="Test Product"
db, search="Test MarketplaceProduct"
)
assert total >= 1
assert len(products) >= 1
# Verify search worked by checking that title contains search term
found_product = next((p for p in products if p.product_id == test_product.product_id), None)
found_product = next((p for p in products if p.marketplace_product_id == test_marketplace_product.marketplace_product_id), None)
assert found_product is not None
def test_update_product_success(self, db, test_product):
def test_update_product_success(self, db, test_marketplace_product):
"""Test successful product update"""
update_data = ProductUpdate(
title="Updated Product Title",
update_data = MarketplaceProductUpdate(
title="Updated MarketplaceProduct Title",
price="39.99"
)
updated_product = self.service.update_product(db, test_product.product_id, update_data)
updated_product = self.service.update_product(db, test_marketplace_product.marketplace_product_id, update_data)
assert updated_product.title == "Updated Product Title"
assert updated_product.title == "Updated MarketplaceProduct Title"
assert updated_product.price == "39.99" # Price is stored as string after processing
assert updated_product.product_id == test_product.product_id # ID unchanged
assert updated_product.marketplace_product_id == test_marketplace_product.marketplace_product_id # ID unchanged
def test_update_product_not_found(self, db):
"""Test updating non-existent product raises ProductNotFoundException"""
update_data = ProductUpdate(title="Updated Title")
"""Test updating non-existent product raises MarketplaceProductNotFoundException"""
update_data = MarketplaceProductUpdate(title="Updated Title")
with pytest.raises(ProductNotFoundException) as exc_info:
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
self.service.update_product(db, "NONEXISTENT", update_data)
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
assert "NONEXISTENT" in str(exc_info.value)
def test_update_product_invalid_gtin(self, db, test_product):
"""Test updating product with invalid GTIN raises InvalidProductDataException"""
update_data = ProductUpdate(gtin="invalid_gtin")
def test_update_product_invalid_gtin(self, db, test_marketplace_product):
"""Test updating product with invalid GTIN raises InvalidMarketplaceProductDataException"""
update_data = MarketplaceProductUpdate(gtin="invalid_gtin")
with pytest.raises(InvalidProductDataException) as exc_info:
self.service.update_product(db, test_product.product_id, update_data)
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
self.service.update_product(db, test_marketplace_product.marketplace_product_id, update_data)
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
assert "Invalid GTIN format" in str(exc_info.value)
assert exc_info.value.details.get("field") == "gtin"
def test_update_product_empty_title(self, db, test_product):
"""Test updating product with empty title raises ProductValidationException"""
update_data = ProductUpdate(title="")
def test_update_product_empty_title(self, db, test_marketplace_product):
"""Test updating product with empty title raises MarketplaceProductValidationException"""
update_data = MarketplaceProductUpdate(title="")
with pytest.raises(ProductValidationException) as exc_info:
self.service.update_product(db, test_product.product_id, update_data)
with pytest.raises(MarketplaceProductValidationException) as exc_info:
self.service.update_product(db, test_marketplace_product.marketplace_product_id, update_data)
assert exc_info.value.error_code == "PRODUCT_VALIDATION_FAILED"
assert "Product title cannot be empty" in str(exc_info.value)
assert "MarketplaceProduct title cannot be empty" in str(exc_info.value)
assert exc_info.value.details.get("field") == "title"
def test_update_product_invalid_price(self, db, test_product):
"""Test updating product with invalid price raises InvalidProductDataException"""
update_data = ProductUpdate(price="invalid_price")
def test_update_product_invalid_price(self, db, test_marketplace_product):
"""Test updating product with invalid price raises InvalidMarketplaceProductDataException"""
update_data = MarketplaceProductUpdate(price="invalid_price")
with pytest.raises(InvalidProductDataException) as exc_info:
self.service.update_product(db, test_product.product_id, update_data)
with pytest.raises(InvalidMarketplaceProductDataException) as exc_info:
self.service.update_product(db, test_marketplace_product.marketplace_product_id, update_data)
assert exc_info.value.error_code == "INVALID_PRODUCT_DATA"
assert "Invalid price format" in str(exc_info.value)
assert exc_info.value.details.get("field") == "price"
def test_delete_product_success(self, db, test_product):
def test_delete_product_success(self, db, test_marketplace_product):
"""Test successful product deletion"""
result = self.service.delete_product(db, test_product.product_id)
result = self.service.delete_product(db, test_marketplace_product.marketplace_product_id)
assert result is True
# Verify product is deleted
deleted_product = self.service.get_product_by_id(db, test_product.product_id)
deleted_product = self.service.get_product_by_id(db, test_marketplace_product.marketplace_product_id)
assert deleted_product is None
def test_delete_product_not_found(self, db):
"""Test deleting non-existent product raises ProductNotFoundException"""
with pytest.raises(ProductNotFoundException) as exc_info:
"""Test deleting non-existent product raises MarketplaceProductNotFoundException"""
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
self.service.delete_product(db, "NONEXISTENT")
assert exc_info.value.error_code == "PRODUCT_NOT_FOUND"
assert "NONEXISTENT" in str(exc_info.value)
def test_get_stock_info_success(self, db, test_product_with_stock):
def test_get_stock_info_success(self, db, test_marketplace_product_with_stock):
"""Test getting stock info for product with stock"""
# Extract the product from the dictionary
product = test_product_with_stock['product']
marketplace_product = test_marketplace_product_with_stock['marketplace_product']
stock_info = self.service.get_stock_info(db, product.gtin)
stock_info = self.service.get_stock_info(db, marketplace_product.gtin)
assert stock_info is not None
assert stock_info.gtin == product.gtin
assert stock_info.gtin == marketplace_product.gtin
assert stock_info.total_quantity > 0
assert len(stock_info.locations) > 0
def test_get_stock_info_no_stock(self, db, test_product):
def test_get_stock_info_no_stock(self, db, test_marketplace_product):
"""Test getting stock info for product without stock"""
stock_info = self.service.get_stock_info(db, test_product.gtin or "1234567890123")
stock_info = self.service.get_stock_info(db, test_marketplace_product.gtin or "1234567890123")
assert stock_info is None
def test_product_exists_true(self, db, test_product):
def test_product_exists_true(self, db, test_marketplace_product):
"""Test product_exists returns True for existing product"""
exists = self.service.product_exists(db, test_product.product_id)
exists = self.service.product_exists(db, test_marketplace_product.marketplace_product_id)
assert exists is True
def test_product_exists_false(self, db):
@@ -257,7 +257,7 @@ class TestProductService:
exists = self.service.product_exists(db, "NONEXISTENT")
assert exists is False
def test_generate_csv_export_success(self, db, test_product):
def test_generate_csv_export_success(self, db, test_marketplace_product):
"""Test CSV export generation"""
csv_generator = self.service.generate_csv_export(db)
@@ -265,17 +265,17 @@ class TestProductService:
csv_lines = list(csv_generator)
assert len(csv_lines) > 1 # Header + at least one data row
assert csv_lines[0].startswith("product_id,title,description") # Check header
assert csv_lines[0].startswith("marketplace_product_id,title,description") # Check header
# Check that test product appears in CSV
csv_content = "".join(csv_lines)
assert test_product.product_id in csv_content
assert test_marketplace_product.marketplace_product_id in csv_content
def test_generate_csv_export_with_filters(self, db, test_product):
def test_generate_csv_export_with_filters(self, db, test_marketplace_product):
"""Test CSV export with marketplace filter"""
csv_generator = self.service.generate_csv_export(
db,
marketplace=test_product.marketplace
marketplace=test_marketplace_product.marketplace
)
csv_lines = list(csv_generator)
@@ -283,4 +283,4 @@ class TestProductService:
if len(csv_lines) > 1: # If there's data
csv_content = "".join(csv_lines)
assert test_product.marketplace in csv_content
assert test_marketplace_product.marketplace in csv_content

22
tests/unit/services/test_shop_service.py
View File

@@ -7,7 +7,7 @@ from app.exceptions import (
ShopAlreadyExistsException,
UnauthorizedShopAccessException,
InvalidShopDataException,
ProductNotFoundException,
MarketplaceProductNotFoundException,
ShopProductAlreadyExistsException,
MaxShopsReachedException,
ValidationException,
@@ -179,7 +179,7 @@ class TestShopService:
def test_add_product_to_shop_success(self, db, test_shop, unique_product):
"""Test successfully adding product to shop"""
shop_product_data = ShopProductCreate(
product_id=unique_product.product_id,
marketplace_product_id=unique_product.marketplace_product_id,
price="15.99",
is_featured=True,
stock_quantity=5,
@@ -191,25 +191,25 @@ class TestShopService:
assert shop_product is not None
assert shop_product.shop_id == test_shop.id
assert shop_product.product_id == unique_product.id
assert shop_product.marketplace_product_id == unique_product.id
def test_add_product_to_shop_product_not_found(self, db, test_shop):
"""Test adding non-existent product to shop fails"""
shop_product_data = ShopProductCreate(product_id="NONEXISTENT", price="15.99")
shop_product_data = ShopProductCreate(marketplace_product_id="NONEXISTENT", price="15.99")
with pytest.raises(ProductNotFoundException) as exc_info:
with pytest.raises(MarketplaceProductNotFoundException) as exc_info:
self.service.add_product_to_shop(db, test_shop, shop_product_data)
exception = exc_info.value
assert exception.status_code == 404
assert exception.error_code == "PRODUCT_NOT_FOUND"
assert exception.details["resource_type"] == "Product"
assert exception.details["resource_type"] == "MarketplaceProduct"
assert exception.details["identifier"] == "NONEXISTENT"
def test_add_product_to_shop_already_exists(self, db, test_shop, shop_product):
"""Test adding product that's already in shop fails"""
shop_product_data = ShopProductCreate(
product_id=shop_product.product.product_id, price="15.99"
marketplace_product_id=shop_product.product.marketplace_product_id, price="15.99"
)
with pytest.raises(ShopProductAlreadyExistsException) as exc_info:
@@ -219,7 +219,7 @@ class TestShopService:
assert exception.status_code == 409
assert exception.error_code == "SHOP_PRODUCT_ALREADY_EXISTS"
assert exception.details["shop_code"] == test_shop.shop_code
assert exception.details["product_id"] == shop_product.product.product_id
assert exception.details["marketplace_product_id"] == shop_product.product.marketplace_product_id
def test_get_shop_products_owner_access(
self, db, test_user, test_shop, shop_product
@@ -229,8 +229,8 @@ class TestShopService:
assert total >= 1
assert len(products) >= 1
product_ids = [p.product_id for p in products]
assert shop_product.product_id in product_ids
product_ids = [p.marketplace_product_id for p in products]
assert shop_product.marketplace_product_id in product_ids
def test_get_shop_products_access_denied(self, db, test_user, inactive_shop):
"""Test non-owner cannot access unverified shop products"""
@@ -300,7 +300,7 @@ class TestShopService:
monkeypatch.setattr(db, "commit", mock_commit)
shop_product_data = ShopProductCreate(
product_id=unique_product.product_id, price="15.99"
marketplace_product_id=unique_product.marketplace_product_id, price="15.99"
)
with pytest.raises(ValidationException) as exc_info:

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

@@ -2,7 +2,7 @@
import pytest
from app.services.stats_service import StatsService
from models.database.product import Product
from models.database.marketplace_product import MarketplaceProduct
from models.database.stock import Stock
@@ -15,7 +15,7 @@ class TestStatsService:
"""Setup method following the same pattern as other service tests"""
self.service = StatsService()
def test_get_comprehensive_stats_basic(self, db, test_product, test_stock):
def test_get_comprehensive_stats_basic(self, db, test_marketplace_product, test_stock):
"""Test getting comprehensive stats with basic data"""
stats = self.service.get_comprehensive_stats(db)
@@ -31,13 +31,13 @@ class TestStatsService:
assert stats["total_stock_entries"] >= 1
assert stats["total_inventory_quantity"] >= 10 # test_stock has quantity 10
def test_get_comprehensive_stats_multiple_products(self, db, test_product):
def test_get_comprehensive_stats_multiple_products(self, db, test_marketplace_product):
"""Test comprehensive stats with multiple products across different dimensions"""
# Create products with different brands, categories, marketplaces
additional_products = [
Product(
product_id="PROD002",
title="Product 2",
MarketplaceProduct(
marketplace_product_id="PROD002",
title="MarketplaceProduct 2",
brand="DifferentBrand",
google_product_category="Different Category",
marketplace="Amazon",
@@ -45,9 +45,9 @@ class TestStatsService:
price="15.99",
currency="EUR",
),
Product(
product_id="PROD003",
title="Product 3",
MarketplaceProduct(
marketplace_product_id="PROD003",
title="MarketplaceProduct 3",
brand="ThirdBrand",
google_product_category="Third Category",
marketplace="eBay",
@@ -55,12 +55,12 @@ class TestStatsService:
price="25.99",
currency="USD",
),
Product(
product_id="PROD004",
title="Product 4",
brand="TestBrand", # Same as test_product
MarketplaceProduct(
marketplace_product_id="PROD004",
title="MarketplaceProduct 4",
brand="TestBrand", # Same as test_marketplace_product
google_product_category="Different Category",
marketplace="Letzshop", # Same as test_product
marketplace="Letzshop", # Same as test_marketplace_product
shop_name="DifferentShop",
price="35.99",
currency="EUR",
@@ -71,7 +71,7 @@ class TestStatsService:
stats = self.service.get_comprehensive_stats(db)
assert stats["total_products"] >= 4 # test_product + 3 additional
assert stats["total_products"] >= 4 # test_marketplace_product + 3 additional
assert stats["unique_brands"] >= 3 # TestBrand, DifferentBrand, ThirdBrand
assert stats["unique_categories"] >= 2 # At least 2 different categories
assert stats["unique_marketplaces"] >= 3 # Letzshop, Amazon, eBay
@@ -81,9 +81,9 @@ class TestStatsService:
"""Test comprehensive stats handles null/empty values correctly"""
# Create products with null/empty values
products_with_nulls = [
Product(
product_id="NULL001",
title="Product with Nulls",
MarketplaceProduct(
marketplace_product_id="NULL001",
title="MarketplaceProduct with Nulls",
brand=None, # Null brand
google_product_category=None, # Null category
marketplace=None, # Null marketplace
@@ -91,9 +91,9 @@ class TestStatsService:
price="10.00",
currency="EUR",
),
Product(
product_id="EMPTY001",
title="Product with Empty Values",
MarketplaceProduct(
marketplace_product_id="EMPTY001",
title="MarketplaceProduct with Empty Values",
brand="", # Empty brand
google_product_category="", # Empty category
marketplace="", # Empty marketplace
@@ -115,7 +115,7 @@ class TestStatsService:
assert isinstance(stats["unique_marketplaces"], int)
assert isinstance(stats["unique_shops"], int)
def test_get_marketplace_breakdown_stats_basic(self, db, test_product):
def test_get_marketplace_breakdown_stats_basic(self, db, test_marketplace_product):
"""Test getting marketplace breakdown stats with basic data"""
stats = self.service.get_marketplace_breakdown_stats(db)
@@ -124,7 +124,7 @@ class TestStatsService:
# Find our test marketplace in the results
test_marketplace_stat = next(
(stat for stat in stats if stat["marketplace"] == test_product.marketplace),
(stat for stat in stats if stat["marketplace"] == test_marketplace_product.marketplace),
None,
)
assert test_marketplace_stat is not None
@@ -133,32 +133,32 @@ class TestStatsService:
assert test_marketplace_stat["unique_brands"] >= 1
def test_get_marketplace_breakdown_stats_multiple_marketplaces(
self, db, test_product
self, db, test_marketplace_product
):
"""Test marketplace breakdown with multiple marketplaces"""
# Create products for different marketplaces
marketplace_products = [
Product(
product_id="AMAZON001",
title="Amazon Product 1",
MarketplaceProduct(
marketplace_product_id="AMAZON001",
title="Amazon MarketplaceProduct 1",
brand="AmazonBrand1",
marketplace="Amazon",
shop_name="AmazonShop1",
price="20.00",
currency="EUR",
),
Product(
product_id="AMAZON002",
title="Amazon Product 2",
MarketplaceProduct(
marketplace_product_id="AMAZON002",
title="Amazon MarketplaceProduct 2",
brand="AmazonBrand2",
marketplace="Amazon",
shop_name="AmazonShop2",
price="25.00",
currency="EUR",
),
Product(
product_id="EBAY001",
title="eBay Product",
MarketplaceProduct(
marketplace_product_id="EBAY001",
title="eBay MarketplaceProduct",
brand="eBayBrand",
marketplace="eBay",
shop_name="eBayShop",
@@ -171,11 +171,11 @@ class TestStatsService:
stats = self.service.get_marketplace_breakdown_stats(db)
# Should have at least 3 marketplaces: test_product.marketplace, Amazon, eBay
# Should have at least 3 marketplaces: test_marketplace_product.marketplace, Amazon, eBay
marketplace_names = [stat["marketplace"] for stat in stats]
assert "Amazon" in marketplace_names
assert "eBay" in marketplace_names
assert test_product.marketplace in marketplace_names
assert test_marketplace_product.marketplace in marketplace_names
# Check Amazon stats specifically
amazon_stat = next(stat for stat in stats if stat["marketplace"] == "Amazon")
@@ -192,9 +192,9 @@ class TestStatsService:
def test_get_marketplace_breakdown_stats_excludes_nulls(self, db):
"""Test marketplace breakdown excludes products with null marketplaces"""
# Create product with null marketplace
null_marketplace_product = Product(
product_id="NULLMARKET001",
title="Product without marketplace",
null_marketplace_product = MarketplaceProduct(
marketplace_product_id="NULLMARKET001",
title="MarketplaceProduct without marketplace",
marketplace=None,
shop_name="SomeShop",
brand="SomeBrand",
@@ -212,29 +212,29 @@ class TestStatsService:
]
assert None not in marketplace_names
def test_get_product_count(self, db, test_product):
def test_get_product_count(self, db, test_marketplace_product):
"""Test getting total product count"""
count = self.service._get_product_count(db)
assert count >= 1
assert isinstance(count, int)
def test_get_unique_brands_count(self, db, test_product):
def test_get_unique_brands_count(self, db, test_marketplace_product):
"""Test getting unique brands count"""
# Add products with different brands
brand_products = [
Product(
product_id="BRAND001",
title="Brand Product 1",
MarketplaceProduct(
marketplace_product_id="BRAND001",
title="Brand MarketplaceProduct 1",
brand="BrandA",
marketplace="Test",
shop_name="TestShop",
price="10.00",
currency="EUR",
),
Product(
product_id="BRAND002",
title="Brand Product 2",
MarketplaceProduct(
marketplace_product_id="BRAND002",
title="Brand MarketplaceProduct 2",
brand="BrandB",
marketplace="Test",
shop_name="TestShop",
@@ -249,25 +249,25 @@ class TestStatsService:
assert (
count >= 2
) # At least BrandA and BrandB, plus possibly test_product brand
) # At least BrandA and BrandB, plus possibly test_marketplace_product brand
assert isinstance(count, int)
def test_get_unique_categories_count(self, db, test_product):
def test_get_unique_categories_count(self, db, test_marketplace_product):
"""Test getting unique categories count"""
# Add products with different categories
category_products = [
Product(
product_id="CAT001",
title="Category Product 1",
MarketplaceProduct(
marketplace_product_id="CAT001",
title="Category MarketplaceProduct 1",
google_product_category="Electronics",
marketplace="Test",
shop_name="TestShop",
price="10.00",
currency="EUR",
),
Product(
product_id="CAT002",
title="Category Product 2",
MarketplaceProduct(
marketplace_product_id="CAT002",
title="Category MarketplaceProduct 2",
google_product_category="Books",
marketplace="Test",
shop_name="TestShop",
@@ -283,21 +283,21 @@ class TestStatsService:
assert count >= 2 # At least Electronics and Books
assert isinstance(count, int)
def test_get_unique_marketplaces_count(self, db, test_product):
def test_get_unique_marketplaces_count(self, db, test_marketplace_product):
"""Test getting unique marketplaces count"""
# Add products with different marketplaces
marketplace_products = [
Product(
product_id="MARKET001",
title="Marketplace Product 1",
MarketplaceProduct(
marketplace_product_id="MARKET001",
title="Marketplace MarketplaceProduct 1",
marketplace="Amazon",
shop_name="AmazonShop",
price="10.00",
currency="EUR",
),
Product(
product_id="MARKET002",
title="Marketplace Product 2",
MarketplaceProduct(
marketplace_product_id="MARKET002",
title="Marketplace MarketplaceProduct 2",
marketplace="eBay",
shop_name="eBayShop",
price="15.00",
@@ -309,24 +309,24 @@ class TestStatsService:
count = self.service._get_unique_marketplaces_count(db)
assert count >= 2 # At least Amazon and eBay, plus test_product marketplace
assert count >= 2 # At least Amazon and eBay, plus test_marketplace_product marketplace
assert isinstance(count, int)
def test_get_unique_shops_count(self, db, test_product):
def test_get_unique_shops_count(self, db, test_marketplace_product):
"""Test getting unique shops count"""
# Add products with different shop names
shop_products = [
Product(
product_id="SHOP001",
title="Shop Product 1",
MarketplaceProduct(
marketplace_product_id="SHOP001",
title="Shop MarketplaceProduct 1",
marketplace="Test",
shop_name="ShopA",
price="10.00",
currency="EUR",
),
Product(
product_id="SHOP002",
title="Shop Product 2",
MarketplaceProduct(
marketplace_product_id="SHOP002",
title="Shop MarketplaceProduct 2",
marketplace="Test",
shop_name="ShopB",
price="15.00",
@@ -338,7 +338,7 @@ class TestStatsService:
count = self.service._get_unique_shops_count(db)
assert count >= 2 # At least ShopA and ShopB, plus test_product shop
assert count >= 2 # At least ShopA and ShopB, plus test_marketplace_product shop
assert isinstance(count, int)
def test_get_stock_statistics(self, db, test_stock):
@@ -374,27 +374,27 @@ class TestStatsService:
"""Test getting brands for a specific marketplace"""
# Create products for specific marketplace
marketplace_products = [
Product(
product_id="SPECIFIC001",
title="Specific Product 1",
MarketplaceProduct(
marketplace_product_id="SPECIFIC001",
title="Specific MarketplaceProduct 1",
brand="SpecificBrand1",
marketplace="SpecificMarket",
shop_name="SpecificShop1",
price="10.00",
currency="EUR",
),
Product(
product_id="SPECIFIC002",
title="Specific Product 2",
MarketplaceProduct(
marketplace_product_id="SPECIFIC002",
title="Specific MarketplaceProduct 2",
brand="SpecificBrand2",
marketplace="SpecificMarket",
shop_name="SpecificShop2",
price="15.00",
currency="EUR",
),
Product(
product_id="OTHER001",
title="Other Product",
MarketplaceProduct(
marketplace_product_id="OTHER001",
title="Other MarketplaceProduct",
brand="OtherBrand",
marketplace="OtherMarket",
shop_name="OtherShop",
@@ -416,18 +416,18 @@ class TestStatsService:
"""Test getting shops for a specific marketplace"""
# Create products for specific marketplace
marketplace_products = [
Product(
product_id="SHOPTEST001",
title="Shop Test Product 1",
MarketplaceProduct(
marketplace_product_id="SHOPTEST001",
title="Shop Test MarketplaceProduct 1",
brand="TestBrand",
marketplace="TestMarketplace",
shop_name="TestShop1",
price="10.00",
currency="EUR",
),
Product(
product_id="SHOPTEST002",
title="Shop Test Product 2",
MarketplaceProduct(
marketplace_product_id="SHOPTEST002",
title="Shop Test MarketplaceProduct 2",
brand="TestBrand",
marketplace="TestMarketplace",
shop_name="TestShop2",
@@ -448,25 +448,25 @@ class TestStatsService:
"""Test getting product count for a specific marketplace"""
# Create products for specific marketplace
marketplace_products = [
Product(
product_id="COUNT001",
title="Count Product 1",
MarketplaceProduct(
marketplace_product_id="COUNT001",
title="Count MarketplaceProduct 1",
marketplace="CountMarketplace",
shop_name="CountShop",
price="10.00",
currency="EUR",
),
Product(
product_id="COUNT002",
title="Count Product 2",
MarketplaceProduct(
marketplace_product_id="COUNT002",
title="Count MarketplaceProduct 2",
marketplace="CountMarketplace",
shop_name="CountShop",
price="15.00",
currency="EUR",
),
Product(
product_id="COUNT003",
title="Count Product 3",
MarketplaceProduct(
marketplace_product_id="COUNT003",
title="Count MarketplaceProduct 3",
marketplace="CountMarketplace",
shop_name="CountShop",
price="20.00",

24
tests/unit/services/test_stock_service.py
View File

@@ -14,7 +14,7 @@ from app.exceptions import (
ValidationException,
)
from models.schemas.stock import StockAdd, StockCreate, StockUpdate
from models.database.product import Product
from models.database.marketplace_product import MarketplaceProduct
from models.database.stock import Stock
@@ -254,7 +254,7 @@ class TestStockService:
# The service prevents negative stock through InsufficientStockException
assert exc_info.value.error_code == "INSUFFICIENT_STOCK"
def test_get_stock_by_gtin_success(self, db, test_stock, test_product):
def test_get_stock_by_gtin_success(self, db, test_stock, test_marketplace_product):
"""Test getting stock summary by GTIN successfully."""
result = self.service.get_stock_by_gtin(db, test_stock.gtin)
@@ -263,11 +263,11 @@ class TestStockService:
assert len(result.locations) == 1
assert result.locations[0].location == test_stock.location
assert result.locations[0].quantity == test_stock.quantity
assert result.product_title == test_product.title
assert result.product_title == test_marketplace_product.title
def test_get_stock_by_gtin_multiple_locations_success(self, db, test_product):
def test_get_stock_by_gtin_multiple_locations_success(self, db, test_marketplace_product):
"""Test getting stock summary with multiple locations successfully."""
unique_gtin = test_product.gtin
unique_gtin = test_marketplace_product.gtin
unique_id = str(uuid.uuid4())[:8]
# Create multiple stock entries for the same GTIN with unique locations
@@ -301,13 +301,13 @@ class TestStockService:
assert exc_info.value.error_code == "STOCK_VALIDATION_FAILED"
assert "Invalid GTIN format" in str(exc_info.value)
def test_get_total_stock_success(self, db, test_stock, test_product):
def test_get_total_stock_success(self, db, test_stock, test_marketplace_product):
"""Test getting total stock for a GTIN successfully."""
result = self.service.get_total_stock(db, test_stock.gtin)
assert result["gtin"] == test_stock.gtin
assert result["total_quantity"] == test_stock.quantity
assert result["product_title"] == test_product.title
assert result["product_title"] == test_marketplace_product.title
assert result["locations_count"] == 1
def test_get_total_stock_invalid_gtin_validation_error(self, db):
@@ -415,7 +415,7 @@ class TestStockService:
assert exc_info.value.error_code == "STOCK_NOT_FOUND"
assert "99999" in str(exc_info.value)
def test_get_low_stock_items_success(self, db, test_stock, test_product):
def test_get_low_stock_items_success(self, db, test_stock, test_marketplace_product):
"""Test getting low stock items successfully."""
# Set stock to a low value
test_stock.quantity = 5
@@ -428,7 +428,7 @@ class TestStockService:
assert low_stock_item is not None
assert low_stock_item["current_quantity"] == 5
assert low_stock_item["location"] == test_stock.location
assert low_stock_item["product_title"] == test_product.title
assert low_stock_item["product_title"] == test_marketplace_product.title
def test_get_low_stock_items_invalid_threshold_error(self, db):
"""Test getting low stock items with invalid threshold returns InvalidQuantityException."""
@@ -491,9 +491,9 @@ class TestStockService:
@pytest.fixture
def test_product_with_stock(db, test_stock):
"""Create a test product that corresponds to the test stock."""
product = Product(
product_id="STOCK_TEST_001",
title="Stock Test Product",
product = MarketplaceProduct(
marketplace_product_id="STOCK_TEST_001",
title="Stock Test MarketplaceProduct",
gtin=test_stock.gtin,
price="29.99",
brand="TestBrand",

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

@@ -18,7 +18,7 @@ class TestCSVProcessor:
def test_download_csv_encoding_fallback(self, mock_get):
"""Test CSV download with encoding fallback"""
# Create content with special characters that would fail UTF-8 if not properly encoded
special_content = "product_id,title,price\nTEST001,Café Product,10.99"
special_content = "marketplace_product_id,title,price\nTEST001,Café MarketplaceProduct,10.99"
mock_response = Mock()
mock_response.status_code = 200
@@ -31,7 +31,7 @@ class TestCSVProcessor:
mock_get.assert_called_once_with("http://example.com/test.csv", timeout=30)
assert isinstance(csv_content, str)
assert "Café Product" in csv_content
assert "Café MarketplaceProduct" in csv_content
@patch("requests.get")
def test_download_csv_encoding_ignore_fallback(self, mock_get):
@@ -41,7 +41,7 @@ class TestCSVProcessor:
mock_response.status_code = 200
# Create bytes that will fail most encodings
mock_response.content = (
b"product_id,title,price\nTEST001,\xff\xfe Product,10.99"
b"marketplace_product_id,title,price\nTEST001,\xff\xfe MarketplaceProduct,10.99"
)
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
@@ -51,7 +51,7 @@ class TestCSVProcessor:
mock_get.assert_called_once_with("http://example.com/test.csv", timeout=30)
assert isinstance(csv_content, str)
# Should still contain basic content even with ignored errors
assert "product_id,title,price" in csv_content
assert "marketplace_product_id,title,price" in csv_content
assert "TEST001" in csv_content
@patch("requests.get")
@@ -91,15 +91,15 @@ class TestCSVProcessor:
def test_parse_csv_content(self):
"""Test CSV content parsing"""
csv_content = """product_id,title,price,marketplace
TEST001,Test Product 1,10.99,TestMarket
TEST002,Test Product 2,15.99,TestMarket"""
csv_content = """marketplace_product_id,title,price,marketplace
TEST001,Test MarketplaceProduct 1,10.99,TestMarket
TEST002,Test MarketplaceProduct 2,15.99,TestMarket"""
df = self.processor.parse_csv(csv_content)
assert len(df) == 2
assert "product_id" in df.columns
assert df.iloc[0]["product_id"] == "TEST001"
assert "marketplace_product_id" in df.columns
assert df.iloc[0]["marketplace_product_id"] == "TEST001"
assert df.iloc[1]["price"] == 15.99
@pytest.mark.asyncio
@@ -112,8 +112,8 @@ TEST002,Test Product 2,15.99,TestMarket"""
mock_download.return_value = "csv_content"
mock_df = pd.DataFrame(
{
"product_id": ["TEST001", "TEST002"],
"title": ["Product 1", "Product 2"],
"marketplace_product_id": ["TEST001", "TEST002"],
"title": ["MarketplaceProduct 1", "MarketplaceProduct 2"],
"price": ["10.99", "15.99"],
"marketplace": ["TestMarket", "TestMarket"],
"shop_name": ["TestShop", "TestShop"],