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

shop product refactoring

Browse Source
This commit is contained in:
Samir Boulahtit
2025-10-04 23:38:53 +02:00
parent 4d2866af5e
commit 0114b6c46e
68 changed files with 2234 additions and 2236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,12 +10,12 @@ from app.exceptions.marketplace_import_job import (
ImportJobCannotBeCancelledException,
ImportJobCannotBeDeletedException,
)
from app.exceptions.shop import ShopNotFoundException, UnauthorizedShopAccessException
from app.exceptions.vendor import VendorNotFoundException, UnauthorizedVendorAccessException
from app.exceptions.base import ValidationException
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.vendor import Vendor
from models.database.user import User
@@ -25,107 +25,107 @@ class TestMarketplaceService:
def setup_method(self):
self.service = MarketplaceImportJobService()
def test_validate_shop_access_success(self, db, test_shop, test_user):
"""Test successful shop access validation"""
# Set the shop owner to the test user
test_shop.owner_id = test_user.id
def test_validate_vendor_access_success(self, db, test_vendor, test_user):
"""Test successful vendor access validation"""
# Set the vendor owner to the test user
test_vendor.owner_id = test_user.id
db.commit()
result = self.service.validate_shop_access(db, test_shop.shop_code, test_user)
result = self.service.validate_vendor_access(db, test_vendor.vendor_code, test_user)
assert result.shop_code == test_shop.shop_code
assert result.vendor_code == test_vendor.vendor_code
assert result.owner_id == test_user.id
def test_validate_shop_access_admin_can_access_any_shop(
self, db, test_shop, test_admin
def test_validate_vendor_access_admin_can_access_any_vendor(
self, db, test_vendor, test_admin
):
"""Test that admin users can access any shop"""
result = self.service.validate_shop_access(db, test_shop.shop_code, test_admin)
"""Test that admin users can access any vendor """
result = self.service.validate_vendor_access(db, test_vendor.vendor_code, test_admin)
assert result.shop_code == test_shop.shop_code
assert result.vendor_code == test_vendor.vendor_code
def test_validate_shop_access_shop_not_found(self, db, test_user):
"""Test shop access validation when shop doesn't exist"""
with pytest.raises(ShopNotFoundException) as exc_info:
self.service.validate_shop_access(db, "NONEXISTENT", test_user)
def test_validate_vendor_access_vendor_not_found(self, db, test_user):
"""Test vendor access validation when vendor doesn't exist"""
with pytest.raises(VendorNotFoundException) as exc_info:
self.service.validate_vendor_access(db, "NONEXISTENT", test_user)
exception = exc_info.value
assert exception.error_code == "SHOP_NOT_FOUND"
assert exception.error_code == "VENDOR_NOT_FOUND"
assert exception.status_code == 404
assert "NONEXISTENT" in exception.message
def test_validate_shop_access_permission_denied(
self, db, test_shop, test_user, other_user
def test_validate_vendor_access_permission_denied(
self, db, test_vendor, test_user, other_user
):
"""Test shop access validation when user doesn't own the shop"""
# Set the shop owner to a different user
test_shop.owner_id = other_user.id
"""Test vendor access validation when user doesn't own the vendor """
# Set the vendor owner to a different user
test_vendor.owner_id = other_user.id
db.commit()
with pytest.raises(UnauthorizedShopAccessException) as exc_info:
self.service.validate_shop_access(db, test_shop.shop_code, test_user)
with pytest.raises(UnauthorizedVendorAccessException) as exc_info:
self.service.validate_vendor_access(db, test_vendor.vendor_code, test_user)
exception = exc_info.value
assert exception.error_code == "UNAUTHORIZED_SHOP_ACCESS"
assert exception.error_code == "UNAUTHORIZED_VENDOR_ACCESS"
assert exception.status_code == 403
assert test_shop.shop_code in exception.message
assert test_vendor.vendor_code in exception.message
def test_create_import_job_success(self, db, test_shop, test_user):
def test_create_import_job_success(self, db, test_vendor, test_user):
"""Test successful creation of import job"""
# Set the shop owner to the test user
test_shop.owner_id = test_user.id
# Set the vendor owner to the test user
test_vendor.owner_id = test_user.id
db.commit()
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code=test_shop.shop_code,
vendor_code=test_vendor.vendor_code,
batch_size=1000,
)
result = self.service.create_import_job(db, request, test_user)
assert result.marketplace == "Amazon"
assert result.shop_id == test_shop.id
assert result.vendor_id == test_vendor.id
assert result.user_id == test_user.id
assert result.status == "pending"
assert result.source_url == "https://example.com/products.csv"
assert result.shop_name == test_shop.shop_name
assert result.vendor_name == test_vendor.vendor_name
def test_create_import_job_invalid_shop(self, db, test_user):
"""Test import job creation with invalid shop"""
def test_create_import_job_invalid_vendor(self, db, test_user):
"""Test import job creation with invalid vendor """
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code="INVALID_SHOP",
vendor_code="INVALID_SHOP",
batch_size=1000,
)
with pytest.raises(ShopNotFoundException) as exc_info:
with pytest.raises(VendorNotFoundException) as exc_info:
self.service.create_import_job(db, request, test_user)
exception = exc_info.value
assert exception.error_code == "SHOP_NOT_FOUND"
assert exception.error_code == "VENDOR_NOT_FOUND"
assert "INVALID_SHOP" in exception.message
def test_create_import_job_unauthorized_access(self, db, test_shop, test_user, other_user):
"""Test import job creation with unauthorized shop access"""
# Set the shop owner to a different user
test_shop.owner_id = other_user.id
def test_create_import_job_unauthorized_access(self, db, test_vendor, test_user, other_user):
"""Test import job creation with unauthorized vendor access"""
# Set the vendor owner to a different user
test_vendor.owner_id = other_user.id
db.commit()
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code=test_shop.shop_code,
vendor_code=test_vendor.vendor_code,
batch_size=1000,
)
with pytest.raises(UnauthorizedShopAccessException) as exc_info:
with pytest.raises(UnauthorizedVendorAccessException) as exc_info:
self.service.create_import_job(db, request, test_user)
exception = exc_info.value
assert exception.error_code == "UNAUTHORIZED_SHOP_ACCESS"
assert exception.error_code == "UNAUTHORIZED_VENDOR_ACCESS"
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"""
@@ -194,7 +194,7 @@ class TestMarketplaceService:
assert len(jobs) >= 1
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):
def test_get_import_jobs_with_pagination(self, db, test_user, test_vendor):
"""Test getting import jobs with pagination"""
unique_id = str(uuid.uuid4())[:8]
@@ -203,9 +203,9 @@ class TestMarketplaceService:
job = MarketplaceImportJob(
status="completed",
marketplace=f"Marketplace_{unique_id}_{i}",
shop_name=f"Test_Shop_{unique_id}_{i}",
vendor_name=f"Test_vendor_{unique_id}_{i}",
user_id=test_user.id,
shop_id=test_shop.id,
vendor_id=test_vendor.id,
source_url=f"https://test-{i}.example.com/import",
imported_count=0,
updated_count=0,
@@ -296,7 +296,7 @@ class TestMarketplaceService:
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):
def test_cancel_import_job_success(self, db, test_user, test_vendor):
"""Test cancelling a pending import job"""
unique_id = str(uuid.uuid4())[:8]
@@ -304,9 +304,9 @@ class TestMarketplaceService:
job = MarketplaceImportJob(
status="pending",
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
vendor_name=f"TEST_VENDOR_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id,
vendor_id=test_vendor.id,
source_url="https://test.example.com/import",
imported_count=0,
updated_count=0,
@@ -354,7 +354,7 @@ class TestMarketplaceService:
assert exception.status_code == 400
assert "completed" in exception.message
def test_delete_import_job_success(self, db, test_user, test_shop):
def test_delete_import_job_success(self, db, test_user, test_vendor):
"""Test deleting a completed import job"""
unique_id = str(uuid.uuid4())[:8]
@@ -362,9 +362,9 @@ class TestMarketplaceService:
job = MarketplaceImportJob(
status="completed",
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
vendor_name=f"TEST_VENDOR_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id,
vendor_id=test_vendor.id,
source_url="https://test.example.com/import",
imported_count=0,
updated_count=0,
@@ -404,7 +404,7 @@ class TestMarketplaceService:
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_OWNED"
def test_delete_import_job_invalid_status(self, db, test_user, test_shop):
def test_delete_import_job_invalid_status(self, db, test_user, test_vendor):
"""Test deleting a job that can't be deleted"""
unique_id = str(uuid.uuid4())[:8]
@@ -412,9 +412,9 @@ class TestMarketplaceService:
job = MarketplaceImportJob(
status="pending",
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
vendor_name=f"TEST_VENDOR_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id,
vendor_id=test_vendor.id,
source_url="https://test.example.com/import",
imported_count=0,
updated_count=0,
@@ -434,25 +434,25 @@ class TestMarketplaceService:
assert "pending" in exception.message
# Test edge cases and error scenarios
def test_validate_shop_access_case_insensitive(self, db, test_shop, test_user):
"""Test shop access validation is case insensitive"""
test_shop.owner_id = test_user.id
def test_validate_vendor_access_case_insensitive(self, db, test_vendor, test_user):
"""Test vendor access validation is case insensitive"""
test_vendor.owner_id = test_user.id
db.commit()
# Test with lowercase shop code
result = self.service.validate_shop_access(db, test_shop.shop_code.lower(), test_user)
assert result.shop_code == test_shop.shop_code
# Test with lowercase vendor code
result = self.service.validate_vendor_access(db, test_vendor.vendor_code.lower(), test_user)
assert result.vendor_code == test_vendor.vendor_code
# Test with uppercase shop code
result = self.service.validate_shop_access(db, test_shop.shop_code.upper(), test_user)
assert result.shop_code == test_shop.shop_code
# Test with uppercase vendor code
result = self.service.validate_vendor_access(db, test_vendor.vendor_code.upper(), test_user)
assert result.vendor_code == test_vendor.vendor_code
def test_create_import_job_database_error(self, db_with_error, test_user):
"""Test import job creation handles database errors"""
request = MarketplaceImportJobRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code="TEST_SHOP",
vendor_code="TEST_SHOP",
batch_size=1000,
)