""" API usage examples and documentation tests for the declarative transition system. These tests serve as both validation and comprehensive documentation, showing how to integrate the transition system with APIs, handle JSON serialization, generate schemas, and implement real-world patterns. """ import json from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from unittest.mock import Mock import pytest from django.test import TestCase from fsm.registry import register_state_transition, transition_registry from fsm.transition_utils import ( get_transition_schema, ) from fsm.transitions import ( BaseTransition, TransitionContext, TransitionValidationError, ) from pydantic import Field, validator class APIIntegrationExampleTests(TestCase): """ API integration examples demonstrating real-world usage patterns. These tests show how to integrate the transition system with REST APIs, handle JSON data, validate requests, and format responses. """ def setUp(self): from copy import deepcopy self.mock_entity = Mock() self.mock_entity.pk = 1 self.mock_entity._meta.model_name = 'task' self.mock_entity.organization_id = 100 self.mock_user = Mock() self.mock_user.id = 42 self.mock_user.username = 'api_user' # Save registry state and clear for this test self._original_transitions = deepcopy(transition_registry._transitions) transition_registry._transitions.clear() def tearDown(self): # Restore original transition registry to prevent test leakage transition_registry._transitions = self._original_transitions def test_rest_api_task_assignment_example(self): """ API EXAMPLE: REST endpoint for task assignment Shows how to implement a REST API endpoint that uses declarative transitions with proper validation and error handling. """ @register_state_transition('task', 'api_assign_task') class APITaskAssignmentTransition(BaseTransition): """Task assignment via API with comprehensive validation""" assignee_id: int = Field(..., description='ID of user to assign task to') priority: str = Field('normal', description='Task priority level') deadline: Optional[datetime] = Field(None, description='Assignment deadline') assignment_notes: str = Field('', description='Notes about the assignment') notify_assignee: bool = Field(True, description='Whether to notify the assignee') @validator('priority') def validate_priority(cls, v): valid_priorities = ['low', 'normal', 'high', 'urgent'] if v not in valid_priorities: raise ValueError(f'Priority must be one of: {valid_priorities}') return v @validator('deadline') def validate_deadline(cls, v): if v and v <= datetime.now(): raise ValueError('Deadline must be in the future') return v def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'ASSIGNED' def validate_transition(self, context: TransitionContext) -> bool: # Business logic validation if context.current_state not in ['CREATED', 'UNASSIGNED']: raise TransitionValidationError( f'Cannot assign task in state: {context.current_state}', {'valid_states': ['CREATED', 'UNASSIGNED']}, ) # Mock user existence check if self.assignee_id <= 0: raise TransitionValidationError('Invalid assignee ID', {'assignee_id': self.assignee_id}) return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'assignee_id': self.assignee_id, 'priority': self.priority, 'deadline': self.deadline.isoformat() if self.deadline else None, 'assignment_notes': self.assignment_notes, 'notify_assignee': self.notify_assignee, 'assigned_by_id': context.current_user.id if context.current_user else None, 'assigned_at': context.timestamp.isoformat(), 'api_version': 'v1', } # Simulate API request data (JSON from client) api_request_data = { 'assignee_id': 123, 'priority': 'high', 'deadline': (datetime.now() + timedelta(days=7)).isoformat(), 'assignment_notes': 'Critical task requiring immediate attention', 'notify_assignee': True, } # API endpoint simulation: Parse and validate JSON try: # Step 1: Create transition from API data transition = APITaskAssignmentTransition(**api_request_data) # Step 2: Execute transition context = TransitionContext( entity=self.mock_entity, current_user=self.mock_user, current_state='CREATED', target_state=transition.get_target_state(), request_data=api_request_data, ) # Validate assert transition.validate_transition(context) # Execute result_data = transition.transition(context) # Step 3: Format API response api_response = { 'success': True, 'message': 'Task assigned successfully', 'data': { 'task_id': self.mock_entity.pk, 'new_state': transition.get_target_state(), 'assignment_details': result_data, }, 'timestamp': datetime.now().isoformat(), } # Validate API response assert api_response['success'] assert api_response['data']['new_state'] == 'ASSIGNED' assert api_response['data']['assignment_details']['assignee_id'] == 123 assert api_response['data']['assignment_details']['priority'] == 'high' except ValueError as e: # Handle Pydantic validation errors api_response = { 'success': False, 'error': 'Validation Error', 'message': str(e), 'timestamp': datetime.now().isoformat(), } except TransitionValidationError as e: # Handle business logic validation errors api_response = { 'success': False, 'error': 'Business Rule Violation', 'message': str(e), 'context': e.context, 'timestamp': datetime.now().isoformat(), } # Test error handling with invalid data invalid_request = { 'assignee_id': -1, # Invalid ID 'priority': 'invalid_priority', # Invalid priority 'deadline': '2020-01-01T00:00:00', # Past deadline } with pytest.raises(ValueError): APITaskAssignmentTransition(**invalid_request) def test_json_schema_generation_for_api_docs(self): """ API DOCUMENTATION: JSON Schema generation Shows how to generate OpenAPI/JSON schemas for API documentation from Pydantic transition models. """ @register_state_transition('annotation', 'api_submit_annotation') class APIAnnotationSubmissionTransition(BaseTransition): """Submit annotation via API with rich metadata""" confidence_score: float = Field( ..., ge=0.0, le=1.0, description="Annotator's confidence in the annotation (0.0-1.0)" ) annotation_quality: str = Field( 'good', description='Subjective quality assessment', pattern='^(excellent|good|fair|poor)$' ) time_spent_seconds: int = Field(..., ge=1, description='Time spent on annotation in seconds') difficulty_level: str = Field('medium', description='Perceived difficulty of the annotation task') review_requested: bool = Field(False, description='Whether the annotator requests manual review') tags: List[str] = Field(default_factory=list, description='Optional tags for categorization') metadata: Dict[str, Any] = Field( default_factory=dict, description='Additional metadata about the annotation process' ) def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'CREATED' def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'confidence_score': self.confidence_score, 'annotation_quality': self.annotation_quality, 'time_spent_seconds': self.time_spent_seconds, 'difficulty_level': self.difficulty_level, 'review_requested': self.review_requested, 'tags': self.tags, 'metadata': self.metadata, 'submitted_at': context.timestamp.isoformat(), } # Generate JSON schema schema = get_transition_schema(APIAnnotationSubmissionTransition) # Validate schema structure assert 'properties' in schema assert 'required' in schema # Check specific field schemas properties = schema['properties'] # confidence_score should have min/max constraints confidence_schema = properties['confidence_score'] assert confidence_schema['type'] == 'number' assert confidence_schema['minimum'] == 0.0 assert confidence_schema['maximum'] == 1.0 assert "Annotator's confidence" in confidence_schema['description'] # annotation_quality should have pattern constraint quality_schema = properties['annotation_quality'] assert quality_schema['type'] == 'string' assert 'pattern' in quality_schema # time_spent_seconds should have minimum constraint time_schema = properties['time_spent_seconds'] assert time_schema['type'] == 'integer' assert time_schema['minimum'] == 1 # tags should be array type tags_schema = properties['tags'] assert tags_schema['type'] == 'array' assert tags_schema['items']['type'] == 'string' # metadata should be object type metadata_schema = properties['metadata'] assert metadata_schema['type'] == 'object' # Required fields required_fields = schema['required'] assert 'confidence_score' in required_fields assert 'time_spent_seconds' in required_fields assert 'tags' not in required_fields # Optional field # Test schema-driven validation valid_data = { 'confidence_score': 0.85, 'annotation_quality': 'good', 'time_spent_seconds': 120, 'difficulty_level': 'hard', 'review_requested': True, 'tags': ['important', 'complex'], 'metadata': {'tool_version': '1.2.3', 'browser': 'chrome'}, } transition = APIAnnotationSubmissionTransition(**valid_data) assert transition.confidence_score == 0.85 assert len(transition.tags) == 2 # Print schema for documentation (would be used in API docs) schema_json = json.dumps(schema, indent=2) assert isinstance(schema_json, str) assert 'confidence_score' in schema_json def test_bulk_operations_api_pattern(self): """ API EXAMPLE: Bulk operations with transitions Shows how to handle bulk operations where multiple entities need to be transitioned with the same or different parameters. """ @register_state_transition('task', 'bulk_status_update') class BulkStatusUpdateTransition(BaseTransition): """Bulk status update for multiple tasks""" new_status: str = Field(..., description='New status for all tasks') update_reason: str = Field(..., description='Reason for bulk update') batch_id: str = Field(..., description='Unique identifier for this batch') force_update: bool = Field(False, description='Force update even if invalid states') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return self.new_status def validate_transition(self, context: TransitionContext) -> bool: valid_statuses = ['CREATED', 'IN_PROGRESS', 'COMPLETED', 'CANCELLED'] if self.new_status not in valid_statuses: raise TransitionValidationError(f'Invalid status: {self.new_status}') # Skip state validation if force update if not self.force_update: if context.current_state == self.new_status: raise TransitionValidationError('Cannot update to same status') return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'new_status': self.new_status, 'update_reason': self.update_reason, 'batch_id': self.batch_id, 'force_update': self.force_update, 'updated_at': context.timestamp.isoformat(), 'entity_id': context.entity.pk, } # Simulate bulk API request bulk_request = { 'task_ids': [1, 2, 3, 4, 5], 'transition_data': { 'new_status': 'IN_PROGRESS', 'update_reason': 'Project phase change', 'batch_id': 'batch_2024_001', 'force_update': False, }, } # Process bulk request batch_results = [] failed_updates = [] for task_id in bulk_request['task_ids']: # Create mock entity for each task mock_task = Mock() mock_task.pk = task_id mock_task._meta.model_name = 'task' try: # Create transition transition = BulkStatusUpdateTransition(**bulk_request['transition_data']) # Mock different current states for testing current_states = ['CREATED', 'CREATED', 'IN_PROGRESS', 'CREATED', 'COMPLETED'] current_state = current_states[task_id - 1] # Adjust for 0-based indexing context = TransitionContext( entity=mock_task, current_user=self.mock_user, current_state=current_state, target_state=transition.get_target_state(), ) # Validate and execute if transition.validate_transition(context): result = transition.transition(context) batch_results.append({'task_id': task_id, 'success': True, 'result': result}) except TransitionValidationError as e: failed_updates.append( {'task_id': task_id, 'success': False, 'error': str(e), 'context': getattr(e, 'context', {})} ) # API response for bulk operation api_response = { 'batch_id': bulk_request['transition_data']['batch_id'], 'total_requested': len(bulk_request['task_ids']), 'successful_updates': len(batch_results), 'failed_updates': len(failed_updates), 'results': batch_results, 'failures': failed_updates, 'timestamp': datetime.now().isoformat(), } # Validate bulk results assert api_response['total_requested'] == 5 assert api_response['successful_updates'] > 0 # Some tasks should succeed, some might fail due to state validation total_processed = api_response['successful_updates'] + api_response['failed_updates'] assert total_processed == 5 # Check individual results for result in batch_results: assert result['success'] assert result['result']['new_status'] == 'IN_PROGRESS' assert result['result']['batch_id'] == 'batch_2024_001' def test_webhook_integration_pattern(self): """ API EXAMPLE: Webhook integration with transitions Shows how to integrate transitions with webhook systems for external notifications and integrations. """ @register_state_transition('task', 'webhook_completion') class WebhookTaskCompletionTransition(BaseTransition): """Task completion with webhook notifications""" completion_quality: float = Field(..., ge=0.0, le=1.0) completion_notes: str = Field('', description='Completion notes') webhook_urls: List[str] = Field(default_factory=list, description='Webhook URLs to notify') notification_data: Dict[str, Any] = Field(default_factory=dict, description='Data to send in webhooks') webhook_responses: List[Dict[str, Any]] = Field( default_factory=list, description='Webhook response tracking' ) def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'COMPLETED' def validate_transition(self, context: TransitionContext) -> bool: if context.current_state != 'IN_PROGRESS': raise TransitionValidationError('Can only complete in-progress tasks') return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'completion_quality': self.completion_quality, 'completion_notes': self.completion_notes, 'webhook_urls': self.webhook_urls, 'notification_data': self.notification_data, 'completed_at': context.timestamp.isoformat(), 'completed_by_id': context.current_user.id if context.current_user else None, } def post_transition_hook(self, context: TransitionContext, state_record) -> None: """Send webhook notifications after successful transition""" if self.webhook_urls: webhook_payload = { 'event': 'task.completed', 'task_id': context.entity.pk, 'state_record_id': getattr(state_record, 'id', 'mock-id'), 'completion_data': { 'quality': self.completion_quality, 'notes': self.completion_notes, 'completed_by': context.current_user.id if context.current_user else None, 'completed_at': context.timestamp.isoformat(), }, 'custom_data': self.notification_data, 'timestamp': datetime.now().isoformat(), } # Mock webhook sending (in real implementation, use async requests) for url in self.webhook_urls: webhook_response = { 'url': url, 'payload': webhook_payload, 'status': 'sent', 'timestamp': datetime.now().isoformat(), } self.webhook_responses.append(webhook_response) # Test webhook transition transition = WebhookTaskCompletionTransition( completion_quality=0.95, completion_notes='Task completed with excellent quality', webhook_urls=[ 'https://api.example.com/webhooks/task-completed', 'https://notifications.example.com/task-events', ], notification_data={'project_id': 123, 'priority': 'high', 'client_id': 'client_456'}, ) context = TransitionContext( entity=self.mock_entity, current_user=self.mock_user, current_state='IN_PROGRESS', target_state=transition.get_target_state(), ) # Validate and execute assert transition.validate_transition(context) transition.transition(context) # Simulate state record creation mock_state_record = Mock() mock_state_record.id = 'state-uuid-123' # Execute post-hook (webhook sending) transition.post_transition_hook(context, mock_state_record) # Validate webhook responses assert len(transition.webhook_responses) == 2 for response in transition.webhook_responses: assert 'url' in response assert 'payload' in response assert response['status'] == 'sent' # Validate webhook payload structure payload = response['payload'] assert payload['event'] == 'task.completed' assert payload['task_id'] == self.mock_entity.pk assert payload['completion_data']['quality'] == 0.95 assert payload['custom_data']['project_id'] == 123 def test_api_error_handling_patterns(self): """ API EXAMPLE: Comprehensive error handling patterns Shows how to implement robust error handling for API endpoints using the transition system with proper HTTP status codes and messages. """ @register_state_transition('task', 'api_critical_update') class APICriticalUpdateTransition(BaseTransition): """Critical update with extensive validation""" update_type: str = Field(..., description='Type of critical update') severity_level: int = Field(..., ge=1, le=5, description='Severity level 1-5') authorization_token: str = Field(..., description='Authorization token for critical updates') backup_required: bool = Field(True, description='Whether backup is required before update') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'CRITICALLY_UPDATED' def validate_transition(self, context: TransitionContext) -> bool: errors = [] # Authorization check if len(self.authorization_token) < 10: errors.append('Invalid authorization token') # Severity validation if self.severity_level >= 4 and not context.current_user: errors.append('High severity updates require authenticated user') # Update type validation valid_types = ['security_patch', 'critical_fix', 'emergency_update'] if self.update_type not in valid_types: errors.append(f'Invalid update type. Must be one of: {valid_types}') # State validation if context.current_state in ['COMPLETED', 'ARCHIVED']: errors.append(f'Cannot perform critical updates on {context.current_state.lower()} tasks') # Backup requirement if self.backup_required and self.severity_level >= 3: # Mock backup check backup_exists = True # In real implementation, check backup system if not backup_exists: errors.append('Backup required but not available') if errors: raise TransitionValidationError( 'Critical update validation failed', { 'validation_errors': errors, 'error_count': len(errors), 'severity_level': self.severity_level, 'update_type': self.update_type, }, ) return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'update_type': self.update_type, 'severity_level': self.severity_level, 'backup_required': self.backup_required, 'authorized_by': context.current_user.id if context.current_user else None, 'updated_at': context.timestamp.isoformat(), 'critical_update_id': f'crit_{int(context.timestamp.timestamp())}', } # Test various error scenarios and API responses # 1. Test successful request valid_request = { 'update_type': 'security_patch', 'severity_level': 3, 'authorization_token': 'valid_token_12345', 'backup_required': True, } def simulate_api_endpoint(request_data, current_state='IN_PROGRESS'): """Simulate API endpoint with proper error handling""" try: # Parse and validate request transition = APICriticalUpdateTransition(**request_data) # Create context context = TransitionContext( entity=self.mock_entity, current_user=self.mock_user, current_state=current_state, target_state=transition.get_target_state(), ) # Validate business logic transition.validate_transition(context) # Execute transition result = transition.transition(context) return { 'status_code': 200, 'success': True, 'data': { 'task_id': self.mock_entity.pk, 'new_state': transition.get_target_state(), 'update_details': result, }, } except ValueError as e: # Pydantic validation error (400 Bad Request) return { 'status_code': 400, 'success': False, 'error': 'Bad Request', 'message': 'Invalid request data', 'details': str(e), } except TransitionValidationError as e: # Business logic validation error (422 Unprocessable Entity) return { 'status_code': 422, 'success': False, 'error': 'Validation Failed', 'message': str(e), 'validation_errors': e.context.get('validation_errors', []), 'context': e.context, } except Exception as e: # Unexpected error (500 Internal Server Error) return { 'status_code': 500, 'success': False, 'error': 'Internal Server Error', 'message': 'An unexpected error occurred', 'details': str(e) if not isinstance(e, Exception) else 'Server error', } # Test successful request response = simulate_api_endpoint(valid_request) assert response['status_code'] == 200 assert response['success'] assert 'update_details' in response['data'] # Test Pydantic validation error (invalid severity level) invalid_request = { 'update_type': 'security_patch', 'severity_level': 10, # Invalid: > 5 'authorization_token': 'valid_token_12345', } response = simulate_api_endpoint(invalid_request) assert response['status_code'] == 400 assert not response['success'] assert response['error'] == 'Bad Request' # Test business logic validation error business_logic_error_request = { 'update_type': 'invalid_type', # Invalid update type 'severity_level': 5, 'authorization_token': 'short', # Too short 'backup_required': True, } response = simulate_api_endpoint(business_logic_error_request) assert response['status_code'] == 422 assert not response['success'] assert response['error'] == 'Validation Failed' assert 'validation_errors' in response assert len(response['validation_errors']) > 0 # Test state validation error response = simulate_api_endpoint(valid_request, current_state='COMPLETED') assert response['status_code'] == 422 # The error message is in validation_errors list, not the main message validation_errors = response.get('validation_errors', []) assert any('completed tasks' in error for error in validation_errors) def test_api_versioning_and_backward_compatibility(self): """ API EXAMPLE: API versioning with backward compatibility Shows how to handle API versioning using transition inheritance and maintain backward compatibility. """ # Version 1 API @register_state_transition('task', 'update_task_v1') class UpdateTaskV1Transition(BaseTransition): """Version 1 task update API""" status: str = Field(..., description='New task status') notes: str = Field('', description='Update notes') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return self.status def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'status': self.status, 'notes': self.notes, 'api_version': 'v1', 'updated_at': context.timestamp.isoformat(), } # Version 2 API with additional features @register_state_transition('task', 'update_task_v2') class UpdateTaskV2Transition(UpdateTaskV1Transition): """Version 2 task update API with enhanced features""" priority: Optional[str] = Field(None, description='Task priority') tags: List[str] = Field(default_factory=list, description='Task tags') estimated_hours: Optional[float] = Field(None, ge=0, description='Estimated hours') metadata: Dict[str, Any] = Field(default_factory=dict, description='Additional metadata') def transition(self, context: TransitionContext) -> Dict[str, Any]: # Call parent method for base functionality base_data = super().transition(context) # Add V2 specific data v2_data = { 'priority': self.priority, 'tags': self.tags, 'estimated_hours': self.estimated_hours, 'metadata': self.metadata, 'api_version': 'v2', } return {**base_data, **v2_data} # Test V1 API (backward compatibility) v1_request = {'status': 'IN_PROGRESS', 'notes': 'Started working on task'} v1_transition = UpdateTaskV1Transition(**v1_request) context = TransitionContext( entity=self.mock_entity, current_state='CREATED', target_state=v1_transition.get_target_state() ) v1_result = v1_transition.transition(context) assert v1_result['api_version'] == 'v1' assert v1_result['status'] == 'IN_PROGRESS' assert 'priority' not in v1_result # V1 doesn't have priority # Test V2 API with enhanced features v2_request = { 'status': 'IN_PROGRESS', 'notes': 'Started working on task with enhanced tracking', 'priority': 'high', 'tags': ['urgent', 'client-facing'], 'estimated_hours': 4.5, 'metadata': {'client_id': 123, 'project_phase': 'development'}, } v2_transition = UpdateTaskV2Transition(**v2_request) v2_result = v2_transition.transition(context) assert v2_result['api_version'] == 'v2' assert v2_result['status'] == 'IN_PROGRESS' # Inherited from V1 assert v2_result['priority'] == 'high' # V2 feature assert len(v2_result['tags']) == 2 # V2 feature assert v2_result['estimated_hours'] == 4.5 # V2 feature assert 'client_id' in v2_result['metadata'] # V2 feature # Test V2 API with minimal data (backward compatible) v2_minimal_request = {'status': 'COMPLETED', 'notes': 'Task finished'} v2_minimal_transition = UpdateTaskV2Transition(**v2_minimal_request) v2_minimal_result = v2_minimal_transition.transition(context) assert v2_minimal_result['api_version'] == 'v2' assert v2_minimal_result['status'] == 'COMPLETED' assert v2_minimal_result['priority'] is None # Optional field assert v2_minimal_result['tags'] == [] # Default value assert v2_minimal_result['estimated_hours'] is None # Optional field assert v2_minimal_result['metadata'] == {} # Default value