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

fix: use custom exceptions in onboarding and add tests

Browse Source 
- Create onboarding-specific exceptions (OnboardingNotFoundException, etc.)
- Remove HTTPException usage from API endpoints per architecture rules
- Let exceptions propagate to global handler
- Add 12 integration tests for onboarding API endpoints

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-12-27 21:55:03 +01:00
parent 409a2eaa05
commit 73f612a01a
5 changed files with 538 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
app/api/v1/vendor/onboarding.py vendored
View File

@@ -13,16 +13,12 @@ Vendor Context: Uses token_vendor_id from JWT token (authenticated vendor API pa
import logging
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.api.deps import get_current_vendor_api
from app.core.database import get_db
from app.services.onboarding_service import (
OnboardingError,
OnboardingService,
OnboardingStepError,
)
from app.services.onboarding_service import OnboardingService
from models.database.user import User
from models.schema.onboarding import (
CompanyProfileRequest,
@@ -96,24 +92,19 @@ def save_company_profile(
Updates vendor and company records with provided data.
"""
service = OnboardingService(db)
try:
result = service.complete_company_profile(
vendor_id=current_user.token_vendor_id,
company_name=request.company_name,
brand_name=request.brand_name,
description=request.description,
contact_email=request.contact_email,
contact_phone=request.contact_phone,
website=request.website,
business_address=request.business_address,
tax_number=request.tax_number,
default_language=request.default_language,
dashboard_language=request.dashboard_language,
)
return result
except OnboardingError as e:
raise HTTPException(status_code=400, detail=str(e))
return service.complete_company_profile(
vendor_id=current_user.token_vendor_id,
company_name=request.company_name,
brand_name=request.brand_name,
description=request.description,
contact_email=request.contact_email,
contact_phone=request.contact_phone,
website=request.website,
business_address=request.business_address,
tax_number=request.tax_number,
default_language=request.default_language,
dashboard_language=request.dashboard_language,
)
# =============================================================================
@@ -151,19 +142,12 @@ def save_letzshop_api(
Tests connection first, only saves if successful.
"""
service = OnboardingService(db)
try:
result = service.complete_letzshop_api(
vendor_id=current_user.token_vendor_id,
api_key=request.api_key,
shop_slug=request.shop_slug,
letzshop_vendor_id=request.vendor_id,
)
return result
except OnboardingStepError as e:
raise HTTPException(status_code=400, detail=str(e))
except OnboardingError as e:
raise HTTPException(status_code=500, detail=str(e))
return service.complete_letzshop_api(
vendor_id=current_user.token_vendor_id,
api_key=request.api_key,
shop_slug=request.shop_slug,
letzshop_vendor_id=request.vendor_id,
)
# =============================================================================
@@ -197,22 +181,15 @@ def save_product_import_config(
At least one CSV URL must be provided.
"""
service = OnboardingService(db)
try:
result = service.complete_product_import(
vendor_id=current_user.token_vendor_id,
csv_url_fr=request.csv_url_fr,
csv_url_en=request.csv_url_en,
csv_url_de=request.csv_url_de,
default_tax_rate=request.default_tax_rate,
delivery_method=request.delivery_method,
preorder_days=request.preorder_days,
)
return result
except OnboardingStepError as e:
raise HTTPException(status_code=400, detail=str(e))
except OnboardingError as e:
raise HTTPException(status_code=500, detail=str(e))
return service.complete_product_import(
vendor_id=current_user.token_vendor_id,
csv_url_fr=request.csv_url_fr,
csv_url_en=request.csv_url_en,
csv_url_de=request.csv_url_de,
default_tax_rate=request.default_tax_rate,
delivery_method=request.delivery_method,
preorder_days=request.preorder_days,
)
# =============================================================================
@@ -232,19 +209,12 @@ def trigger_order_sync(
Creates a background job that imports orders from Letzshop.
"""
service = OnboardingService(db)
try:
result = service.trigger_order_sync(
vendor_id=current_user.token_vendor_id,
user_id=current_user.id,
days_back=request.days_back,
include_products=request.include_products,
)
return result
except OnboardingStepError as e:
raise HTTPException(status_code=400, detail=str(e))
except OnboardingError as e:
raise HTTPException(status_code=500, detail=str(e))
return service.trigger_order_sync(
vendor_id=current_user.token_vendor_id,
user_id=current_user.id,
days_back=request.days_back,
include_products=request.include_products,
)
@router.get(
@@ -281,14 +251,7 @@ def complete_order_sync(
This also marks the entire onboarding as complete.
"""
service = OnboardingService(db)
try:
result = service.complete_order_sync(
vendor_id=current_user.token_vendor_id,
job_id=request.job_id,
)
return result
except OnboardingStepError as e:
raise HTTPException(status_code=400, detail=str(e))
except OnboardingError as e:
raise HTTPException(status_code=500, detail=str(e))
return service.complete_order_sync(
vendor_id=current_user.token_vendor_id,
job_id=request.job_id,
)

19
app/exceptions/__init__.py
View File

@@ -252,6 +252,17 @@ from .vendor_theme import (
VendorThemeNotFoundException,
)
# Onboarding exceptions
from .onboarding import (
LetzshopConnectionFailedException,
OnboardingAlreadyCompletedException,
OnboardingCsvUrlRequiredException,
OnboardingNotFoundException,
OnboardingStepOrderException,
OnboardingSyncJobNotFoundException,
OnboardingSyncNotCompleteException,
)
__all__ = [
# Base exceptions
"WizamartException",
@@ -437,4 +448,12 @@ __all__ = [
"SubscriptionAlreadyCancelledException",
"InvalidWebhookSignatureException",
"WebhookMissingSignatureException",
# Onboarding exceptions
"OnboardingNotFoundException",
"OnboardingStepOrderException",
"OnboardingAlreadyCompletedException",
"LetzshopConnectionFailedException",
"OnboardingCsvUrlRequiredException",
"OnboardingSyncJobNotFoundException",
"OnboardingSyncNotCompleteException",
]

87
app/exceptions/onboarding.py Normal file
View File

@@ -0,0 +1,87 @@
# app/exceptions/onboarding.py
"""
Onboarding-specific exceptions.
Exceptions for the vendor onboarding wizard flow.
"""
from typing import Any
from .base import BusinessLogicException, ResourceNotFoundException, ValidationException
class OnboardingNotFoundException(ResourceNotFoundException):
"""Raised when onboarding record is not found for a vendor."""
def __init__(self, vendor_id: int):
super().__init__(
resource_type="VendorOnboarding",
identifier=str(vendor_id),
)
class OnboardingStepOrderException(ValidationException):
"""Raised when trying to access a step out of order."""
def __init__(self, current_step: str, required_step: str):
super().__init__(
message=f"Please complete the {required_step} step first",
field="step",
details={
"current_step": current_step,
"required_step": required_step,
},
)
class OnboardingAlreadyCompletedException(BusinessLogicException):
"""Raised when trying to modify a completed onboarding."""
def __init__(self, vendor_id: int):
super().__init__(
message="Onboarding has already been completed",
error_code="ONBOARDING_ALREADY_COMPLETED",
details={"vendor_id": vendor_id},
)
class LetzshopConnectionFailedException(BusinessLogicException):
"""Raised when Letzshop API connection test fails."""
def __init__(self, error_message: str):
super().__init__(
message=f"Letzshop connection failed: {error_message}",
error_code="LETZSHOP_CONNECTION_FAILED",
details={"error": error_message},
)
class OnboardingCsvUrlRequiredException(ValidationException):
"""Raised when no CSV URL is provided in product import step."""
def __init__(self):
super().__init__(
message="At least one CSV URL must be provided",
field="csv_url",
)
class OnboardingSyncJobNotFoundException(ResourceNotFoundException):
"""Raised when sync job is not found."""
def __init__(self, job_id: int):
super().__init__(
resource_type="LetzshopHistoricalImportJob",
identifier=str(job_id),
)
class OnboardingSyncNotCompleteException(BusinessLogicException):
"""Raised when trying to complete onboarding before sync is done."""
def __init__(self, job_status: str):
super().__init__(
message=f"Import job is still {job_status}, please wait",
error_code="SYNC_NOT_COMPLETE",
details={"job_status": job_status},
)

65
app/services/onboarding_service.py
View File

@@ -14,10 +14,16 @@ from datetime import UTC, datetime
from sqlalchemy.orm import Session
from app.exceptions import (
OnboardingCsvUrlRequiredException,
OnboardingNotFoundException,
OnboardingStepOrderException,
OnboardingSyncJobNotFoundException,
OnboardingSyncNotCompleteException,
VendorNotFoundException,
)
from app.services.letzshop.credentials_service import LetzshopCredentialsService
from app.services.letzshop.order_service import LetzshopOrderService
from models.database.company import Company
from models.database.letzshop import LetzshopHistoricalImportJob
from models.database.onboarding import (
OnboardingStatus,
OnboardingStep,
@@ -28,18 +34,6 @@ from models.database.vendor import Vendor
logger = logging.getLogger(__name__)
class OnboardingError(Exception):
"""Base exception for onboarding errors."""
class OnboardingNotFoundError(OnboardingError):
"""Raised when onboarding record is not found."""
class OnboardingStepError(OnboardingError):
"""Raised when a step operation fails."""
class OnboardingService:
"""
Service for managing vendor onboarding workflow.
@@ -69,12 +63,10 @@ class OnboardingService:
)
def get_onboarding_or_raise(self, vendor_id: int) -> VendorOnboarding:
"""Get onboarding record or raise OnboardingNotFoundError."""
"""Get onboarding record or raise OnboardingNotFoundException."""
onboarding = self.get_onboarding(vendor_id)
if onboarding is None:
raise OnboardingNotFoundError(
f"Onboarding not found for vendor {vendor_id}"
)
raise OnboardingNotFoundException(vendor_id)
return onboarding
def create_onboarding(self, vendor_id: int) -> VendorOnboarding:
@@ -221,7 +213,7 @@ class OnboardingService:
# Get vendor and company
vendor = self.db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
raise OnboardingStepError(f"Vendor {vendor_id} not found")
raise VendorNotFoundException(vendor_id)
company = vendor.company
@@ -326,8 +318,9 @@ class OnboardingService:
# Verify step order
if not onboarding.can_proceed_to_step(OnboardingStep.LETZSHOP_API.value):
raise OnboardingStepError(
"Please complete the Company Profile step first"
raise OnboardingStepOrderException(
current_step=onboarding.current_step,
required_step=OnboardingStep.COMPANY_PROFILE.value,
)
# Test connection first
@@ -412,8 +405,9 @@ class OnboardingService:
# Verify step order
if not onboarding.can_proceed_to_step(OnboardingStep.PRODUCT_IMPORT.value):
raise OnboardingStepError(
"Please complete the Letzshop API step first"
raise OnboardingStepOrderException(
current_step=onboarding.current_step,
required_step=OnboardingStep.LETZSHOP_API.value,
)
# Validate at least one CSV URL
@@ -424,18 +418,12 @@ class OnboardingService:
])
if csv_urls_count == 0:
return {
"success": False,
"step_completed": False,
"next_step": None,
"message": "At least one CSV URL must be provided",
"csv_urls_configured": 0,
}
raise OnboardingCsvUrlRequiredException()
# Update vendor settings
vendor = self.db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
raise OnboardingStepError(f"Vendor {vendor_id} not found")
raise VendorNotFoundException(vendor_id)
vendor.letzshop_csv_url_fr = csv_url_fr
vendor.letzshop_csv_url_en = csv_url_en
@@ -480,8 +468,9 @@ class OnboardingService:
# Verify step order
if not onboarding.can_proceed_to_step(OnboardingStep.ORDER_SYNC.value):
raise OnboardingStepError(
"Please complete the Product Import step first"
raise OnboardingStepOrderException(
current_step=onboarding.current_step,
required_step=OnboardingStep.PRODUCT_IMPORT.value,
)
# Create historical import job
@@ -597,16 +586,10 @@ class OnboardingService:
job = order_service.get_historical_import_job_by_id(vendor_id, job_id)
if not job:
raise OnboardingStepError(f"Job {job_id} not found")
raise OnboardingSyncJobNotFoundException(job_id)
if job.status not in ("completed", "failed"):
return {
"success": False,
"step_completed": False,
"onboarding_completed": False,
"message": f"Job is still {job.status}, please wait",
"redirect_url": None,
}
raise OnboardingSyncNotCompleteException(job.status)
# Mark step complete (even if job failed - they can retry later)
onboarding.mark_step_complete(OnboardingStep.ORDER_SYNC.value)

368
tests/integration/api/v1/vendor/test_onboarding.py vendored Normal file
View File

@@ -0,0 +1,368 @@
# tests/integration/api/v1/vendor/test_onboarding.py
"""
Integration tests for vendor onboarding API endpoints.
Tests cover:
1. Onboarding status retrieval
2. Step 1: Company profile setup
3. Step 2: Letzshop API configuration
4. Step 3: Product import configuration
5. Step 4: Order sync (mocked)
6. Step order validation (can't skip ahead)
7. Onboarding completion flow
"""
import pytest
from models.database.onboarding import OnboardingStatus, OnboardingStep, VendorOnboarding
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.vendor
class TestVendorOnboardingStatusAPI:
"""Test onboarding status endpoint"""
def test_get_status_creates_onboarding_if_missing(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test that getting status creates onboarding record if none exists"""
# First ensure no onboarding exists
existing = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
if existing:
db.delete(existing)
db.commit()
response = client.get(
"/api/v1/vendor/onboarding/status", headers=vendor_user_headers
)
assert response.status_code == 200
data = response.json()
assert data["vendor_id"] == test_vendor_with_vendor_user.id
assert data["status"] == OnboardingStatus.NOT_STARTED.value
assert data["current_step"] == OnboardingStep.COMPANY_PROFILE.value
assert data["completed_steps_count"] == 0
assert data["total_steps"] == 4
assert data["is_completed"] is False
def test_get_status_structure(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test status response has correct structure"""
response = client.get(
"/api/v1/vendor/onboarding/status", headers=vendor_user_headers
)
assert response.status_code == 200
data = response.json()
# Verify structure
assert "id" in data
assert "vendor_id" in data
assert "status" in data
assert "current_step" in data
assert "company_profile" in data
assert "letzshop_api" in data
assert "product_import" in data
assert "order_sync" in data
assert "completion_percentage" in data
assert "completed_steps_count" in data
assert "total_steps" in data
assert "is_completed" in data
def test_get_status_requires_auth(self, client):
"""Test that onboarding status requires authentication"""
response = client.get("/api/v1/vendor/onboarding/status")
assert response.status_code == 401
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.vendor
class TestVendorOnboardingStep1API:
"""Test Step 1: Company Profile endpoints"""
def test_get_company_profile_data(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test getting company profile data"""
response = client.get(
"/api/v1/vendor/onboarding/step/company-profile",
headers=vendor_user_headers,
)
assert response.status_code == 200
data = response.json()
# Should return pre-filled data
assert "brand_name" in data
assert "contact_email" in data
assert "default_language" in data
def test_save_company_profile_success(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test saving company profile completes step 1"""
# First ensure clean state
existing = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
if existing:
db.delete(existing)
db.commit()
response = client.post(
"/api/v1/vendor/onboarding/step/company-profile",
headers=vendor_user_headers,
json={
"company_name": "Test Company Ltd",
"brand_name": "Test Brand",
"description": "A test company for testing",
"contact_email": "test@example.com",
"contact_phone": "+352123456789",
"website": "https://test.example.com",
"business_address": "123 Test Street, Luxembourg",
"tax_number": "LU12345678",
"default_language": "fr",
"dashboard_language": "en",
},
)
assert response.status_code == 200
data = response.json()
assert data["success"] is True
assert data["step_completed"] is True
assert data["next_step"] == OnboardingStep.LETZSHOP_API.value
# Verify onboarding was updated
onboarding = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
db.refresh(onboarding)
assert onboarding.step_company_profile_completed is True
assert onboarding.status == OnboardingStatus.IN_PROGRESS.value
def test_save_company_profile_with_minimal_data(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test saving company profile with minimal data"""
# Clear existing onboarding
existing = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
if existing:
db.delete(existing)
db.commit()
response = client.post(
"/api/v1/vendor/onboarding/step/company-profile",
headers=vendor_user_headers,
json={
"default_language": "fr",
"dashboard_language": "fr",
},
)
assert response.status_code == 200
data = response.json()
assert data["success"] is True
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.vendor
class TestVendorOnboardingStep2API:
"""Test Step 2: Letzshop API Configuration endpoints"""
def test_letzshop_api_test_endpoint(
self, client, vendor_user_headers, test_vendor_with_vendor_user
):
"""Test the Letzshop API test endpoint"""
response = client.post(
"/api/v1/vendor/onboarding/step/letzshop-api/test",
headers=vendor_user_headers,
json={
"api_key": "test_invalid_key_12345",
"shop_slug": "test-shop",
},
)
assert response.status_code == 200
data = response.json()
# Response should indicate success or failure
assert "success" in data
assert "message" in data
def test_letzshop_api_requires_step1_complete(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test that step 2 requires step 1 to be completed"""
# Ensure fresh state with no steps completed
existing = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
if existing:
db.delete(existing)
db.commit()
# Try to save step 2 directly
response = client.post(
"/api/v1/vendor/onboarding/step/letzshop-api",
headers=vendor_user_headers,
json={
"api_key": "test_api_key_12345678901234567890",
"shop_slug": "test-shop",
},
)
# Should fail because step 1 is not complete
assert response.status_code == 422 # Validation error
data = response.json()
assert "step" in str(data).lower() or "complete" in str(data).lower()
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.vendor
class TestVendorOnboardingStep3API:
"""Test Step 3: Product Import Configuration endpoints"""
def test_get_product_import_config(
self, client, vendor_user_headers, test_vendor_with_vendor_user
):
"""Test getting product import configuration"""
response = client.get(
"/api/v1/vendor/onboarding/step/product-import",
headers=vendor_user_headers,
)
assert response.status_code == 200
data = response.json()
assert "csv_url_fr" in data
assert "csv_url_en" in data
assert "csv_url_de" in data
assert "default_tax_rate" in data
assert "delivery_method" in data
def test_product_import_requires_csv_url(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test that product import requires at least one CSV URL"""
# Set up: complete steps 1 and 2 first
onboarding = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
if not onboarding:
onboarding = VendorOnboarding(
vendor_id=test_vendor_with_vendor_user.id,
status=OnboardingStatus.IN_PROGRESS.value,
current_step=OnboardingStep.PRODUCT_IMPORT.value,
)
db.add(onboarding)
onboarding.step_company_profile_completed = True
onboarding.step_letzshop_api_completed = True
onboarding.current_step = OnboardingStep.PRODUCT_IMPORT.value
db.commit()
# Try to save without any CSV URL
response = client.post(
"/api/v1/vendor/onboarding/step/product-import",
headers=vendor_user_headers,
json={
"default_tax_rate": 17,
"delivery_method": "package_delivery",
"preorder_days": 1,
},
)
# Should fail with validation error
assert response.status_code == 422
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.vendor
class TestVendorOnboardingStep4API:
"""Test Step 4: Order Sync endpoints"""
def test_order_sync_progress_endpoint(
self, client, vendor_user_headers, test_vendor_with_vendor_user
):
"""Test getting order sync progress for non-existent job"""
response = client.get(
"/api/v1/vendor/onboarding/step/order-sync/progress/99999",
headers=vendor_user_headers,
)
assert response.status_code == 200
data = response.json()
# Should return not_found status for non-existent job
assert data["status"] == "not_found"
assert data["job_id"] == 99999
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.vendor
class TestVendorOnboardingFlowAPI:
"""Test complete onboarding flow"""
def test_completion_percentage_updates(
self, client, vendor_user_headers, test_vendor_with_vendor_user, db
):
"""Test that completion percentage updates as steps are completed"""
# Clear existing onboarding
existing = (
db.query(VendorOnboarding)
.filter(VendorOnboarding.vendor_id == test_vendor_with_vendor_user.id)
.first()
)
if existing:
db.delete(existing)
db.commit()
# Check initial state
response = client.get(
"/api/v1/vendor/onboarding/status", headers=vendor_user_headers
)
data = response.json()
assert data["completion_percentage"] == 0
# Complete step 1
client.post(
"/api/v1/vendor/onboarding/step/company-profile",
headers=vendor_user_headers,
json={
"default_language": "fr",
"dashboard_language": "fr",
},
)
# Check progress
response = client.get(
"/api/v1/vendor/onboarding/status", headers=vendor_user_headers
)
data = response.json()
assert data["completion_percentage"] == 25 # 1/4 = 25%
assert data["completed_steps_count"] == 1