Samir Boulahtit
1cb659e3a5
All checks were successful
Refactor 10 db.add() loops to db.add_all() in services (menu, admin, orders, dev_tools), suppress 65 in tests/seeds/complex patterns with noqa: PERF006, suppress 2 polling interval warnings with noqa: PERF062, and add JS comment noqa support to base validator. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
413 lines
15 KiB
Python
413 lines
15 KiB
Python
# tests/unit/services/test_marketplace_service.py
|
|
"""Unit tests for MarketplaceImportJobService."""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from app.modules.marketplace.exceptions import (
|
|
ImportJobNotFoundException,
|
|
ImportJobNotOwnedException,
|
|
ImportValidationError,
|
|
)
|
|
from app.modules.marketplace.models import MarketplaceImportJob
|
|
from app.modules.marketplace.schemas import MarketplaceImportJobRequest
|
|
from app.modules.marketplace.services.marketplace_import_job_service import (
|
|
MarketplaceImportJobService,
|
|
)
|
|
from app.modules.tenancy.exceptions import UnauthorizedStoreAccessException
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.marketplace
|
|
class TestMarketplaceImportJobService:
|
|
"""Test suite for MarketplaceImportJobService."""
|
|
|
|
def setup_method(self):
|
|
self.service = MarketplaceImportJobService()
|
|
|
|
# ==================== create_import_job Tests ====================
|
|
|
|
def test_create_import_job_success(self, db, test_store, test_user):
|
|
"""Test successful creation of import job."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
marketplace="Amazon",
|
|
batch_size=1000,
|
|
)
|
|
|
|
result = self.service.create_import_job(db, request, test_store, test_user)
|
|
|
|
assert result.marketplace == "Amazon"
|
|
assert result.store_id == test_store.id
|
|
assert result.user_id == test_user.id
|
|
assert result.status == "pending"
|
|
assert result.source_url == "https://example.com/products.csv"
|
|
|
|
def test_create_import_job_default_marketplace(self, db, test_store, test_user):
|
|
"""Test import job creation with default marketplace."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
)
|
|
|
|
result = self.service.create_import_job(db, request, test_store, test_user)
|
|
|
|
assert result.marketplace == "Letzshop" # Default
|
|
|
|
def test_create_import_job_database_error(
|
|
self, db, test_store, test_user, monkeypatch
|
|
):
|
|
"""Test import job creation handles database errors."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
marketplace="Amazon",
|
|
)
|
|
|
|
def mock_flush():
|
|
raise SQLAlchemyError("Database flush failed")
|
|
|
|
monkeypatch.setattr(db, "flush", mock_flush)
|
|
|
|
with pytest.raises(ImportValidationError) as exc_info:
|
|
self.service.create_import_job(db, request, test_store, test_user)
|
|
|
|
exception = exc_info.value
|
|
assert exception.error_code == "IMPORT_VALIDATION_ERROR"
|
|
assert "Failed to create import job" in exception.message
|
|
|
|
# ==================== get_import_job_by_id Tests ====================
|
|
|
|
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_import_job.id, test_user
|
|
)
|
|
|
|
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_import_job, test_admin
|
|
):
|
|
"""Test that admin can access any import job."""
|
|
result = self.service.get_import_job_by_id(
|
|
db, test_marketplace_import_job.id, test_admin
|
|
)
|
|
|
|
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."""
|
|
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_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_import_job.id, other_user
|
|
)
|
|
|
|
exception = exc_info.value
|
|
assert exception.error_code == "IMPORT_JOB_NOT_OWNED"
|
|
assert exception.status_code == 403
|
|
|
|
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 SQLAlchemyError("Database query failed")
|
|
|
|
monkeypatch.setattr(db, "query", mock_query)
|
|
|
|
with pytest.raises(ImportValidationError) as exc_info:
|
|
self.service.get_import_job_by_id(db, 1, test_user)
|
|
|
|
exception = exc_info.value
|
|
assert exception.error_code == "IMPORT_VALIDATION_ERROR"
|
|
|
|
# ==================== get_import_job_for_store Tests ====================
|
|
|
|
def test_get_import_job_for_store_success(
|
|
self, db, test_marketplace_import_job, test_store
|
|
):
|
|
"""Test getting import job for store."""
|
|
result = self.service.get_import_job_for_store(
|
|
db, test_marketplace_import_job.id, test_store.id
|
|
)
|
|
|
|
assert result.id == test_marketplace_import_job.id
|
|
assert result.store_id == test_store.id
|
|
|
|
def test_get_import_job_for_store_not_found(self, db, test_store):
|
|
"""Test getting non-existent import job for store."""
|
|
with pytest.raises(ImportJobNotFoundException) as exc_info:
|
|
self.service.get_import_job_for_store(db, 99999, test_store.id)
|
|
|
|
exception = exc_info.value
|
|
assert exception.error_code == "IMPORT_JOB_NOT_FOUND"
|
|
|
|
def test_get_import_job_for_store_wrong_store(
|
|
self, db, test_marketplace_import_job, other_user, other_merchant
|
|
):
|
|
"""Test getting import job for wrong store."""
|
|
from app.modules.tenancy.models import Store
|
|
|
|
# Create another store
|
|
unique_id = str(uuid.uuid4())[:8]
|
|
other_store = Store(
|
|
merchant_id=other_merchant.id,
|
|
store_code=f"OTHER_{unique_id.upper()}",
|
|
subdomain=f"other{unique_id.lower()}",
|
|
name=f"Other Store {unique_id}",
|
|
is_active=True,
|
|
)
|
|
db.add(other_store)
|
|
db.commit()
|
|
db.refresh(other_store)
|
|
|
|
with pytest.raises(UnauthorizedStoreAccessException):
|
|
self.service.get_import_job_for_store(
|
|
db, test_marketplace_import_job.id, other_store.id
|
|
)
|
|
|
|
# ==================== get_import_jobs Tests ====================
|
|
|
|
def test_get_import_jobs_success(
|
|
self, db, test_marketplace_import_job, test_store, test_user
|
|
):
|
|
"""Test getting import jobs for store."""
|
|
jobs = self.service.get_import_jobs(db, test_store, test_user)
|
|
|
|
assert len(jobs) >= 1
|
|
assert any(job.id == test_marketplace_import_job.id for job in jobs)
|
|
|
|
def test_get_import_jobs_admin_sees_all_store_jobs(
|
|
self, db, test_marketplace_import_job, test_store, test_admin
|
|
):
|
|
"""Test that admin sees all store jobs."""
|
|
jobs = self.service.get_import_jobs(db, test_store, test_admin)
|
|
|
|
assert len(jobs) >= 1
|
|
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_import_job, test_store, test_user
|
|
):
|
|
"""Test getting import jobs with marketplace filter."""
|
|
jobs = self.service.get_import_jobs(
|
|
db,
|
|
test_store,
|
|
test_user,
|
|
marketplace=test_marketplace_import_job.marketplace,
|
|
)
|
|
|
|
assert len(jobs) >= 1
|
|
assert all(
|
|
test_marketplace_import_job.marketplace.lower() in job.marketplace.lower()
|
|
for job in jobs
|
|
)
|
|
|
|
def test_get_import_jobs_with_pagination(self, db, test_store, test_user):
|
|
"""Test getting import jobs with pagination."""
|
|
unique_id = str(uuid.uuid4())[:8]
|
|
|
|
# Create multiple import jobs
|
|
for i in range(5):
|
|
job = MarketplaceImportJob(
|
|
status="completed",
|
|
marketplace=f"Marketplace_{unique_id}_{i}",
|
|
store_id=test_store.id,
|
|
user_id=test_user.id,
|
|
source_url=f"https://test-{i}.example.com/import",
|
|
imported_count=0,
|
|
updated_count=0,
|
|
total_processed=0,
|
|
error_count=0,
|
|
)
|
|
db.add(job) # noqa: PERF006
|
|
db.commit()
|
|
|
|
jobs = self.service.get_import_jobs(db, test_store, test_user, skip=2, limit=2)
|
|
|
|
assert len(jobs) <= 2
|
|
|
|
def test_get_import_jobs_empty(self, db, test_user, other_user, other_merchant):
|
|
"""Test getting import jobs when none exist."""
|
|
from app.modules.tenancy.models import Store
|
|
|
|
# Create a store with no jobs
|
|
unique_id = str(uuid.uuid4())[:8]
|
|
empty_store = Store(
|
|
merchant_id=other_merchant.id,
|
|
store_code=f"EMPTY_{unique_id.upper()}",
|
|
subdomain=f"empty{unique_id.lower()}",
|
|
name=f"Empty Store {unique_id}",
|
|
is_active=True,
|
|
)
|
|
db.add(empty_store)
|
|
db.commit()
|
|
db.refresh(empty_store)
|
|
|
|
jobs = self.service.get_import_jobs(db, empty_store, other_user)
|
|
|
|
assert len(jobs) == 0
|
|
|
|
def test_get_import_jobs_database_error(
|
|
self, db, test_store, test_user, monkeypatch
|
|
):
|
|
"""Test get import jobs handles database errors."""
|
|
|
|
def mock_query(*args):
|
|
raise SQLAlchemyError("Database query failed")
|
|
|
|
monkeypatch.setattr(db, "query", mock_query)
|
|
|
|
with pytest.raises(ImportValidationError) as exc_info:
|
|
self.service.get_import_jobs(db, test_store, test_user)
|
|
|
|
exception = exc_info.value
|
|
assert exception.error_code == "IMPORT_VALIDATION_ERROR"
|
|
assert "Failed to retrieve import jobs" in exception.message
|
|
|
|
# ==================== convert_to_response_model Tests ====================
|
|
|
|
def test_convert_to_response_model(
|
|
self, db, test_marketplace_import_job, test_store
|
|
):
|
|
"""Test converting database model to response model."""
|
|
from app.modules.marketplace.models import MarketplaceImportJob as MIJ
|
|
|
|
# Re-query to get fresh instance with relationships
|
|
job = db.query(MIJ).filter(MIJ.id == test_marketplace_import_job.id).first()
|
|
|
|
response = self.service.convert_to_response_model(job)
|
|
|
|
assert response.job_id == job.id
|
|
assert response.status == job.status
|
|
assert response.marketplace == job.marketplace
|
|
assert response.store_id == job.store_id
|
|
assert response.imported == (job.imported_count or 0)
|
|
|
|
def test_convert_to_response_model_with_all_fields(
|
|
self, db, test_store, test_user
|
|
):
|
|
"""Test converting model with all fields populated."""
|
|
str(uuid.uuid4())[:8]
|
|
from datetime import datetime
|
|
|
|
job = MarketplaceImportJob(
|
|
status="completed",
|
|
marketplace="TestMarket",
|
|
store_id=test_store.id,
|
|
user_id=test_user.id,
|
|
source_url="https://test.example.com/import",
|
|
imported_count=100,
|
|
updated_count=50,
|
|
total_processed=150,
|
|
error_count=5,
|
|
error_message="Some errors occurred",
|
|
started_at=datetime.utcnow(),
|
|
completed_at=datetime.utcnow(),
|
|
)
|
|
db.add(job)
|
|
db.commit()
|
|
db.refresh(job)
|
|
|
|
response = self.service.convert_to_response_model(job)
|
|
|
|
assert response.imported == 100
|
|
assert response.updated == 50
|
|
assert response.total_processed == 150
|
|
assert response.error_count == 5
|
|
assert response.error_message == "Some errors occurred"
|
|
assert response.started_at is not None
|
|
assert response.completed_at is not None
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.marketplace
|
|
class TestMarketplaceImportJobSchema:
|
|
"""Test suite for MarketplaceImportJobRequest schema validation."""
|
|
|
|
def test_valid_request(self):
|
|
"""Test valid request schema."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
marketplace="Amazon",
|
|
batch_size=1000,
|
|
)
|
|
|
|
assert request.source_url == "https://example.com/products.csv"
|
|
assert request.marketplace == "Amazon"
|
|
assert request.batch_size == 1000
|
|
|
|
def test_default_values(self):
|
|
"""Test default values for optional fields."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
)
|
|
|
|
assert request.marketplace == "Letzshop"
|
|
assert request.batch_size == 1000
|
|
|
|
def test_url_validation_http(self):
|
|
"""Test URL validation accepts http."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="http://example.com/products.csv",
|
|
)
|
|
|
|
assert request.source_url == "http://example.com/products.csv"
|
|
|
|
def test_url_validation_invalid(self):
|
|
"""Test URL validation rejects invalid URLs."""
|
|
with pytest.raises(ValueError) as exc_info:
|
|
MarketplaceImportJobRequest(
|
|
source_url="ftp://example.com/products.csv",
|
|
)
|
|
|
|
assert "URL must start with http://" in str(exc_info.value) # noqa: SEC034
|
|
|
|
def test_url_with_trailing_slash(self):
|
|
"""Test URL with trailing slash is accepted."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv/",
|
|
)
|
|
|
|
assert request.source_url == "https://example.com/products.csv/"
|
|
|
|
def test_batch_size_validation_min(self):
|
|
"""Test batch_size validation minimum."""
|
|
with pytest.raises(ValueError):
|
|
MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
batch_size=50, # Below minimum of 100
|
|
)
|
|
|
|
def test_batch_size_validation_max(self):
|
|
"""Test batch_size validation maximum."""
|
|
with pytest.raises(ValueError):
|
|
MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
batch_size=20000, # Above maximum of 10000
|
|
)
|
|
|
|
def test_marketplace_strips_whitespace(self):
|
|
"""Test marketplace strips whitespace."""
|
|
request = MarketplaceImportJobRequest(
|
|
source_url="https://example.com/products.csv",
|
|
marketplace=" Amazon ",
|
|
)
|
|
|
|
assert request.marketplace == "Amazon"
|