Samir Boulahtit
0c63f387aa
- Move Product/ProductTranslation to app/modules/catalog/models/ - Move VendorOnboarding to app/modules/marketplace/models/ - Delete legacy re-export files for marketplace models: - letzshop.py, marketplace.py, marketplace_product.py - marketplace_product_translation.py, marketplace_import_job.py - Delete legacy product.py, product_translation.py, onboarding.py - Update all imports across services, tasks, tests to use module locations Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
601 lines
22 KiB
Python
601 lines
22 KiB
Python
# tests/unit/services/test_onboarding_service.py
|
|
"""
|
|
Unit tests for OnboardingService.
|
|
|
|
Tests cover:
|
|
- Onboarding CRUD operations
|
|
- Step completion logic
|
|
- Step order validation
|
|
- Order sync progress tracking
|
|
- Admin skip functionality
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.services.onboarding_service import OnboardingService
|
|
from app.modules.marketplace.models import OnboardingStatus, OnboardingStep, VendorOnboarding
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceCRUD:
|
|
"""Test CRUD operations"""
|
|
|
|
def test_get_onboarding_returns_existing(self, db, test_vendor):
|
|
"""Test get_onboarding returns existing record"""
|
|
# Create onboarding
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
current_step=OnboardingStep.LETZSHOP_API.value,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_onboarding(test_vendor.id)
|
|
|
|
assert result is not None
|
|
assert result.id == onboarding.id
|
|
assert result.vendor_id == test_vendor.id
|
|
|
|
def test_get_onboarding_returns_none_if_missing(self, db):
|
|
"""Test get_onboarding returns None if no record"""
|
|
service = OnboardingService(db)
|
|
result = service.get_onboarding(99999)
|
|
assert result is None
|
|
|
|
def test_get_onboarding_or_raise_raises_exception(self, db):
|
|
"""Test get_onboarding_or_raise raises OnboardingNotFoundException"""
|
|
from app.exceptions import OnboardingNotFoundException
|
|
|
|
service = OnboardingService(db)
|
|
with pytest.raises(OnboardingNotFoundException):
|
|
service.get_onboarding_or_raise(99999)
|
|
|
|
def test_create_onboarding_creates_new(self, db, test_vendor):
|
|
"""Test create_onboarding creates new record"""
|
|
service = OnboardingService(db)
|
|
result = service.create_onboarding(test_vendor.id)
|
|
|
|
assert result is not None
|
|
assert result.vendor_id == test_vendor.id
|
|
assert result.status == OnboardingStatus.NOT_STARTED.value
|
|
assert result.current_step == OnboardingStep.COMPANY_PROFILE.value
|
|
|
|
def test_create_onboarding_returns_existing(self, db, test_vendor):
|
|
"""Test create_onboarding returns existing record if already exists"""
|
|
# Create existing
|
|
existing = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
current_step=OnboardingStep.LETZSHOP_API.value,
|
|
)
|
|
db.add(existing)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
result = service.create_onboarding(test_vendor.id)
|
|
|
|
assert result.id == existing.id
|
|
assert result.status == OnboardingStatus.IN_PROGRESS.value
|
|
|
|
def test_get_or_create_creates_if_missing(self, db, test_vendor):
|
|
"""Test get_or_create_onboarding creates if missing"""
|
|
service = OnboardingService(db)
|
|
result = service.get_or_create_onboarding(test_vendor.id)
|
|
|
|
assert result is not None
|
|
assert result.vendor_id == test_vendor.id
|
|
|
|
def test_is_completed_returns_false_if_no_record(self, db):
|
|
"""Test is_completed returns False if no record"""
|
|
service = OnboardingService(db)
|
|
assert service.is_completed(99999) is False
|
|
|
|
def test_is_completed_returns_false_if_in_progress(self, db, test_vendor):
|
|
"""Test is_completed returns False if in progress"""
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
assert service.is_completed(test_vendor.id) is False
|
|
|
|
def test_is_completed_returns_true_if_completed(self, db, test_vendor):
|
|
"""Test is_completed returns True if completed"""
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.COMPLETED.value,
|
|
step_company_profile_completed=True,
|
|
step_letzshop_api_completed=True,
|
|
step_product_import_completed=True,
|
|
step_order_sync_completed=True,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
assert service.is_completed(test_vendor.id) is True
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceStatusResponse:
|
|
"""Test status response generation"""
|
|
|
|
def test_get_status_response_structure(self, db, test_vendor):
|
|
"""Test status response has correct structure"""
|
|
service = OnboardingService(db)
|
|
result = service.get_status_response(test_vendor.id)
|
|
|
|
assert "id" in result
|
|
assert "vendor_id" in result
|
|
assert "status" in result
|
|
assert "current_step" in result
|
|
assert "company_profile" in result
|
|
assert "letzshop_api" in result
|
|
assert "product_import" in result
|
|
assert "order_sync" in result
|
|
assert "completion_percentage" in result
|
|
assert "completed_steps_count" in result
|
|
assert "total_steps" in result
|
|
assert "is_completed" in result
|
|
|
|
def test_get_status_response_step_details(self, db, test_vendor):
|
|
"""Test status response has step details"""
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
step_company_profile_completed=True,
|
|
step_company_profile_data={"company_name": "Test"},
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_status_response(test_vendor.id)
|
|
|
|
assert result["company_profile"]["completed"] is True
|
|
assert result["company_profile"]["data"]["company_name"] == "Test"
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceStep1:
|
|
"""Test Step 1: Company Profile"""
|
|
|
|
def test_get_company_profile_data_empty_vendor(self, db):
|
|
"""Test get_company_profile_data returns empty for non-existent vendor"""
|
|
service = OnboardingService(db)
|
|
result = service.get_company_profile_data(99999)
|
|
assert result == {}
|
|
|
|
def test_get_company_profile_data_with_data(self, db, test_vendor):
|
|
"""Test get_company_profile_data returns vendor data"""
|
|
test_vendor.name = "Test Brand"
|
|
test_vendor.description = "Test Description"
|
|
test_vendor.default_language = "fr"
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_company_profile_data(test_vendor.id)
|
|
|
|
assert result["brand_name"] == "Test Brand"
|
|
assert result["description"] == "Test Description"
|
|
assert result["default_language"] == "fr"
|
|
|
|
def test_complete_company_profile_updates_status(self, db, test_vendor):
|
|
"""Test complete_company_profile updates onboarding status"""
|
|
service = OnboardingService(db)
|
|
result = service.complete_company_profile(
|
|
vendor_id=test_vendor.id,
|
|
company_name="Test Company",
|
|
brand_name="Test Brand",
|
|
default_language="en",
|
|
dashboard_language="en",
|
|
)
|
|
db.commit()
|
|
|
|
assert result["success"] is True
|
|
assert result["step_completed"] is True
|
|
assert result["next_step"] == OnboardingStep.LETZSHOP_API.value
|
|
|
|
# Verify onboarding updated
|
|
onboarding = service.get_onboarding(test_vendor.id)
|
|
assert onboarding.status == OnboardingStatus.IN_PROGRESS.value
|
|
assert onboarding.step_company_profile_completed is True
|
|
|
|
def test_complete_company_profile_raises_for_missing_vendor(self, db):
|
|
"""Test complete_company_profile raises for non-existent vendor"""
|
|
from app.exceptions import VendorNotFoundException
|
|
|
|
service = OnboardingService(db)
|
|
|
|
# Use a vendor_id that doesn't exist
|
|
# The service should check vendor exists before doing anything
|
|
non_existent_vendor_id = 999999
|
|
|
|
with pytest.raises(VendorNotFoundException):
|
|
service.complete_company_profile(
|
|
vendor_id=non_existent_vendor_id,
|
|
default_language="en",
|
|
dashboard_language="en",
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceStep2:
|
|
"""Test Step 2: Letzshop API Configuration"""
|
|
|
|
def test_test_letzshop_api_returns_result(self, db, test_vendor):
|
|
"""Test test_letzshop_api returns connection test result"""
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopCredentialsService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_instance.test_api_key.return_value = (True, 150.0, None)
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.test_letzshop_api(
|
|
api_key="test_key",
|
|
shop_slug="test-shop",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert "150" in result["message"]
|
|
|
|
def test_test_letzshop_api_returns_error(self, db, test_vendor):
|
|
"""Test test_letzshop_api returns error on failure"""
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopCredentialsService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_instance.test_api_key.return_value = (False, None, "Invalid API key")
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.test_letzshop_api(
|
|
api_key="invalid_key",
|
|
shop_slug="test-shop",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "Invalid API key" in result["message"]
|
|
|
|
def test_complete_letzshop_api_requires_step1(self, db, test_vendor):
|
|
"""Test complete_letzshop_api requires step 1 complete"""
|
|
from app.exceptions import OnboardingStepOrderException
|
|
|
|
# Create onboarding with step 1 not complete
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.NOT_STARTED.value,
|
|
current_step=OnboardingStep.COMPANY_PROFILE.value,
|
|
step_company_profile_completed=False,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
with pytest.raises(OnboardingStepOrderException):
|
|
service.complete_letzshop_api(
|
|
vendor_id=test_vendor.id,
|
|
api_key="test_key",
|
|
shop_slug="test-shop",
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceStep3:
|
|
"""Test Step 3: Product Import Configuration"""
|
|
|
|
def test_get_product_import_config_empty(self, db):
|
|
"""Test get_product_import_config returns empty for non-existent vendor"""
|
|
service = OnboardingService(db)
|
|
result = service.get_product_import_config(99999)
|
|
assert result == {}
|
|
|
|
def test_get_product_import_config_with_data(self, db, test_vendor):
|
|
"""Test get_product_import_config returns vendor CSV settings"""
|
|
test_vendor.letzshop_csv_url_fr = "https://example.com/fr.csv"
|
|
test_vendor.letzshop_default_tax_rate = 17
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_product_import_config(test_vendor.id)
|
|
|
|
assert result["csv_url_fr"] == "https://example.com/fr.csv"
|
|
assert result["default_tax_rate"] == 17
|
|
|
|
def test_complete_product_import_requires_csv_url(self, db, test_vendor):
|
|
"""Test complete_product_import requires at least one CSV URL"""
|
|
from app.exceptions import OnboardingCsvUrlRequiredException
|
|
|
|
# Create onboarding with steps 1 and 2 complete
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
current_step=OnboardingStep.PRODUCT_IMPORT.value,
|
|
step_company_profile_completed=True,
|
|
step_letzshop_api_completed=True,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
with pytest.raises(OnboardingCsvUrlRequiredException):
|
|
service.complete_product_import(
|
|
vendor_id=test_vendor.id,
|
|
# No CSV URLs provided
|
|
)
|
|
|
|
def test_complete_product_import_success(self, db, test_vendor):
|
|
"""Test complete_product_import saves settings"""
|
|
# Create onboarding with steps 1 and 2 complete
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
current_step=OnboardingStep.PRODUCT_IMPORT.value,
|
|
step_company_profile_completed=True,
|
|
step_letzshop_api_completed=True,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
service = OnboardingService(db)
|
|
result = service.complete_product_import(
|
|
vendor_id=test_vendor.id,
|
|
csv_url_fr="https://example.com/fr.csv",
|
|
default_tax_rate=17,
|
|
delivery_method="package_delivery",
|
|
preorder_days=2,
|
|
)
|
|
db.commit()
|
|
|
|
assert result["success"] is True
|
|
assert result["csv_urls_configured"] == 1
|
|
|
|
# Verify vendor updated
|
|
db.refresh(test_vendor)
|
|
assert test_vendor.letzshop_csv_url_fr == "https://example.com/fr.csv"
|
|
assert test_vendor.letzshop_default_tax_rate == 17
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceStep4:
|
|
"""Test Step 4: Order Sync"""
|
|
|
|
def test_trigger_order_sync_creates_job(self, db, test_vendor, test_user):
|
|
"""Test trigger_order_sync creates import job"""
|
|
# Create onboarding with steps 1-3 complete
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
current_step=OnboardingStep.ORDER_SYNC.value,
|
|
step_company_profile_completed=True,
|
|
step_letzshop_api_completed=True,
|
|
step_product_import_completed=True,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_instance.get_running_historical_import_job.return_value = None
|
|
mock_job = MagicMock()
|
|
mock_job.id = 123
|
|
mock_instance.create_historical_import_job.return_value = mock_job
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.trigger_order_sync(
|
|
vendor_id=test_vendor.id,
|
|
user_id=test_user.id,
|
|
days_back=90,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["job_id"] == 123
|
|
|
|
def test_trigger_order_sync_returns_existing_job(self, db, test_vendor, test_user):
|
|
"""Test trigger_order_sync returns existing job if running"""
|
|
# Create onboarding with steps 1-3 complete
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
current_step=OnboardingStep.ORDER_SYNC.value,
|
|
step_company_profile_completed=True,
|
|
step_letzshop_api_completed=True,
|
|
step_product_import_completed=True,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
existing_job = MagicMock()
|
|
existing_job.id = 456
|
|
mock_instance.get_running_historical_import_job.return_value = existing_job
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.trigger_order_sync(
|
|
vendor_id=test_vendor.id,
|
|
user_id=test_user.id,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["job_id"] == 456
|
|
assert "already running" in result["message"]
|
|
|
|
def test_get_order_sync_progress_not_found(self, db, test_vendor):
|
|
"""Test get_order_sync_progress for non-existent job"""
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_instance.get_historical_import_job_by_id.return_value = None
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_order_sync_progress(
|
|
vendor_id=test_vendor.id,
|
|
job_id=99999,
|
|
)
|
|
|
|
assert result["status"] == "not_found"
|
|
assert result["progress_percentage"] == 0
|
|
|
|
def test_get_order_sync_progress_completed(self, db, test_vendor):
|
|
"""Test get_order_sync_progress for completed job"""
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_job = MagicMock()
|
|
mock_job.id = 123
|
|
mock_job.status = "completed"
|
|
mock_job.current_phase = "complete"
|
|
mock_job.orders_imported = 50
|
|
mock_job.shipments_fetched = 50
|
|
mock_job.orders_processed = 50
|
|
mock_job.products_matched = 100
|
|
mock_job.started_at = datetime.now(UTC)
|
|
mock_job.completed_at = datetime.now(UTC)
|
|
mock_job.error_message = None
|
|
mock_instance.get_historical_import_job_by_id.return_value = mock_job
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_order_sync_progress(
|
|
vendor_id=test_vendor.id,
|
|
job_id=123,
|
|
)
|
|
|
|
assert result["status"] == "completed"
|
|
assert result["progress_percentage"] == 100
|
|
assert result["orders_imported"] == 50
|
|
|
|
def test_get_order_sync_progress_processing(self, db, test_vendor):
|
|
"""Test get_order_sync_progress for processing job"""
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_job = MagicMock()
|
|
mock_job.id = 123
|
|
mock_job.status = "processing"
|
|
mock_job.current_phase = "orders"
|
|
mock_job.orders_imported = 25
|
|
mock_job.shipments_fetched = 50
|
|
mock_job.orders_processed = 25
|
|
mock_job.products_matched = 50
|
|
mock_job.started_at = datetime.now(UTC)
|
|
mock_job.completed_at = None
|
|
mock_job.error_message = None
|
|
mock_job.total_pages = None
|
|
mock_job.current_page = None
|
|
mock_instance.get_historical_import_job_by_id.return_value = mock_job
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
result = service.get_order_sync_progress(
|
|
vendor_id=test_vendor.id,
|
|
job_id=123,
|
|
)
|
|
|
|
assert result["status"] == "processing"
|
|
assert result["progress_percentage"] == 50 # 25/50
|
|
assert result["current_phase"] == "orders"
|
|
|
|
def test_complete_order_sync_raises_for_missing_job(self, db, test_vendor):
|
|
"""Test complete_order_sync raises for non-existent job"""
|
|
from app.exceptions import OnboardingSyncJobNotFoundException
|
|
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_instance.get_historical_import_job_by_id.return_value = None
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
with pytest.raises(OnboardingSyncJobNotFoundException):
|
|
service.complete_order_sync(
|
|
vendor_id=test_vendor.id,
|
|
job_id=99999,
|
|
)
|
|
|
|
def test_complete_order_sync_raises_if_not_complete(self, db, test_vendor):
|
|
"""Test complete_order_sync raises if job still running"""
|
|
from app.exceptions import OnboardingSyncNotCompleteException
|
|
|
|
onboarding = VendorOnboarding(
|
|
vendor_id=test_vendor.id,
|
|
status=OnboardingStatus.IN_PROGRESS.value,
|
|
)
|
|
db.add(onboarding)
|
|
db.commit()
|
|
|
|
with patch(
|
|
"app.services.onboarding_service.LetzshopOrderService"
|
|
) as mock_service:
|
|
mock_instance = MagicMock()
|
|
mock_job = MagicMock()
|
|
mock_job.status = "processing"
|
|
mock_instance.get_historical_import_job_by_id.return_value = mock_job
|
|
mock_service.return_value = mock_instance
|
|
|
|
service = OnboardingService(db)
|
|
with pytest.raises(OnboardingSyncNotCompleteException):
|
|
service.complete_order_sync(
|
|
vendor_id=test_vendor.id,
|
|
job_id=123,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.service
|
|
class TestOnboardingServiceAdminSkip:
|
|
"""Test admin skip functionality"""
|
|
|
|
def test_skip_onboarding_success(self, db, test_vendor, test_admin):
|
|
"""Test skip_onboarding marks onboarding as skipped"""
|
|
service = OnboardingService(db)
|
|
result = service.skip_onboarding(
|
|
vendor_id=test_vendor.id,
|
|
admin_user_id=test_admin.id,
|
|
reason="Manual setup required",
|
|
)
|
|
db.commit()
|
|
|
|
assert result["success"] is True
|
|
|
|
# Verify onboarding updated
|
|
onboarding = service.get_onboarding(test_vendor.id)
|
|
assert onboarding.skipped_by_admin is True
|
|
assert onboarding.skipped_reason == "Manual setup required"
|
|
assert onboarding.status == OnboardingStatus.SKIPPED.value
|