""" Tests for django_cronjob_utils hooks integration. Tests cover standard behavior, edge cases, and outlier scenarios. """ from datetime import date from unittest import skipIf from unittest.mock import patch from django.test import TestCase from django_cronjob_utils.base import CronTask, ExecutionResult from django_cronjob_utils.hooks import ( CronjobHookContext, get_global_cronjob_hook_manager, clear_cronjob_hooks, HOOKS_AVAILABLE, ) from django_cronjob_utils.models import CronExecution # Import hook-specific classes only if available if HOOKS_AVAILABLE: from django_package_hooks import HookType, HookRejectionError else: # Create stubs for when hooks are not available class HookType: PRE = 'pre' POST = 'post' class HookRejectionError(Exception): def __init__(self, error_code, message, details=None): self.error_code = error_code self.message = message self.details = details or {} super().__init__(message) class TestTask(CronTask): """Test task for hook tests.""" task_name = 'test-task' task_code = 'T001' def execute(self, date: date) -> dict: return {'error': False, 'message': 'Success'} class FailingTask(CronTask): """Task that always fails.""" task_name = 'failing-task' task_code = 'T002' def execute(self, date: date) -> dict: return {'error': True, 'message': 'Task failed', 'error_code': 'FAIL'} class ExceptionTask(CronTask): """Task that raises exception.""" task_name = 'exception-task' task_code = 'T003' def execute(self, date: date) -> dict: raise ValueError("Test exception") # ============================================================================ # STANDARD TESTS - Normal hook behavior # ============================================================================ @skipIf(not HOOKS_AVAILABLE, "django-package-hooks not installed") class HookContextCreationTests(TestCase): """Test CronjobHookContext creation and properties.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) clear_cronjob_hooks() def tearDown(self): """Clean up after test.""" clear_cronjob_hooks() super().tearDown() def test_context_creation_with_all_fields(self): """Test creating context with all fields.""" context = CronjobHookContext( operation='execute', task_name='test-task', task_code='T001', execution_date=self.execution_date, execution_id=123, options={'force': True}, metadata={'custom': 'data'} ) self.assertEqual(context.task_name, 'test-task') self.assertEqual(context.task_code, 'T001') self.assertEqual(context.execution_date, self.execution_date) self.assertEqual(context.execution_id, 123) self.assertEqual(context.options['force'], True) self.assertEqual(context.metadata['custom'], 'data') def test_context_creation_minimal(self): """Test creating context with minimal required fields.""" context = CronjobHookContext( operation='execute', task_name='test-task', task_code='T001', execution_date=self.execution_date ) self.assertEqual(context.task_name, 'test-task') self.assertEqual(context.task_code, 'T001') self.assertEqual(context.execution_date, self.execution_date) self.assertIsNone(context.execution_id) self.assertEqual(context.options, {}) self.assertEqual(context.metadata, {}) def test_context_is_frozen(self): """Test that context is immutable (frozen dataclass).""" context = CronjobHookContext( operation='execute', task_name='test-task', task_code='T001', execution_date=self.execution_date ) with self.assertRaises(Exception): # FrozenInstanceError context.task_name = 'modified' def test_context_metadata_is_mutable(self): """Test that metadata dict is mutable for hook communication.""" context = CronjobHookContext( operation='execute', task_name='test-task', task_code='T001', execution_date=self.execution_date, metadata={} ) # Should be able to modify metadata context.metadata['new_key'] = 'new_value' self.assertEqual(context.metadata['new_key'], 'new_value') @skipIf(not HOOKS_AVAILABLE, "django-package-hooks not installed") class PreHookStandardTests(TestCase): """Test standard PRE hook behavior.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) clear_cronjob_hooks() CronExecution.objects.all().delete() def tearDown(self): """Clean up after test.""" clear_cronjob_hooks() super().tearDown() def test_pre_hook_allows_execution(self): """Test PRE hook that allows execution.""" hook_called = [] def allow_hook(context): hook_called.append(True) self.assertIsNone(context.execution_id) # No ID yet # Don't raise HookRejectionError - allow execution manager = get_global_cronjob_hook_manager() manager.registry.register('allow_hook', HookType.PRE, allow_hook, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() self.assertTrue(hook_called) self.assertTrue(result.success) self.assertEqual(result.message, 'Success') def test_pre_hook_rejects_execution(self): """Test PRE hook that rejects execution.""" def reject_hook(context): raise HookRejectionError( error_code='DEPENDENCY_NOT_MET', message='Required task not completed' ) manager = get_global_cronjob_hook_manager() manager.registry.register('reject_hook', HookType.PRE, reject_hook, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) self.assertEqual(result.error_code, 'DEPENDENCY_NOT_MET') self.assertEqual(result.message, 'Required task not completed') # No execution record should be created self.assertEqual(CronExecution.objects.count(), 0) def test_pre_hook_receives_options(self): """Test PRE hook receives task options.""" captured_context = [] def capture_hook(context): captured_context.append(context) manager = get_global_cronjob_hook_manager() manager.registry.register('capture_hook', HookType.PRE, capture_hook, operation='execute') task = TestTask(execution_date=self.execution_date, force=True, rerun=True) task.run() self.assertEqual(len(captured_context), 1) context = captured_context[0] self.assertTrue(context.options.get('force')) self.assertTrue(context.options.get('rerun')) def test_multiple_pre_hooks_priority_order(self): """Test multiple PRE hooks execute in priority order.""" execution_order = [] def hook_priority_10(context): execution_order.append('hook_10') def hook_priority_5(context): execution_order.append('hook_5') def hook_priority_1(context): execution_order.append('hook_1') manager = get_global_cronjob_hook_manager() manager.registry.register('hook_priority_10', HookType.PRE, hook_priority_10, operation='execute', priority=10) manager.registry.register('hook_priority_5', HookType.PRE, hook_priority_5, operation='execute', priority=5) manager.registry.register('hook_priority_1', HookType.PRE, hook_priority_1, operation='execute', priority=1) task = TestTask(execution_date=self.execution_date) task.run() # Lower priority number = higher priority (executes first) self.assertEqual(execution_order, ['hook_1', 'hook_5', 'hook_10']) def test_first_pre_hook_rejection_stops_execution(self): """Test that first PRE hook rejection stops other hooks.""" execution_order = [] def hook_reject(context): execution_order.append('reject') raise HookRejectionError(error_code='REJECTED', message='Rejected') def hook_after(context): execution_order.append('after') manager = get_global_cronjob_hook_manager() manager.registry.register('hook_reject', HookType.PRE, hook_reject, operation='execute', priority=1) manager.registry.register('hook_after', HookType.PRE, hook_after, operation='execute', priority=2) task = TestTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) # Only reject hook should execute self.assertEqual(execution_order, ['reject']) @skipIf(not HOOKS_AVAILABLE, "django-package-hooks not installed") class PostHookStandardTests(TestCase): """Test standard POST hook behavior.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) clear_cronjob_hooks() CronExecution.objects.all().delete() def test_post_hook_receives_success_result(self): """Test POST hook receives successful execution result.""" captured_context = [] def post_hook(context): captured_context.append(context) manager = get_global_cronjob_hook_manager() manager.registry.register('post_hook', HookType.POST, post_hook, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() self.assertTrue(result.success) self.assertEqual(len(captured_context), 1) context = captured_context[0] self.assertIsNotNone(context.execution_id) # Has ID now self.assertEqual(context.task_name, 'test-task') def test_post_hook_receives_failure_result(self): """Test POST hook receives failed execution result.""" captured_context = [] def post_hook(context): captured_context.append(context) manager = get_global_cronjob_hook_manager() manager.registry.register('post_hook', HookType.POST, post_hook, operation='execute') task = FailingTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) self.assertEqual(len(captured_context), 1) context = captured_context[0] self.assertIsNotNone(context.execution_id) def test_post_hook_cannot_reject(self): """Test POST hook rejections don't affect execution result.""" def post_hook_reject(context): # POST hooks should not raise rejections, but if they do, # it should not affect the result raise HookRejectionError(error_code='POST_REJECT', message='Should be ignored') manager = get_global_cronjob_hook_manager() manager.registry.register('post_hook_reject', HookType.POST, post_hook_reject, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() # Task should still succeed despite POST hook rejection self.assertTrue(result.success) self.assertEqual(result.message, 'Success') def test_multiple_post_hooks_all_execute(self): """Test multiple POST hooks all execute.""" execution_order = [] def hook_a(context): execution_order.append('a') def hook_b(context): execution_order.append('b') manager = get_global_cronjob_hook_manager() manager.registry.register('hook_a', HookType.POST, hook_a, operation='execute', priority=1) manager.registry.register('hook_b', HookType.POST, hook_b, operation='execute', priority=2) task = TestTask(execution_date=self.execution_date) task.run() self.assertEqual(execution_order, ['a', 'b']) def test_post_hook_exception_logged_but_not_raised(self): """Test POST hook exceptions are logged but don't affect result.""" def post_hook_exception(context): raise ValueError("POST hook error") manager = get_global_cronjob_hook_manager() manager.registry.register('post_hook_exception', HookType.POST, post_hook_exception, operation='execute') task = TestTask(execution_date=self.execution_date) # Should not raise exception with patch('django_cronjob_utils.base.logger') as mock_logger: result = task.run() # Task should still succeed self.assertTrue(result.success) # POST hook error should be logged as warning mock_logger.warning.assert_called() # ============================================================================ # EDGE CASE TESTS # ============================================================================ @skipIf(not HOOKS_AVAILABLE, "django-package-hooks not installed") class HookEdgeCaseTests(TestCase): """Test edge cases in hook behavior.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) clear_cronjob_hooks() CronExecution.objects.all().delete() def test_pre_hook_modifies_metadata(self): """Test PRE hook can modify metadata for POST hooks.""" def pre_hook(context): context.metadata['pre_hook_data'] = 'from_pre' captured_metadata = [] def post_hook(context): captured_metadata.append(context.metadata.copy()) manager = get_global_cronjob_hook_manager() manager.registry.register('pre_hook', HookType.PRE, pre_hook, operation='execute') manager.registry.register('post_hook', HookType.POST, post_hook, operation='execute') task = TestTask(execution_date=self.execution_date) task.run() # POST hook should see PRE hook's metadata self.assertEqual(len(captured_metadata), 1) self.assertEqual(captured_metadata[0]['pre_hook_data'], 'from_pre') def test_hook_with_empty_task_code(self): """Test hook with task that has empty code.""" class NoCodeTask(CronTask): task_name = 'no-code-task' task_code = '' def execute(self, date: date) -> dict: return {'error': False} captured_context = [] def pre_hook(context): captured_context.append(context) manager = get_global_cronjob_hook_manager() manager.registry.register('pre_hook', HookType.PRE, pre_hook, operation='execute') task = NoCodeTask(execution_date=self.execution_date) result = task.run() self.assertTrue(result.success) self.assertEqual(captured_context[0].task_code, '') def test_pre_hook_rejection_with_minimal_info(self): """Test PRE hook rejection with minimal error info.""" def reject_hook(context): # Only provide error_code raise HookRejectionError(error_code='MINIMAL', message='') manager = get_global_cronjob_hook_manager() manager.registry.register('reject_hook', HookType.PRE, reject_hook, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) self.assertEqual(result.error_code, 'MINIMAL') def test_pre_hook_rejection_with_extra_details(self): """Test PRE hook rejection with extra details.""" def reject_hook(context): raise HookRejectionError( error_code='WITH_DETAILS', message='Rejected with details', details={'reason': 'dependency', 'required_task': 'other-task'} ) manager = get_global_cronjob_hook_manager() manager.registry.register('reject_hook', HookType.PRE, reject_hook, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) self.assertEqual(result.error_code, 'WITH_DETAILS') self.assertIsNotNone(result.data) self.assertEqual(result.data['reason'], 'dependency') def test_task_with_exception_still_calls_post_hooks(self): """Test POST hooks behavior when task raises exception.""" captured_context = [] def post_hook(context): captured_context.append(context) manager = get_global_cronjob_hook_manager() manager.registry.register('post_hook_except', HookType.POST, post_hook, operation='execute') task = ExceptionTask(execution_date=self.execution_date) # Exception tasks re-raise the exception with self.assertRaises(ValueError): task.run() # Note: In current implementation, POST hooks may not be called if # exception is raised during execution, as the exception handler # doesn't call POST hooks. This is by design - POST hooks run after # successful completion of the task execution logic. # ============================================================================ # OUTLIER TESTS - Unusual or unexpected scenarios # ============================================================================ @skipIf(not HOOKS_AVAILABLE, "django-package-hooks not installed") class HookOutlierTests(TestCase): """Test outlier scenarios for hooks.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) clear_cronjob_hooks() CronExecution.objects.all().delete() def test_hook_registered_but_never_triggered(self): """Test hook registered for different event is not triggered.""" hook_called = [] def hook(context): hook_called.append(True) # Don't register the hook at all - it won't be triggered # (We can't register for invalid operations in HookRegistry) task = TestTask(execution_date=self.execution_date) task.run() # Hook should not be called self.assertEqual(hook_called, []) def test_many_hooks_performance(self): """Test system with many hooks registered.""" execution_order = [] # Register 20 hooks (reduced from 50 for faster tests) manager = get_global_cronjob_hook_manager() for i in range(20): def make_hook(index): def hook(context): execution_order.append(index) return hook manager.registry.register(f'hook_{i}', HookType.PRE, make_hook(i), operation='execute', priority=i) task = TestTask(execution_date=self.execution_date) result = task.run() # All hooks should execute in order self.assertTrue(result.success) self.assertEqual(len(execution_order), 20) self.assertEqual(execution_order, list(range(20))) def test_pre_hook_rejection_preserves_database_state(self): """Test PRE hook rejection doesn't create orphaned records.""" # Create some initial records initial_count = CronExecution.objects.count() def reject_hook(context): raise HookRejectionError(error_code='REJECTED', message='No records') manager = get_global_cronjob_hook_manager() manager.registry.register('reject_hook', HookType.PRE, reject_hook, operation='execute') task = TestTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) # No new records should be created final_count = CronExecution.objects.count() self.assertEqual(initial_count, final_count) # ============================================================================ # INTEGRATION TESTS - Real-world scenarios # ============================================================================ @skipIf(not HOOKS_AVAILABLE, "django-package-hooks not installed") class HookIntegrationTests(TestCase): """Test real-world hook integration scenarios.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) clear_cronjob_hooks() CronExecution.objects.all().delete() def tearDown(self): """Clean up after test.""" clear_cronjob_hooks() super().tearDown() def test_dependency_checking_hook(self): """Test realistic dependency checking hook.""" # Simulate dependency task completion dependency_task = TestTask(execution_date=self.execution_date) dependency_task.task_name = 'dependency-task' dependency_task.task_code = 'DEP001' dependency_result = dependency_task.run() self.assertTrue(dependency_result.success) def check_dependency(context): if context.task_code == 'T001': # Check if dependency completed dependency_exists = CronExecution.objects.filter( task_code='DEP001', execution_date=context.execution_date, completed=True, success=True ).exists() if not dependency_exists: raise HookRejectionError( error_code='DEPENDENCY_NOT_MET', message='Required task DEP001 not completed' ) manager = get_global_cronjob_hook_manager() manager.registry.register('check_dependency', HookType.PRE, check_dependency, operation='execute') # Now run the dependent task task = TestTask(execution_date=self.execution_date) result = task.run() # Should succeed because dependency exists self.assertTrue(result.success) def test_audit_logging_hook(self): """Test realistic audit logging POST hook.""" audit_log = [] def audit_hook(context): # Fetch execution record to get full details if context.execution_id: execution = CronExecution.objects.get(id=context.execution_id) audit_log.append({ 'task': context.task_name, 'completed': execution.completed, 'success': execution.success, 'duration': execution.duration, }) manager = get_global_cronjob_hook_manager() manager.registry.register('audit_hook', HookType.POST, audit_hook, operation='execute') # Run successful task task1 = TestTask(execution_date=self.execution_date) task1.run() # Run failing task task2 = FailingTask(execution_date=self.execution_date) task2.run() # Check audit log self.assertEqual(len(audit_log), 2) self.assertTrue(audit_log[0]['success']) self.assertFalse(audit_log[1]['success']) def test_retry_limiting_hook(self): """Test hook that limits retry attempts.""" def limit_retries(context): if context.task_code == 'T002': # FailingTask # Check previous retry count retry_count = CronExecution.objects.filter( task_code=context.task_code, execution_date=context.execution_date ).count() if retry_count >= 3: raise HookRejectionError( error_code='MAX_RETRIES_EXCEEDED', message=f'Task has already failed {retry_count} times' ) manager = get_global_cronjob_hook_manager() manager.registry.register('limit_retries', HookType.PRE, limit_retries, operation='execute') # Run failing task multiple times # First 3 attempts should execute for i in range(3): task = FailingTask(execution_date=self.execution_date, force=True) result = task.run() self.assertFalse(result.success) if result.error_code != 'MAX_RETRIES_EXCEEDED': self.assertEqual(result.error_code, 'FAIL') # 4th attempt should be rejected by hook task = FailingTask(execution_date=self.execution_date, force=True) result = task.run() self.assertFalse(result.success) self.assertEqual(result.error_code, 'MAX_RETRIES_EXCEEDED')