""" Comprehensive tests for the declarative Pydantic-based transition system. This test suite provides extensive coverage of the new transition system, including usage examples, edge cases, validation scenarios, and integration patterns to serve as both tests and documentation. """ from datetime import datetime, timedelta from typing import Any, Dict, Optional from unittest.mock import Mock, patch import pytest from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone from fsm.registry import register_state_transition, transition_registry from fsm.transition_utils import get_available_transitions from fsm.transitions import ( BaseTransition, TransitionContext, TransitionValidationError, ) from pydantic import Field, ValidationError User = get_user_model() # Test state choices for testing class TestStateChoices: CREATED = 'CREATED' IN_PROGRESS = 'IN_PROGRESS' COMPLETED = 'COMPLETED' class MockEntity: """Mock entity model for testing""" def __init__(self, pk=1): self.pk = pk self.id = pk self.organization_id = 1 self._meta = Mock() self._meta.model_name = 'test_entity' self._meta.label_lower = 'test.testentity' class MockTask: """Mock task model for testing""" def __init__(self, pk=1): self.pk = pk self.id = pk self.organization_id = 1 self._meta = Mock() self._meta.model_name = 'task' self._meta.label_lower = 'tasks.task' class MockAnnotation: """Mock annotation model for testing""" def __init__(self, pk=1): self.pk = pk self.id = pk self.result = {'test': 'data'} # Mock annotation data self.organization_id = 1 self._meta = Mock() self._meta.model_name = 'annotation' self._meta.label_lower = 'tasks.annotation' class CoreFrameworkTests(TestCase): """Test core framework functionality""" def setUp(self): """Set up test data""" self.user = User.objects.create_user(email='test@example.com', password='test123') self.mock_entity = MockEntity() def test_base_transition_class(self): """Test BaseTransition abstract functionality""" @register_state_transition('test_entity', 'test_transition') class TestTransition(BaseTransition): test_field: str = Field('default', description='Test field') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.IN_PROGRESS def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'test_field': self.test_field} # Test instantiation transition = TestTransition(test_field='test_value') assert transition.test_field == 'test_value' assert transition.get_target_state() == TestStateChoices.IN_PROGRESS assert transition.transition_name == 'test_transition' def test_transition_context(self): """Test TransitionContext functionality""" context = TransitionContext( entity=self.mock_entity, current_user=self.user, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, organization_id=1, ) assert context.entity == self.mock_entity assert context.current_state == TestStateChoices.CREATED assert context.target_state == TestStateChoices.IN_PROGRESS assert context.current_user == self.user assert context.has_current_state assert not context.is_initial_transition def test_transition_context_properties(self): """Test TransitionContext computed properties""" # Test initial transition context = TransitionContext(entity=self.mock_entity, current_state=None, target_state=TestStateChoices.CREATED) assert context.is_initial_transition assert not context.has_current_state # Test with current state context_with_state = TransitionContext( entity=self.mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, ) assert not context_with_state.is_initial_transition assert context_with_state.has_current_state def test_transition_registry(self): """Test transition registration and retrieval""" @register_state_transition('test_entity', 'test_transition') class TestTransition(BaseTransition): def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def transition(self, context: TransitionContext) -> Dict[str, Any]: return {} # Test registration retrieved = transition_registry.get_transition('test_entity', 'test_transition') assert retrieved == TestTransition # Test entity transitions entity_transitions = transition_registry.get_transitions_for_entity('test_entity') assert 'test_transition' in entity_transitions assert entity_transitions['test_transition'] == TestTransition def test_pydantic_validation(self): """Test Pydantic validation in transitions""" @register_state_transition('test_entity', 'validated_transition') class ValidatedTransition(BaseTransition): required_field: str = Field(..., description='Required field') optional_field: int = Field(42, description='Optional field') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'required_field': self.required_field, 'optional_field': self.optional_field} # Test valid instantiation transition = ValidatedTransition(required_field='test') assert transition.required_field == 'test' assert transition.optional_field == 42 # Test validation error with pytest.raises(ValidationError): ValidatedTransition() # Missing required field def test_transition_execution(self): """Test transition execution logic""" @register_state_transition('test_entity', 'execution_test') class ExecutionTestTransition(BaseTransition): value: str = Field('test', description='Test value') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def validate_transition(self, context: TransitionContext) -> bool: return context.current_state == TestStateChoices.IN_PROGRESS def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'value': self.value, 'entity_id': context.entity.pk, 'timestamp': context.timestamp.isoformat(), } transition = ExecutionTestTransition(value='execution_test') context = TransitionContext( entity=self.mock_entity, current_state=TestStateChoices.IN_PROGRESS, target_state=transition.get_target_state(), timestamp=datetime.now(), ) # Test validation assert transition.validate_transition(context) # Test execution result = transition.transition(context) assert result['value'] == 'execution_test' assert result['entity_id'] == self.mock_entity.pk assert 'timestamp' in result def test_validation_error_handling(self): """Test transition validation error handling""" @register_state_transition('test_entity', 'validation_test') class ValidationTestTransition(BaseTransition): def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def validate_transition(self, context: TransitionContext) -> bool: if context.current_state != TestStateChoices.IN_PROGRESS: raise TransitionValidationError( 'Can only complete from IN_PROGRESS state', {'current_state': context.current_state} ) return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return {} transition = ValidationTestTransition() # Test valid transition valid_context = TransitionContext( entity=self.mock_entity, current_state=TestStateChoices.IN_PROGRESS, target_state=transition.get_target_state(), ) assert transition.validate_transition(valid_context) # Test invalid transition invalid_context = TransitionContext( entity=self.mock_entity, current_state=TestStateChoices.CREATED, target_state=transition.get_target_state(), ) # Test validation error with pytest.raises(TransitionValidationError) as cm: transition.validate_transition(invalid_context) error = cm.value assert 'Can only complete from IN_PROGRESS state' in str(error) assert 'current_state' in error.context def test_state_manager_transition_execution(self): """Test StateManager-based transition execution""" @register_state_transition('test_entity', 'state_manager_test') class StateManagerTestTransition(BaseTransition): value: str = Field('default', description='Test value') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'value': self.value} # Test StateManager execution using the registry directly (simpler test) # This validates that the consolidated approach works through the registry from fsm.registry import transition_registry # Get the transition class transition_class = transition_registry.get_transition('test_entity', 'state_manager_test') assert transition_class is not None # Create instance and verify it works transition = transition_class(value='state_manager_test_value') assert transition.value == 'state_manager_test_value' assert transition.get_target_state() == TestStateChoices.COMPLETED def test_transition_hooks(self): """Test transition lifecycle hooks""" hook_calls = [] @register_state_transition('test_entity', 'hook_test') class HookTestTransition(BaseTransition): def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def pre_transition_hook(self, context: TransitionContext) -> None: hook_calls.append('pre') def transition(self, context: TransitionContext) -> Dict[str, Any]: hook_calls.append('transition') return {} def post_transition_hook(self, context: TransitionContext, state_record) -> None: hook_calls.append('post') transition = HookTestTransition() context = TransitionContext( entity=self.mock_entity, current_state=TestStateChoices.IN_PROGRESS, target_state=transition.get_target_state(), ) # Test hook execution order transition.pre_transition_hook(context) transition.transition(context) transition.post_transition_hook(context, Mock()) assert hook_calls == ['pre', 'transition', 'post'] class TransitionUtilsTests(TestCase): """Test cases for transition utility functions""" def setUp(self): self.user = User.objects.create_user(email='test@example.com', password='test123') self.mock_entity = MockEntity() def test_get_available_transitions(self): """Test get_available_transitions utility""" @register_state_transition('test_entity', 'available_test') class AvailableTestTransition(BaseTransition): def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def transition(self, context: TransitionContext) -> Dict[str, Any]: return {} available = get_available_transitions(self.mock_entity) assert 'available_test' in available assert available['available_test'] == AvailableTestTransition # Test with non-existent entity mock_other = MockEntity() mock_other._meta.model_name = 'other_entity' other_available = get_available_transitions(mock_other) assert len(other_available) == 0 def test_get_available_transitions_with_validation(self): """Test the validation behavior of get_available_transitions""" from fsm.state_manager import StateManager @register_state_transition('test_entity', 'validation_test_1') class ValidationTestTransition1(BaseTransition): def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.IN_PROGRESS @classmethod def can_transition_from_state(cls, context) -> bool: # Only allow from CREATED state return context.current_state == TestStateChoices.CREATED def transition(self, context: TransitionContext) -> Dict[str, Any]: return {} @register_state_transition('test_entity', 'validation_test_2') class ValidationTestTransition2(BaseTransition): def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED @classmethod def can_transition_from_state(cls, context) -> bool: # Only allow from IN_PROGRESS state return context.current_state == TestStateChoices.IN_PROGRESS def transition(self, context: TransitionContext) -> Dict[str, Any]: return {} # Test validate=False (should return all registered transitions) all_available = get_available_transitions(self.mock_entity, validate=False) assert len(all_available) == 2 assert 'validation_test_1' in all_available assert 'validation_test_2' in all_available # Mock current state as CREATED mock_state_object = Mock() mock_state_object.state = TestStateChoices.CREATED with patch.object(StateManager, 'get_current_state_object', return_value=mock_state_object): # Test validate=True with CREATED state (should only return validation_test_1) valid_transitions = get_available_transitions(self.mock_entity, validate=True) assert len(valid_transitions) == 1 assert 'validation_test_1' in valid_transitions assert 'validation_test_2' not in valid_transitions # Mock current state as IN_PROGRESS mock_state_object.state = TestStateChoices.IN_PROGRESS with patch.object(StateManager, 'get_current_state_object', return_value=mock_state_object): # Test validate=True with IN_PROGRESS state (should only return validation_test_2) valid_transitions = get_available_transitions(self.mock_entity, validate=True) assert len(valid_transitions) == 1 assert 'validation_test_2' in valid_transitions assert 'validation_test_1' not in valid_transitions # Mock current state as COMPLETED mock_state_object.state = TestStateChoices.COMPLETED with patch.object(StateManager, 'get_current_state_object', return_value=mock_state_object): # Test validate=True with COMPLETED state (should return no transitions) valid_transitions = get_available_transitions(self.mock_entity, validate=True) assert len(valid_transitions) == 0 def test_get_available_transitions_with_required_fields(self): """Test that transitions with required fields are handled correctly during validation""" from fsm.state_manager import StateManager @register_state_transition('test_entity', 'required_field_transition') class RequiredFieldTransition(BaseTransition): required_field: str = Field(..., description='This field is required') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.IN_PROGRESS @classmethod def can_transition_from_state(cls, context) -> bool: # This should never be called since we can't instantiate without required_field return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'required_field': self.required_field} # Test validate=False (should return the transition even though it has required fields) all_available = get_available_transitions(self.mock_entity, validate=False) assert 'required_field_transition' in all_available # Mock current state mock_state_object = Mock() mock_state_object.state = TestStateChoices.CREATED with patch.object(StateManager, 'get_current_state_object', return_value=mock_state_object): # Test validate=True - should include transitions that can't be instantiated for validation # This is the behavior: we can't validate transitions with required fields # without knowing what data will be provided, so we include them as "available" valid_transitions = get_available_transitions(self.mock_entity, validate=True) # The transition should be included since we can't validate it (better to be permissive) # This avoids false negatives where valid transitions appear unavailable due to # validation limitations assert 'required_field_transition' in valid_transitions class ComprehensiveUsageExampleTests(TestCase): """ Comprehensive test cases demonstrating various usage patterns. These tests serve as both validation and documentation for how to implement and use the declarative transition system. """ def setUp(self): self.task = MockTask() self.user = Mock() self.user.id = 123 self.user.username = 'testuser' def test_basic_transition_implementation(self): """ USAGE EXAMPLE: Basic transition implementation Shows how to implement a simple transition with validation. """ class BasicTransition(BaseTransition): """Example: Simple transition with required field""" message: str = Field(..., description='Message for the transition') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'PROCESSED' def validate_transition(self, context: TransitionContext) -> bool: # Business logic validation if context.current_state == 'COMPLETED': raise TransitionValidationError('Cannot process completed items') return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'message': self.message, 'processed_by': context.current_user.username if context.current_user else 'system', 'processed_at': context.timestamp.isoformat(), } # Test the implementation transition = BasicTransition(message='Processing task') assert transition.message == 'Processing task' assert transition.get_target_state() == 'PROCESSED' # Test validation context = TransitionContext( entity=self.task, current_user=self.user, current_state='CREATED', target_state=transition.get_target_state(), ) assert transition.validate_transition(context) # Test data generation data = transition.transition(context) assert data['message'] == 'Processing task' assert data['processed_by'] == 'testuser' assert 'processed_at' in data def test_complex_validation_example(self): """ USAGE EXAMPLE: Complex validation with multiple conditions Shows how to implement sophisticated business logic validation. """ class TaskAssignmentTransition(BaseTransition): """Example: Complex validation for task assignment""" assignee_id: int = Field(..., description='User to assign task to') priority: str = Field('normal', description='Task priority') deadline: datetime = Field(None, description='Task deadline') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'ASSIGNED' def validate_transition(self, context: TransitionContext) -> bool: # Multiple validation conditions if context.current_state not in ['CREATED', 'UNASSIGNED']: raise TransitionValidationError( f'Cannot assign task in state {context.current_state}', {'current_state': context.current_state, 'task_id': context.entity.pk}, ) # Check deadline is in future if self.deadline and self.deadline <= timezone.now(): raise TransitionValidationError( 'Deadline must be in the future', {'deadline': self.deadline.isoformat()} ) # Check priority is valid valid_priorities = ['low', 'normal', 'high', 'urgent'] if self.priority not in valid_priorities: raise TransitionValidationError( f'Invalid priority: {self.priority}', {'valid_priorities': valid_priorities} ) 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, 'assigned_by': context.current_user.id if context.current_user else None, 'assignment_reason': f'Task assigned to user {self.assignee_id}', } # Test valid assignment future_deadline = timezone.now() + timedelta(days=7) transition = TaskAssignmentTransition(assignee_id=456, priority='high', deadline=future_deadline) context = TransitionContext( entity=self.task, current_user=self.user, current_state='CREATED', target_state=transition.get_target_state(), ) assert transition.validate_transition(context) # Test invalid state context.current_state = 'COMPLETED' with pytest.raises(TransitionValidationError) as cm: transition.validate_transition(context) assert 'Cannot assign task in state' in str(cm.value) assert 'COMPLETED' in str(cm.value) # Test invalid deadline past_deadline = timezone.now() - timedelta(days=1) invalid_transition = TaskAssignmentTransition(assignee_id=456, deadline=past_deadline) context.current_state = 'CREATED' with pytest.raises(TransitionValidationError) as cm: invalid_transition.validate_transition(context) assert 'Deadline must be in the future' in str(cm.value) def test_registry_and_decorator_usage(self): """ USAGE EXAMPLE: Using the registry and decorator system Shows how to register transitions and use the decorator syntax. """ @register_state_transition('document', 'publish') class PublishDocumentTransition(BaseTransition): """Example: Using the registration decorator""" publish_immediately: bool = Field(True, description='Publish immediately') scheduled_time: datetime = Field(None, description='Scheduled publish time') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'PUBLISHED' if self.publish_immediately else 'SCHEDULED' def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'publish_immediately': self.publish_immediately, 'scheduled_time': self.scheduled_time.isoformat() if self.scheduled_time else None, 'published_by': context.current_user.id if context.current_user else None, } # Test registration worked registered_class = transition_registry.get_transition('document', 'publish') assert registered_class == PublishDocumentTransition # Test getting transitions for entity document_transitions = transition_registry.get_transitions_for_entity('document') assert 'publish' in document_transitions # Test execution through registry mock_document = Mock() mock_document.pk = 1 mock_document._meta.model_name = 'document' # This would normally go through the full execution workflow transition_data = {'publish_immediately': False, 'scheduled_time': timezone.now() + timedelta(hours=2)} # Test transition creation and validation transition = PublishDocumentTransition(**transition_data) assert transition.get_target_state() == 'SCHEDULED' class ValidationAndErrorHandlingTests(TestCase): """ Tests focused on validation scenarios and error handling. These tests demonstrate proper error handling patterns and validation edge cases. """ def setUp(self): self.task = MockTask() def test_pydantic_validation_errors(self): """Test Pydantic field validation errors""" class StrictValidationTransition(BaseTransition): required_field: str = Field(..., description='Required field') email_field: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$', description='Valid email') number_field: int = Field(..., ge=1, le=100, description='Number between 1-100') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'VALIDATED' @classmethod def get_target_state(cls) -> str: return 'VALIDATED' @classmethod def can_transition_from_state(cls, context: TransitionContext) -> bool: return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'validated': True} # Test missing required field with pytest.raises(ValidationError): StrictValidationTransition(email_field='test@example.com', number_field=50) # Test invalid email with pytest.raises(ValidationError): StrictValidationTransition(required_field='test', email_field='invalid-email', number_field=50) # Test number out of range with pytest.raises(ValidationError): StrictValidationTransition(required_field='test', email_field='test@example.com', number_field=150) # Test valid data valid_transition = StrictValidationTransition( required_field='test', email_field='user@example.com', number_field=75 ) assert valid_transition.required_field == 'test' def test_business_logic_validation_errors(self): """Test business logic validation with detailed error context""" class BusinessRuleTransition(BaseTransition): amount: float = Field(..., description='Transaction amount') currency: str = Field('USD', description='Currency code') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'PROCESSED' def validate_transition(self, context: TransitionContext) -> bool: # Complex business rule validation errors = [] if self.amount <= 0: errors.append('Amount must be positive') if self.amount > 10000 and context.current_user is None: errors.append('Large amounts require authenticated user') if self.currency not in ['USD', 'EUR', 'GBP']: errors.append(f'Unsupported currency: {self.currency}') if context.current_state == 'CANCELLED': errors.append('Cannot process cancelled transactions') if errors: raise TransitionValidationError( f"Validation failed: {'; '.join(errors)}", { 'validation_errors': errors, 'amount': self.amount, 'currency': self.currency, 'current_state': context.current_state, }, ) return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'amount': self.amount, 'currency': self.currency} context = TransitionContext(entity=self.task, current_state='PENDING', target_state='PROCESSED') # Test negative amount negative_transition = BusinessRuleTransition(amount=-100) with pytest.raises(TransitionValidationError) as cm: negative_transition.validate_transition(context) error = cm.value assert 'Amount must be positive' in str(error) assert 'validation_errors' in error.context # Test large amount without user large_transition = BusinessRuleTransition(amount=15000) with pytest.raises(TransitionValidationError) as cm: large_transition.validate_transition(context) assert 'Large amounts require authenticated user' in str(cm.value) # Test invalid currency invalid_currency_transition = BusinessRuleTransition(amount=100, currency='XYZ') with pytest.raises(TransitionValidationError) as cm: invalid_currency_transition.validate_transition(context) assert 'Unsupported currency' in str(cm.value) # Test multiple errors multi_error_transition = BusinessRuleTransition(amount=-50, currency='XYZ') with pytest.raises(TransitionValidationError) as cm: multi_error_transition.validate_transition(context) error_msg = str(cm.value) assert 'Amount must be positive' in error_msg assert 'Unsupported currency' in error_msg # Pytest-style tests for compatibility @pytest.fixture def task(): """Pytest fixture for mock task""" return MockTask() @pytest.fixture def user(): """Pytest fixture for mock user""" user = Mock() user.id = 1 user.username = 'testuser' return user def test_transition_context_properties(task, user): """Test TransitionContext properties using pytest""" context = TransitionContext(entity=task, current_user=user, current_state='CREATED', target_state='IN_PROGRESS') assert context.has_current_state assert not context.is_initial_transition assert context.current_state == 'CREATED' assert context.target_state == 'IN_PROGRESS' def test_pydantic_validation(): """Test Pydantic validation in transitions""" class SampleTransition(BaseTransition): test_field: str optional_field: int = 42 def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return 'TEST_STATE' def transition(self, context: TransitionContext) -> dict: return {'test_field': self.test_field} # Valid data transition = SampleTransition(test_field='valid') assert transition.test_field == 'valid' assert transition.optional_field == 42 # Invalid data should raise validation error with pytest.raises(ValidationError): # Pydantic validation error SampleTransition() # Missing required field def test_side_effect_only_transition(): """Test side-effect only transitions (target_state=None)""" class SideEffectTransition(BaseTransition): action_performed: str = 'notification_sent' def get_target_state(self, context: Optional[TransitionContext] = None) -> Optional[str]: # Return None to indicate no state change, only side effects return None def transition(self, context: TransitionContext) -> dict: # Perform side effect (e.g., send notification, log event) return {'action': self.action_performed} def post_transition_hook(self, context: TransitionContext, state_record) -> None: # State record should be None for side-effect only transitions assert state_record is None # Test instantiation transition = SideEffectTransition(action_performed='email_sent') assert transition.get_target_state() is None # Test context creation with None target_state mock_entity = type('MockEntity', (), {'pk': 1, '_meta': type('Meta', (), {'model_name': 'test'})})() context = TransitionContext( entity=mock_entity, target_state=None, # Should be allowed for side-effect only transitions ) assert context.target_state is None # Test transition execution result = transition.transition(context) assert result['action'] == 'email_sent' def test_skip_validation_flag(): """ Test the skip_validation flag in TransitionContext. This test validates step by step: - Creating a transition with validation logic that would normally fail - Verifying that validation runs and fails when skip_validation=False (default) - Verifying that validation is skipped when skip_validation=True - Confirming that prepare_and_validate respects the skip_validation flag - Ensuring the transition can execute successfully when validation is skipped Critical validation: The skip_validation flag provides a mechanism to bypass validation checks for special cases like system migrations, data imports, or administrative operations that need to override normal business rules. """ class StrictValidationTransition(BaseTransition): """Test transition with strict validation rules""" action: str = Field(..., description='Action to perform') def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.COMPLETED def validate_transition(self, context: TransitionContext) -> bool: """Validation that only allows transition from IN_PROGRESS state""" if context.current_state != TestStateChoices.IN_PROGRESS: raise TransitionValidationError( f'Can only complete from IN_PROGRESS state, not {context.current_state}', {'current_state': context.current_state, 'target_state': context.target_state}, ) return True def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'action': self.action, 'completed': True} # Create mock entity mock_entity = MockEntity() # Test 1: Normal validation (skip_validation=False, default behavior) # This should fail because current_state is CREATED, not IN_PROGRESS transition = StrictValidationTransition(action='test_action') context_with_validation = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, # Invalid state for this transition target_state=transition.get_target_state(), skip_validation=False, # Explicit False (same as default) ) # Verify that validation fails as expected with pytest.raises(TransitionValidationError) as cm: transition.prepare_and_validate(context_with_validation) error = cm.value assert 'Can only complete from IN_PROGRESS state' in str(error) assert error.context['current_state'] == TestStateChoices.CREATED # Test 2: Skip validation (skip_validation=True) # This should succeed even though current_state is CREATED transition_skip = StrictValidationTransition(action='skip_validation_action') context_skip_validation = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, # Same invalid state target_state=transition_skip.get_target_state(), skip_validation=True, # Validation should be skipped ) # This should NOT raise an error because validation is skipped result = transition_skip.prepare_and_validate(context_skip_validation) # Verify the transition executed successfully assert result['action'] == 'skip_validation_action' assert result['completed'] is True # Test 3: Verify default behavior (skip_validation not specified, defaults to False) transition_default = StrictValidationTransition(action='default_action') context_default = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=transition_default.get_target_state(), # skip_validation not specified, should default to False ) # Verify default is False (validation should run and fail) assert context_default.skip_validation is False with pytest.raises(TransitionValidationError) as cm: transition_default.prepare_and_validate(context_default) assert 'Can only complete from IN_PROGRESS state' in str(cm.value) # Test 4: Verify that with correct state and skip_validation=False, it succeeds transition_valid = StrictValidationTransition(action='valid_action') context_valid = TransitionContext( entity=mock_entity, current_state=TestStateChoices.IN_PROGRESS, # Correct state target_state=transition_valid.get_target_state(), skip_validation=False, ) # This should succeed because state is valid result_valid = transition_valid.prepare_and_validate(context_valid) assert result_valid['action'] == 'valid_action' assert result_valid['completed'] is True def test_transition_context_reason_field(): """ Test the reason field in TransitionContext. This test validates step by step: - Creating a TransitionContext with no reason (defaults to None) - Creating a TransitionContext with a custom reason - Verifying the reason is accessible on the context Critical validation: The reason field allows callers to provide context-specific reasons for transitions (e.g., "Bulk import completed") that override the transition's default get_reason() method. """ mock_entity = MockEntity() # Test 1: Default reason is None context_default = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, ) assert context_default.reason is None # Test 2: Custom reason can be set custom_reason = 'Bulk import completed - 100 tasks added' context_with_reason = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, reason=custom_reason, ) assert context_with_reason.reason == custom_reason def test_transition_context_context_data_field(): """ Test the context_data field in TransitionContext. This test validates step by step: - Creating a TransitionContext with no context_data (defaults to empty dict) - Creating a TransitionContext with custom context_data - Verifying the context_data is accessible and correct Critical validation: The context_data field allows callers to add additional data to be stored in the state record's JSONB context_data field for historical tracking and auditing purposes. """ mock_entity = MockEntity() # Test 1: Default context_data is empty dict context_default = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, ) assert context_default.context_data == {} # Test 2: Custom context_data can be set custom_context_data = { 'import_source': 'cloud_storage', 'import_id': 123, 'task_count': 100, 'triggered_by': 'api', 'batch_id': 456, 'is_automatic': False, } context_with_data = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, context_data=custom_context_data, ) assert context_with_data.context_data == custom_context_data assert context_with_data.context_data['import_id'] == 123 assert context_with_data.context_data['is_automatic'] is False def test_transition_reason_override(): """ Test that context.reason overrides transition.get_reason() in executor. This test validates step by step: - Creating a transition with a custom get_reason() implementation - Executing with no context.reason (should use transition's get_reason) - Executing with context.reason set (should override get_reason) Critical validation: The reason override mechanism allows for context-specific reasons without modifying transition classes. """ class CustomReasonTransition(BaseTransition): """Transition with custom get_reason""" def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.IN_PROGRESS def get_reason(self, context: TransitionContext) -> str: return 'Default transition reason' def transition(self, context: TransitionContext) -> Dict[str, Any]: return {'executed': True} mock_entity = MockEntity() transition = CustomReasonTransition() # Test 1: Without reason override - uses transition's get_reason context_no_override = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, ) default_reason = transition.get_reason(context_no_override) assert default_reason == 'Default transition reason' # Verify context.reason is None assert context_no_override.reason is None # The executor would use: context.reason if context.reason else transition.get_reason(context) effective_reason = ( context_no_override.reason if context_no_override.reason else transition.get_reason(context_no_override) ) assert effective_reason == 'Default transition reason' # Test 2: With reason override - uses context.reason custom_reason = 'Bulk import completed with 500 tasks' context_with_override = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, reason=custom_reason, ) # The executor would use: context.reason if context.reason else transition.get_reason(context) effective_reason = ( context_with_override.reason if context_with_override.reason else transition.get_reason(context_with_override) ) assert effective_reason == custom_reason def test_transition_context_data_merge(): """ Test that context.context_data is merged with transition output. This test validates step by step: - Creating a transition that returns context data - Adding additional context_data via TransitionContext - Verifying both are merged correctly Critical validation: Additional context_data from TransitionContext should be merged with the data returned by transition.transition() method. """ class DataProducingTransition(BaseTransition): """Transition that produces context data""" action: str = 'test_action' def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.IN_PROGRESS def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'action': self.action, 'timestamp': context.timestamp.isoformat(), } mock_entity = MockEntity() transition = DataProducingTransition(action='bulk_import') # Create context with additional context_data additional_context_data = { 'import_source_id': 123, 'task_count': 456, } context = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, context_data=additional_context_data, ) # Get transition output transition_output = transition.transition(context) # Simulate merge (as done in transition_executor.py) merged_data = {**transition_output, **context.context_data} # Verify both transition output and additional context_data are present assert merged_data['action'] == 'bulk_import' assert 'timestamp' in merged_data assert merged_data['import_source_id'] == 123 assert merged_data['task_count'] == 456 def test_transition_context_data_override(): """ Test that context.context_data can override transition output keys. This test validates step by step: - Creating a transition that returns a key - Adding the same key in context_data with different value - Verifying context_data wins (as it's merged second) Critical validation: When the same key exists in both transition output and context_data, the context_data value should win since it represents caller-provided context that should take precedence. """ class OverrideTestTransition(BaseTransition): """Transition that produces a 'reason_code' key""" def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TestStateChoices.IN_PROGRESS def transition(self, context: TransitionContext) -> Dict[str, Any]: return { 'reason_code': 'default_reason', 'other_data': 'preserved', } mock_entity = MockEntity() transition = OverrideTestTransition() # Create context with context_data that overrides 'reason_code' context = TransitionContext( entity=mock_entity, current_state=TestStateChoices.CREATED, target_state=TestStateChoices.IN_PROGRESS, context_data={'reason_code': 'custom_reason'}, ) # Get transition output transition_output = transition.transition(context) # Simulate merge (as done in transition_executor.py) merged_data = {**transition_output, **context.context_data} # Verify context_data wins for 'reason_code' assert merged_data['reason_code'] == 'custom_reason' # Other data should be preserved assert merged_data['other_data'] == 'preserved'