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

Marketplace tests update

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-24 22:28:44 +02:00
parent f9879126c8
commit cea88a46c5
16 changed files with 613 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,6 +4,14 @@ from datetime import datetime
import pytest
from app.exceptions.marketplace import (
ImportJobNotFoundException,
ImportJobNotOwnedException,
ImportJobCannotBeCancelledException,
ImportJobCannotBeDeletedException,
)
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
@@ -38,9 +46,14 @@ class TestMarketplaceService:
def test_validate_shop_access_shop_not_found(self, db, test_user):
"""Test shop access validation when shop doesn't exist"""
with pytest.raises(ValueError, match="Shop not found"):
with pytest.raises(ShopNotFoundException) as exc_info:
self.service.validate_shop_access(db, "NONEXISTENT", test_user)
exception = exc_info.value
assert exception.error_code == "SHOP_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
):
@@ -49,9 +62,14 @@ class TestMarketplaceService:
test_shop.owner_id = other_user.id
db.commit()
with pytest.raises(PermissionError, match="Access denied to this shop"):
with pytest.raises(UnauthorizedShopAccessException) as exc_info:
self.service.validate_shop_access(db, test_shop.shop_code, test_user)
exception = exc_info.value
assert exception.error_code == "UNAUTHORIZED_SHOP_ACCESS"
assert exception.status_code == 403
assert test_shop.shop_code in exception.message
def test_create_import_job_success(self, db, test_shop, test_user):
"""Test successful creation of import job"""
# Set the shop owner to the test user
@@ -68,11 +86,11 @@ class TestMarketplaceService:
result = self.service.create_import_job(db, request, test_user)
assert result.marketplace == "Amazon"
# Check the correct field based on your model
assert result.shop_id == test_shop.id # Changed from shop_code to shop_id
assert result.user_id == test_user.id if hasattr(result, "user_id") else True
assert result.shop_id == test_shop.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
def test_create_import_job_invalid_shop(self, db, test_user):
"""Test import job creation with invalid shop"""
@@ -83,9 +101,32 @@ class TestMarketplaceService:
batch_size=1000,
)
with pytest.raises(ValueError, match="Shop not found"):
with pytest.raises(ShopNotFoundException) as exc_info:
self.service.create_import_job(db, request, test_user)
exception = exc_info.value
assert exception.error_code == "SHOP_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
db.commit()
request = MarketplaceImportRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code=test_shop.shop_code,
batch_size=1000,
)
with pytest.raises(UnauthorizedShopAccessException) as exc_info:
self.service.create_import_job(db, request, test_user)
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):
"""Test getting import job by ID for job owner"""
result = self.service.get_import_job_by_id(
@@ -93,9 +134,7 @@ class TestMarketplaceService:
)
assert result.id == test_marketplace_job.id
# Check user_id if the field exists
if hasattr(result, "user_id"):
assert result.user_id == test_user.id
assert result.user_id == test_user.id
def test_get_import_job_by_id_admin_access(
self, db, test_marketplace_job, test_admin
@@ -109,25 +148,33 @@ class TestMarketplaceService:
def test_get_import_job_by_id_not_found(self, db, test_user):
"""Test getting non-existent import job"""
with pytest.raises(ValueError, match="Marketplace import job not found"):
with pytest.raises(ImportJobNotFoundException) as exc_info:
self.service.get_import_job_by_id(db, 99999, test_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_FOUND"
assert exception.status_code == 404
assert "99999" in exception.message
def test_get_import_job_by_id_access_denied(
self, db, test_marketplace_job, other_user
):
"""Test access denied when user doesn't own the job"""
with pytest.raises(PermissionError, match="Access denied to this import job"):
with pytest.raises(ImportJobNotOwnedException) as exc_info:
self.service.get_import_job_by_id(db, test_marketplace_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
def test_get_import_jobs_user_filter(self, db, test_marketplace_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)
# Check user_id if the field exists
if hasattr(test_marketplace_job, "user_id"):
assert test_marketplace_job.user_id == test_user.id
assert test_marketplace_job.user_id == test_user.id
def test_get_import_jobs_admin_sees_all(self, db, test_marketplace_job, test_admin):
"""Test that admin sees all import jobs"""
@@ -158,7 +205,7 @@ class TestMarketplaceService:
marketplace=f"Marketplace_{unique_id}_{i}",
shop_name=f"Test_Shop_{unique_id}_{i}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
shop_id=test_shop.id,
source_url=f"https://test-{i}.example.com/import",
imported_count=0,
updated_count=0,
@@ -172,6 +219,15 @@ class TestMarketplaceService:
assert len(jobs) <= 2 # Should be at most 2
def test_get_import_jobs_database_error(self, db_with_error, test_user):
"""Test getting import jobs handles database errors"""
with pytest.raises(ValidationException) as exc_info:
self.service.get_import_jobs(db_with_error, test_user)
exception = exc_info.value
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):
"""Test updating job status"""
result = self.service.update_job_status(
@@ -188,9 +244,22 @@ class TestMarketplaceService:
def test_update_job_status_not_found(self, db):
"""Test updating non-existent job status"""
with pytest.raises(ValueError, match="Marketplace import job not found"):
with pytest.raises(ImportJobNotFoundException) as exc_info:
self.service.update_job_status(db, 99999, "completed")
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_NOT_FOUND"
assert "99999" in exception.message
def test_update_job_status_database_error(self, db_with_error):
"""Test updating job status handles database errors"""
with pytest.raises(ValidationException) as exc_info:
self.service.update_job_status(db_with_error, 1, "completed")
exception = exc_info.value
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):
"""Test getting job statistics for user"""
stats = self.service.get_job_stats(db, test_user)
@@ -200,12 +269,23 @@ class TestMarketplaceService:
assert "running_jobs" in stats
assert "completed_jobs" in stats
assert "failed_jobs" in stats
assert isinstance(stats["total_jobs"], int)
def test_get_job_stats_admin(self, db, test_marketplace_job, test_admin):
"""Test getting job statistics for admin"""
stats = self.service.get_job_stats(db, test_admin)
assert stats["total_jobs"] >= 1
assert isinstance(stats["total_jobs"], int)
def test_get_job_stats_database_error(self, db_with_error, test_user):
"""Test getting job stats handles database errors"""
with pytest.raises(ValidationException) as exc_info:
self.service.get_job_stats(db_with_error, test_user)
exception = exc_info.value
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):
"""Test converting database model to response model"""
@@ -226,7 +306,7 @@ class TestMarketplaceService:
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
shop_id=test_shop.id,
source_url="https://test.example.com/import",
imported_count=0,
updated_count=0,
@@ -242,6 +322,22 @@ class TestMarketplaceService:
assert result.status == "cancelled"
assert result.completed_at is not None
def test_cancel_import_job_not_found(self, db, test_user):
"""Test cancelling non-existent import job"""
with pytest.raises(ImportJobNotFoundException) as exc_info:
self.service.cancel_import_job(db, 99999, test_user)
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):
"""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)
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
):
@@ -250,11 +346,14 @@ class TestMarketplaceService:
test_marketplace_job.status = "completed"
db.commit()
with pytest.raises(
ValueError, match="Cannot cancel job with status: completed"
):
with pytest.raises(ImportJobCannotBeCancelledException) as exc_info:
self.service.cancel_import_job(db, test_marketplace_job.id, test_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_CANNOT_BE_CANCELLED"
assert exception.status_code == 400
assert "completed" in exception.message
def test_delete_import_job_success(self, db, test_user, test_shop):
"""Test deleting a completed import job"""
unique_id = str(uuid.uuid4())[:8]
@@ -265,7 +364,7 @@ class TestMarketplaceService:
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
shop_id=test_shop.id,
source_url="https://test.example.com/import",
imported_count=0,
updated_count=0,
@@ -289,6 +388,22 @@ class TestMarketplaceService:
)
assert deleted_job is None
def test_delete_import_job_not_found(self, db, test_user):
"""Test deleting non-existent import job"""
with pytest.raises(ImportJobNotFoundException) as exc_info:
self.service.delete_import_job(db, 99999, test_user)
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):
"""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)
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):
"""Test deleting a job that can't be deleted"""
unique_id = str(uuid.uuid4())[:8]
@@ -299,7 +414,7 @@ class TestMarketplaceService:
marketplace="Amazon",
shop_name=f"TEST_SHOP_{unique_id}",
user_id=test_user.id,
shop_id=test_shop.id, # Use shop_id instead of shop_code
shop_id=test_shop.id,
source_url="https://test.example.com/import",
imported_count=0,
updated_count=0,
@@ -310,5 +425,39 @@ class TestMarketplaceService:
db.commit()
db.refresh(job)
with pytest.raises(ValueError, match="Cannot delete job with status: pending"):
with pytest.raises(ImportJobCannotBeDeletedException) as exc_info:
self.service.delete_import_job(db, job.id, test_user)
exception = exc_info.value
assert exception.error_code == "IMPORT_JOB_CANNOT_BE_DELETED"
assert exception.status_code == 400
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
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 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
def test_create_import_job_database_error(self, db_with_error, test_user):
"""Test import job creation handles database errors"""
request = MarketplaceImportRequest(
url="https://example.com/products.csv",
marketplace="Amazon",
shop_code="TEST_SHOP",
batch_size=1000,
)
with pytest.raises(ValidationException) as exc_info:
self.service.create_import_job(db_with_error, request, test_user)
exception = exc_info.value
assert exception.error_code == "VALIDATION_ERROR"