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

Admin service tests update

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-24 21:02:17 +02:00
parent 98285aa8aa
commit 8b86b3225a
4 changed files with 526 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

608
docs/development/exception-handling.md
View File

@@ -2,7 +2,7 @@
## Overview
The LetzShop API uses a structured custom exception system to provide consistent, meaningful error responses across all endpoints. This guide covers how to work with exceptions, maintain the system, and extend it for new features.
The LetzShop API uses a unified custom exception system to provide consistent, meaningful error responses across all endpoints. This system was redesigned to eliminate competing exception handlers and provide a single source of truth for error handling.
## Architecture
@@ -10,14 +10,14 @@ The LetzShop API uses a structured custom exception system to provide consistent
```
LetzShopException (Base)
├── ValidationException (400)
├── ValidationException (422)
├── AuthenticationException (401)
├── AuthorizationException (403)
├── ResourceNotFoundException (404)
├── ConflictException (409)
├── BusinessLogicException (422)
├── BusinessLogicException (400)
├── RateLimitException (429)
└── ExternalServiceException (503)
└── ExternalServiceException (502)
```
### File Structure
@@ -26,8 +26,8 @@ LetzShopException (Base)
app/exceptions/
├── __init__.py # All exception exports
├── base.py # Base exception classes
├── handler.py # FastAPI exception handlers
├── auth.py # Authentication exceptions
├── handler.py # Unified FastAPI exception handlers
├── auth.py # Authentication/authorization exceptions
├── admin.py # Admin operation exceptions
├── marketplace.py # Import/marketplace exceptions
├── product.py # Product management exceptions
@@ -37,6 +37,10 @@ app/exceptions/
## Core Concepts
### Unified Exception Handling
The system uses **only FastAPI's built-in exception handlers** - no middleware or competing systems. All exceptions are processed through a single, consistent pipeline in `handler.py`.
### Exception Response Format
All custom exceptions return a consistent JSON structure:
@@ -46,7 +50,6 @@ All custom exceptions return a consistent JSON structure:
"error_code": "PRODUCT_NOT_FOUND",
"message": "Product with ID 'ABC123' not found",
"status_code": 404,
"field": "product_id",
"details": {
"resource_type": "Product",
"identifier": "ABC123"
@@ -54,125 +57,156 @@ All custom exceptions return a consistent JSON structure:
}
```
### Properties
### HTTP Status Codes
- **error_code**: Machine-readable identifier for programmatic handling
- **message**: Human-readable description for users
- **status_code**: HTTP status code
- **field**: Optional field name for validation errors
- **details**: Additional context data
The system uses appropriate HTTP status codes:
- **400**: Business logic violations (cannot modify own admin account)
- **401**: Authentication failures (invalid credentials, expired tokens)
- **403**: Authorization failures (admin privileges required)
- **404**: Resource not found (user, shop, product not found)
- **409**: Resource conflicts (duplicate shop codes, existing products)
- **422**: Validation errors (Pydantic validation failures)
- **429**: Rate limiting (too many requests)
- **500**: Internal server errors (unexpected exceptions)
## Implementation Details
### Exception Handler Setup
The main application registers all exception handlers in a single function:
```python
# main.py
from app.exceptions.handler import setup_exception_handlers
app = FastAPI(...)
setup_exception_handlers(app) # Single registration point
```
### Handler Types
The system handles four categories of exceptions:
```python
# app/exceptions/handler.py
@app.exception_handler(LetzShopException)
async def custom_exception_handler(request, exc):
"""Handle all custom business exceptions"""
return JSONResponse(
status_code=exc.status_code,
content=exc.to_dict()
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
"""Handle FastAPI HTTP exceptions"""
return JSONResponse(
status_code=exc.status_code,
content={
"error_code": f"HTTP_{exc.status_code}",
"message": exc.detail,
"status_code": exc.status_code,
}
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
"""Handle Pydantic validation errors"""
return JSONResponse(
status_code=422,
content={
"error_code": "VALIDATION_ERROR",
"message": "Request validation failed",
"status_code": 422,
"details": {"validation_errors": exc.errors()}
}
)
@app.exception_handler(Exception)
async def generic_exception_handler(request, exc):
"""Handle unexpected exceptions"""
return JSONResponse(
status_code=500,
content={
"error_code": "INTERNAL_SERVER_ERROR",
"message": "Internal server error",
"status_code": 500,
}
)
```
## Working with Exceptions
### Basic Usage
### Service Layer Pattern
Services raise custom exceptions with specific error codes:
```python
from app.exceptions import ProductNotFoundException, ProductAlreadyExistsException
# app/services/admin_service.py
from app.exceptions.admin import UserNotFoundException, CannotModifySelfException
# Simple not found
raise ProductNotFoundException("ABC123")
# Conflict with existing resource
raise ProductAlreadyExistsException("ABC123")
def toggle_user_status(self, db: Session, user_id: int, current_admin_id: int):
user = self._get_user_by_id_or_raise(db, user_id)
# Prevent self-modification
if user.id == current_admin_id:
raise CannotModifySelfException(user_id, "deactivate account")
# Prevent admin-to-admin modification
if user.role == "admin" and user.id != current_admin_id:
raise UserStatusChangeException(
user_id=user_id,
current_status="admin",
attempted_action="toggle status",
reason="Cannot modify another admin user"
)
```
### Validation Exceptions
### Authentication Integration
Authentication dependencies use custom exceptions instead of HTTPException:
```python
from app.exceptions import ValidationException, InvalidProductDataException
# middleware/auth.py
from app.exceptions.auth import AdminRequiredException, TokenExpiredException
# Generic validation error
raise ValidationException("Invalid data format", field="price")
def require_admin(self, current_user: User):
if current_user.role != "admin":
raise AdminRequiredException() # Returns error_code: "ADMIN_REQUIRED"
return current_user
# Specific product validation
raise InvalidProductDataException(
"Price must be positive",
field="price",
details={"provided_price": -10}
)
```
### Business Logic Exceptions
```python
from app.exceptions import InsufficientStockException
# Detailed business logic error
raise InsufficientStockException(
gtin="1234567890",
location="warehouse_a",
requested=100,
available=50
)
```
## Endpoint Implementation Patterns
### Service Layer Integration
Map service layer `ValueError` exceptions to specific custom exceptions:
```python
@router.post("/product")
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
def verify_token(self, token: str):
try:
result = product_service.create_product(db, product)
return result
except ValueError as e:
error_msg = str(e).lower()
if "already exists" in error_msg:
raise ProductAlreadyExistsException(product.product_id)
elif "invalid gtin" in error_msg:
raise InvalidProductDataException(str(e), field="gtin")
else:
raise ValidationException(str(e))
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise ValidationException("Failed to create product")
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
# ... token verification logic
except jwt.ExpiredSignatureError:
raise TokenExpiredException() # Returns error_code: "TOKEN_EXPIRED"
```
### Error Message Parsing
### Endpoint Implementation
Use regex patterns to extract context from service errors:
Endpoints rely on service layer exceptions and don't handle HTTPException:
```python
import re
try:
service.remove_stock(gtin, location, quantity)
except ValueError as e:
error_msg = str(e).lower()
if "insufficient stock" in error_msg:
# Extract quantities from error message
available_match = re.search(r'available[:\s]+(\d+)', error_msg)
requested_match = re.search(r'requested[:\s]+(\d+)', error_msg)
available = int(available_match.group(1)) if available_match else 0
requested = int(requested_match.group(1)) if requested_match else quantity
raise InsufficientStockException(gtin, location, requested, available)
```
### Exception Re-raising
Always re-raise custom exceptions to maintain their specific type:
```python
try:
shop = get_user_shop(shop_code, current_user, db)
# ... operations
except UnauthorizedShopAccessException:
raise # Re-raise without modification
except ValueError as e:
# Handle other ValueError cases
raise InvalidShopDataException(str(e))
# app/api/v1/admin.py
@router.put("/admin/users/{user_id}/status")
def toggle_user_status(
user_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user), # May raise AdminRequiredException
):
# Service raises UserNotFoundException, CannotModifySelfException, etc.
user, message = admin_service.toggle_user_status(db, user_id, current_admin.id)
return {"message": message}
```
## Creating New Exceptions
### Step 1: Define the Exception
### Step 1: Define Domain-Specific Exceptions
Create new exceptions in the appropriate module:
Create exceptions in the appropriate module:
```python
# app/exceptions/payment.py
@@ -206,286 +240,170 @@ class InsufficientFundsException(BusinessLogicException):
### Step 2: Export in __init__.py
Add new exceptions to the exports:
```python
# app/exceptions/__init__.py
from .payment import (
PaymentNotFoundException,
InsufficientFundsException,
)
from .payment import PaymentNotFoundException, InsufficientFundsException
__all__ = [
# ... existing exports
"PaymentNotFoundException",
"PaymentNotFoundException",
"InsufficientFundsException",
]
```
### Step 3: Use in Endpoints
Import and use the new exceptions:
### Step 3: Use in Services
```python
# app/services/payment_service.py
from app.exceptions import PaymentNotFoundException, InsufficientFundsException
@router.post("/payments/{payment_id}/process")
def process_payment(payment_id: str, db: Session = Depends(get_db)):
try:
result = payment_service.process_payment(db, payment_id)
return result
except ValueError as e:
if "not found" in str(e).lower():
raise PaymentNotFoundException(payment_id)
elif "insufficient funds" in str(e).lower():
# Extract amounts from error message
raise InsufficientFundsException(100.0, 50.0, "account_123")
else:
raise ValidationException(str(e))
def process_payment(self, db: Session, payment_id: str, amount: float):
payment = self._get_payment_by_id_or_raise(db, payment_id)
if payment.account_balance < amount:
raise InsufficientFundsException(amount, payment.account_balance, payment.account_id)
# Process payment...
return payment
```
## Testing Exceptions
## Testing Exception Handling
### Unit Tests
Test exception raising and properties:
Test exception properties and inheritance:
```python
# tests/unit/exceptions/test_admin_exceptions.py
import pytest
from app.exceptions import ProductNotFoundException, InsufficientStockException
from app.exceptions.admin import UserNotFoundException, CannotModifySelfException
def test_product_not_found_exception():
with pytest.raises(ProductNotFoundException) as exc_info:
raise ProductNotFoundException("ABC123")
def test_user_not_found_exception():
exc = UserNotFoundException("123")
exception = exc_info.value
assert exception.error_code == "PRODUCT_NOT_FOUND"
assert exception.status_code == 404
assert "ABC123" in exception.message
assert exception.details["identifier"] == "ABC123"
assert exc.error_code == "USER_NOT_FOUND"
assert exc.status_code == 404
assert "123" in exc.message
assert exc.details["identifier"] == "123"
def test_insufficient_stock_exception():
exc = InsufficientStockException(
gtin="1234567890",
location="warehouse",
requested=100,
available=50
)
def test_cannot_modify_self_exception():
exc = CannotModifySelfException(456, "deactivate account")
assert exc.error_code == "INSUFFICIENT_STOCK"
assert exc.status_code == 422
assert exc.details["requested_quantity"] == 100
assert exc.details["available_quantity"] == 50
assert exc.error_code == "CANNOT_MODIFY_SELF"
assert exc.status_code == 400
assert "deactivate account" in exc.message
```
### Integration Tests
Test endpoint exception handling:
Test complete error handling flow:
```python
def test_create_duplicate_product(client, auth_headers):
# Create initial product
product_data = {"product_id": "TEST123", "name": "Test Product"}
client.post("/api/v1/product", json=product_data, headers=auth_headers)
# Attempt to create duplicate
response = client.post("/api/v1/product", json=product_data, headers=auth_headers)
assert response.status_code == 409
# tests/integration/api/v1/test_admin_endpoints.py
def test_toggle_user_status_cannot_modify_self(client, admin_headers, test_admin):
"""Test that admin cannot modify their own account"""
response = client.put(
f"/api/v1/admin/users/{test_admin.id}/status",
headers=admin_headers
)
assert response.status_code == 400
data = response.json()
assert data["error_code"] == "PRODUCT_ALREADY_EXISTS"
assert "TEST123" in data["message"]
assert data["details"]["product_id"] == "TEST123"
assert data["error_code"] == "CANNOT_MODIFY_SELF"
assert "Cannot perform 'deactivate account' on your own account" in data["message"]
def test_get_all_users_non_admin(client, auth_headers):
"""Test non-admin trying to access admin endpoint"""
response = client.get("/api/v1/admin/users", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED" # Custom exception, not HTTP_403
assert "Admin privileges required" in data["message"]
```
## Migration from Legacy System
### What Was Changed
1. **Removed competing handlers**: Eliminated `ExceptionHandlerMiddleware` and duplicate exception handling
2. **Unified response format**: All exceptions now return consistent JSON structure
3. **Enhanced authentication**: Replaced `HTTPException` in auth layer with custom exceptions
4. **Improved logging**: Added structured logging with request context
### Breaking Changes
- Authentication failures now return `ADMIN_REQUIRED` instead of `HTTP_403`
- Rate limiting returns structured error with `RATE_LIMIT_EXCEEDED` code
- All custom exceptions use 400 for business logic errors (not 422)
### Migration Steps
1. **Update exception imports**: Replace `HTTPException` imports with custom exceptions
2. **Fix test expectations**: Update tests to expect specific error codes (e.g., `ADMIN_REQUIRED`)
3. **Remove legacy handlers**: Delete any remaining middleware or duplicate handlers
4. **Update frontend**: Map new error codes to user-friendly messages
## Error Code Reference
### Authentication (401)
- `INVALID_CREDENTIALS`: Invalid username or password
- `TOKEN_EXPIRED`: JWT token has expired
- `INVALID_TOKEN`: Malformed or invalid token
- `USER_NOT_ACTIVE`: User account is deactivated
### Authorization (403)
- `ADMIN_REQUIRED`: Admin privileges required for operation
- `INSUFFICIENT_PERMISSIONS`: User lacks required permissions
- `UNAUTHORIZED_SHOP_ACCESS`: Cannot access shop (not owner)
### Resource Not Found (404)
- `USER_NOT_FOUND`: User with specified ID not found
- `SHOP_NOT_FOUND`: Shop with specified code/ID not found
- `PRODUCT_NOT_FOUND`: Product with specified ID not found
### Business Logic (400)
- `CANNOT_MODIFY_SELF`: Admin cannot modify own account
- `USER_STATUS_CHANGE_FAILED`: Cannot change user status
- `SHOP_VERIFICATION_FAILED`: Shop verification operation failed
- `ADMIN_OPERATION_FAILED`: Generic admin operation failure
### Validation (422)
- `VALIDATION_ERROR`: Pydantic request validation failed
- `INVALID_PRODUCT_DATA`: Product data validation failed
- `INVALID_SHOP_DATA`: Shop data validation failed
### Rate Limiting (429)
- `RATE_LIMIT_EXCEEDED`: API rate limit exceeded
### Service Errors (500/502)
- `INTERNAL_SERVER_ERROR`: Unexpected server error
- `EXTERNAL_SERVICE_ERROR`: Third-party service failure
## Best Practices
### Do's
### Exception Design
- Use specific error codes for programmatic handling
- Include relevant context in exception details
- Inherit from appropriate base exception class
- Provide clear, actionable error messages
- **Use specific exceptions** rather than generic ValidationException
- **Include relevant context** in exception details
- **Map service errors** to appropriate custom exceptions
- **Re-raise custom exceptions** without modification
- **Log errors** before raising exceptions
- **Extract data** from error messages when possible
### Service Layer
- Raise custom exceptions, never HTTPException
- Include identifiers and context in exception details
- Use private helper methods with `_or_raise` suffix
- Log errors before raising exceptions
### Don'ts
### Testing
- Test both exception raising and response format
- Verify error codes, status codes, and message content
- Test edge cases and error conditions
- Use custom exceptions in test assertions
- **Don't use HTTPException** in endpoints (use custom exceptions)
- **Don't lose exception context** when mapping errors
- **Don't include sensitive data** in error messages
- **Don't create exceptions** for every possible error case
- **Don't ignore the exception hierarchy** when creating new exceptions
### Frontend Integration
- Map error codes to user-friendly messages
- Handle field-specific validation errors
- Implement retry logic for transient errors
- Log errors for debugging and monitoring
### Error Message Guidelines
```python
# Good: Specific, actionable message
raise InsufficientStockException(
gtin="1234567890",
location="warehouse_a",
requested=100,
available=50
)
# Bad: Generic, unhelpful message
raise ValidationException("Operation failed")
# Good: Include identifier context
raise ProductNotFoundException("ABC123")
# Bad: No context about what wasn't found
raise ValidationException("Not found")
```
## Debugging and Logging
### Exception Logging
The exception handler automatically logs all exceptions:
```python
# Logs include structured data
logger.error(
f"Custom exception in {request.method} {request.url}: "
f"{exc.error_code} - {exc.message}",
extra={
"error_code": exc.error_code,
"status_code": exc.status_code,
"details": exc.details,
"url": str(request.url),
"method": request.method,
}
)
```
### Adding Debug Information
Include additional context in exception details:
```python
try:
result = complex_operation()
except ValueError as e:
raise BusinessLogicException(
message="Complex operation failed",
error_code="COMPLEX_OPERATION_FAILED",
details={
"original_error": str(e),
"operation_params": {"param1": value1, "param2": value2},
"timestamp": datetime.utcnow().isoformat(),
}
)
```
## Frontend Integration
### Error Code Handling
Frontend can handle errors programmatically:
```javascript
// JavaScript example
try {
const response = await api.createProduct(productData);
} catch (error) {
switch (error.error_code) {
case 'PRODUCT_ALREADY_EXISTS':
showError('Product ID already exists. Please choose a different ID.');
break;
case 'INVALID_PRODUCT_DATA':
showFieldError(error.field, error.message);
break;
default:
showGenericError(error.message);
}
}
```
### User-Friendly Messages
Map error codes to user-friendly messages:
```javascript
const ERROR_MESSAGES = {
'PRODUCT_NOT_FOUND': 'The requested product could not be found.',
'INSUFFICIENT_STOCK': 'Not enough stock available for this operation.',
'UNAUTHORIZED_SHOP_ACCESS': 'You do not have permission to access this shop.',
'IMPORT_JOB_CANNOT_BE_CANCELLED': 'This import job cannot be cancelled at this time.',
};
```
## Monitoring and Metrics
### Error Tracking
Track exception frequency and patterns:
```python
# Add to exception handler
from prometheus_client import Counter
exception_counter = Counter(
'api_exceptions_total',
'Total number of API exceptions',
['error_code', 'endpoint', 'method']
)
# In exception handler
exception_counter.labels(
error_code=exc.error_code,
endpoint=request.url.path,
method=request.method
).inc()
```
### Health Monitoring
Monitor exception rates for system health:
- Track error rates per endpoint
- Alert on unusual exception patterns
- Monitor specific business logic errors
- Track client vs server error ratios
## Migration Guide
### From HTTPException to Custom Exceptions
Replace existing HTTPException usage:
```python
# Before
from fastapi import HTTPException
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# After
from app.exceptions import ProductNotFoundException
if not product:
raise ProductNotFoundException(product_id)
```
### Updating Service Layer
Enhance service layer error messages:
```python
# Before
def get_product(db, product_id):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise ValueError("Not found")
return product
# After
def get_product(db, product_id):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise ValueError(f"Product not found: {product_id}")
return product
```
This exception system provides a robust foundation for error handling that scales with your application while maintaining consistency for both developers and frontend clients.
This unified exception system provides consistent error handling across your entire application while maintaining clean separation of concerns and excellent developer experience.

1
tests/conftest.py
View File

@@ -90,4 +90,5 @@ pytest_plugins = [
"tests.fixtures.product_fixtures",
"tests.fixtures.shop_fixtures",
"tests.fixtures.marketplace_fixtures",
"tests.fixtures.testing_fixtures",
]

54
tests/fixtures/testing_fixtures.py vendored Normal file
View File

@@ -0,0 +1,54 @@
# tests/fixtures/testing_fixtures.py
"""
Testing utility fixtures for edge cases and error handling.
This module provides fixtures for:
- Empty database sessions for edge case testing
- Mock database sessions for error simulation
- Additional testing utilities
"""
import pytest
from unittest.mock import Mock
from sqlalchemy.exc import SQLAlchemyError
@pytest.fixture
def empty_db(db):
"""Empty database session for edge case testing"""
from sqlalchemy import text
# Clear only the tables that are relevant for admin service testing
# In order to respect foreign key constraints
tables_to_clear = [
"marketplace_import_jobs", # Has foreign keys to shops and users
"shop_products", # Has foreign keys to shops and products
"stock", # Fixed: singular not plural
"products", # Referenced by shop_products
"shops", # Has foreign key to users
"users" # Base table
]
for table in tables_to_clear:
try:
db.execute(text(f"DELETE FROM {table}"))
except Exception:
# If table doesn't exist or delete fails, continue
pass
db.commit()
return db
@pytest.fixture
def db_with_error():
"""Database session that raises errors for testing error handling"""
mock_db = Mock()
# Configure the mock to raise SQLAlchemy errors on query operations
mock_db.query.side_effect = SQLAlchemyError("Database connection failed")
mock_db.add.side_effect = SQLAlchemyError("Database insert failed")
mock_db.commit.side_effect = SQLAlchemyError("Database commit failed")
mock_db.rollback.return_value = None
return mock_db

225
tests/unit/services/test_admin_service.py
View File

@@ -1,7 +1,14 @@
# tests/unit/services/test_admin_service.py
import pytest
from fastapi import HTTPException
from app.exceptions import (
UserNotFoundException,
UserStatusChangeException,
CannotModifySelfException,
ShopNotFoundException,
ShopVerificationException,
AdminOperationException,
)
from app.services.admin_service import AdminService
from models.database.marketplace import MarketplaceImportJob
from models.database.shop import Shop
@@ -16,6 +23,7 @@ class TestAdminService:
"""Setup method following the same pattern as product service tests"""
self.service = AdminService()
# User Management Tests
def test_get_all_users(self, db, test_user, test_admin):
"""Test getting all users with pagination"""
users = self.service.get_all_users(db, skip=0, limit=10)
@@ -28,7 +36,6 @@ class TestAdminService:
def test_get_all_users_with_pagination(self, db, test_user, test_admin):
"""Test user pagination works correctly"""
users = self.service.get_all_users(db, skip=0, limit=1)
assert len(users) == 1
users_second_page = self.service.get_all_users(db, skip=1, limit=1)
@@ -43,7 +50,8 @@ class TestAdminService:
assert user.id == test_user.id
assert user.is_active is False
assert f"{user.username} has been deactivated" in message
assert test_user.username in message
assert "deactivated" in message
def test_toggle_user_status_activate(self, db, test_user, test_admin):
"""Test activating a user"""
@@ -55,24 +63,37 @@ class TestAdminService:
assert user.id == test_user.id
assert user.is_active is True
assert f"{user.username} has been activated" in message
assert test_user.username in message
assert "activated" in message
def test_toggle_user_status_user_not_found(self, db, test_admin):
"""Test toggle user status when user not found"""
with pytest.raises(HTTPException) as exc_info:
with pytest.raises(UserNotFoundException) as exc_info:
self.service.toggle_user_status(db, 99999, test_admin.id)
assert exc_info.value.status_code == 404
assert "User not found" in str(exc_info.value.detail)
exception = exc_info.value
assert exception.error_code == "USER_NOT_FOUND"
assert "99999" in exception.message
def test_toggle_user_status_cannot_deactivate_self(self, db, test_admin):
"""Test that admin cannot deactivate their own account"""
with pytest.raises(HTTPException) as exc_info:
def test_toggle_user_status_cannot_modify_self(self, db, test_admin):
"""Test that admin cannot modify their own account"""
with pytest.raises(CannotModifySelfException) as exc_info:
self.service.toggle_user_status(db, test_admin.id, test_admin.id)
assert exc_info.value.status_code == 400
assert "Cannot deactivate your own account" in str(exc_info.value.detail)
exception = exc_info.value
assert exception.error_code == "CANNOT_MODIFY_SELF"
assert "deactivate account" in exception.message
def test_toggle_user_status_cannot_modify_admin(self, db, test_admin, another_admin):
"""Test that admin cannot modify another admin"""
with pytest.raises(UserStatusChangeException) as exc_info:
self.service.toggle_user_status(db, another_admin.id, test_admin.id)
exception = exc_info.value
assert exception.error_code == "USER_STATUS_CHANGE_FAILED"
assert "Cannot modify another admin user" in exception.message
# Shop Management Tests
def test_get_all_shops(self, db, test_shop):
"""Test getting all shops with total count"""
shops, total = self.service.get_all_shops(db, skip=0, limit=10)
@@ -82,6 +103,18 @@ class TestAdminService:
shop_codes = [shop.shop_code for shop in shops]
assert test_shop.shop_code in shop_codes
def test_get_all_shops_with_pagination(self, db, test_shop, verified_shop):
"""Test shop pagination works correctly"""
shops, total = self.service.get_all_shops(db, skip=0, limit=1)
assert total >= 2
assert len(shops) == 1
shops_second_page, _ = self.service.get_all_shops(db, skip=1, limit=1)
assert len(shops_second_page) >= 0
if len(shops_second_page) > 0:
assert shops[0].id != shops_second_page[0].id
def test_verify_shop_mark_verified(self, db, test_shop):
"""Test marking shop as verified"""
# Ensure shop starts unverified
@@ -92,18 +125,52 @@ class TestAdminService:
assert shop.id == test_shop.id
assert shop.is_verified is True
assert f"{shop.shop_code} has been verified" in message
assert test_shop.shop_code in message
assert "verified" in message
def test_verify_shop_mark_unverified(self, db, verified_shop):
"""Test marking verified shop as unverified"""
shop, message = self.service.verify_shop(db, verified_shop.id)
assert shop.id == verified_shop.id
assert shop.is_verified is False
assert verified_shop.shop_code in message
assert "unverified" in message
def test_verify_shop_not_found(self, db):
"""Test verify shop when shop not found"""
with pytest.raises(HTTPException) as exc_info:
with pytest.raises(ShopNotFoundException) as exc_info:
self.service.verify_shop(db, 99999)
assert exc_info.value.status_code == 404
assert "Shop not found" in str(exc_info.value.detail)
exception = exc_info.value
assert exception.error_code == "SHOP_NOT_FOUND"
assert "99999" in exception.message
def test_toggle_shop_status_deactivate(self, db, test_shop):
"""Test deactivating a shop"""
original_status = test_shop.is_active
shop, message = self.service.toggle_shop_status(db, test_shop.id)
assert shop.id == test_shop.id
assert shop.is_active != original_status
assert test_shop.shop_code in message
if original_status:
assert "deactivated" in message
else:
assert "activated" in message
def test_toggle_shop_status_not_found(self, db):
"""Test toggle shop status when shop not found"""
with pytest.raises(ShopNotFoundException) as exc_info:
self.service.toggle_shop_status(db, 99999)
exception = exc_info.value
assert exception.error_code == "SHOP_NOT_FOUND"
# Marketplace Import Jobs Tests
def test_get_marketplace_import_jobs_no_filters(self, db, test_marketplace_job):
"""Test getting marketplace import jobs without filters using fixture"""
"""Test getting marketplace import jobs without filters"""
result = self.service.get_marketplace_import_jobs(db, skip=0, limit=10)
assert len(result) >= 1
@@ -115,3 +182,127 @@ class TestAdminService:
assert test_job.marketplace == test_marketplace_job.marketplace
assert test_job.shop_name == test_marketplace_job.shop_name
assert test_job.status == test_marketplace_job.status
def test_get_marketplace_import_jobs_with_marketplace_filter(self, db, test_marketplace_job):
"""Test filtering marketplace import jobs by marketplace"""
result = self.service.get_marketplace_import_jobs(
db, marketplace=test_marketplace_job.marketplace, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert test_marketplace_job.marketplace.lower() in job.marketplace.lower()
def test_get_marketplace_import_jobs_with_shop_filter(self, db, test_marketplace_job):
"""Test filtering marketplace import jobs by shop name"""
result = self.service.get_marketplace_import_jobs(
db, shop_name=test_marketplace_job.shop_name, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert test_marketplace_job.shop_name.lower() in job.shop_name.lower()
def test_get_marketplace_import_jobs_with_status_filter(self, db, test_marketplace_job):
"""Test filtering marketplace import jobs by status"""
result = self.service.get_marketplace_import_jobs(
db, status=test_marketplace_job.status, skip=0, limit=10
)
assert len(result) >= 1
for job in result:
assert job.status == test_marketplace_job.status
def test_get_marketplace_import_jobs_pagination(self, db, test_marketplace_job):
"""Test marketplace import jobs pagination"""
result_page1 = self.service.get_marketplace_import_jobs(db, skip=0, limit=1)
result_page2 = self.service.get_marketplace_import_jobs(db, skip=1, limit=1)
assert len(result_page1) >= 0
assert len(result_page2) >= 0
if len(result_page1) > 0 and len(result_page2) > 0:
assert result_page1[0].job_id != result_page2[0].job_id
# Statistics Tests
def test_get_user_statistics(self, db, test_user, test_admin):
"""Test getting user statistics"""
stats = self.service.get_user_statistics(db)
assert "total_users" in stats
assert "active_users" in stats
assert "inactive_users" in stats
assert "activation_rate" in stats
assert isinstance(stats["total_users"], int)
assert isinstance(stats["active_users"], int)
assert isinstance(stats["inactive_users"], int)
assert isinstance(stats["activation_rate"], (int, float))
assert stats["total_users"] >= 2 # test_user + test_admin
assert stats["active_users"] + stats["inactive_users"] == stats["total_users"]
def test_get_shop_statistics(self, db, test_shop):
"""Test getting shop statistics"""
stats = self.service.get_shop_statistics(db)
assert "total_shops" in stats
assert "active_shops" in stats
assert "verified_shops" in stats
assert "verification_rate" in stats
assert isinstance(stats["total_shops"], int)
assert isinstance(stats["active_shops"], int)
assert isinstance(stats["verified_shops"], int)
assert isinstance(stats["verification_rate"], (int, float))
assert stats["total_shops"] >= 1
# Error Handling Tests
def test_get_all_users_database_error(self, db_with_error, test_admin):
"""Test handling database errors in get_all_users"""
with pytest.raises(AdminOperationException) as exc_info:
self.service.get_all_users(db_with_error, skip=0, limit=10)
exception = exc_info.value
assert exception.error_code == "ADMIN_OPERATION_FAILED"
assert "get_all_users" in exception.message
def test_get_all_shops_database_error(self, db_with_error):
"""Test handling database errors in get_all_shops"""
with pytest.raises(AdminOperationException) as exc_info:
self.service.get_all_shops(db_with_error, skip=0, limit=10)
exception = exc_info.value
assert exception.error_code == "ADMIN_OPERATION_FAILED"
assert "get_all_shops" in exception.message
# Edge Cases
def test_get_all_users_empty_database(self, empty_db):
"""Test getting users when database is empty"""
users = self.service.get_all_users(empty_db, skip=0, limit=10)
assert len(users) == 0
def test_get_all_shops_empty_database(self, empty_db):
"""Test getting shops when database is empty"""
shops, total = self.service.get_all_shops(empty_db, skip=0, limit=10)
assert len(shops) == 0
assert total == 0
def test_user_statistics_empty_database(self, empty_db):
"""Test user statistics when no users exist"""
stats = self.service.get_user_statistics(empty_db)
assert stats["total_users"] == 0
assert stats["active_users"] == 0
assert stats["inactive_users"] == 0
assert stats["activation_rate"] == 0
def test_shop_statistics_empty_database(self, empty_db):
"""Test shop statistics when no shops exist"""
stats = self.service.get_shop_statistics(empty_db)
assert stats["total_shops"] == 0
assert stats["active_shops"] == 0
assert stats["verified_shops"] == 0
assert stats["verification_rate"] == 0