""" Tests for django_cronjob_utils base classes. """ from datetime import date, timedelta from unittest.mock import patch, MagicMock from django.test import TestCase from django.utils import timezone from django_cronjob_utils.base import CronTask, ExecutionResult from django_cronjob_utils.registry import ExecutionPattern from django_cronjob_utils.models import CronExecution from django_cronjob_utils.exceptions import ValidationError, ConcurrentExecutionError class TestTask(CronTask): """Test task for base class tests.""" task_name = 'test-task' task_code = 'A001' def execute(self, date: date) -> dict: return {'error': False, 'message': 'Success'} class FailingTask(CronTask): """Task that always fails.""" task_name = 'failing-task' task_code = 'A002' 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 = 'A003' def execute(self, date: date) -> dict: raise ValueError("Test exception") class SkippedTask(CronTask): """Task that returns skipped result.""" task_name = 'skipped-task' task_code = 'A004' def execute(self, date: date) -> ExecutionResult: return ExecutionResult(skipped=True, reason="Already processed") class CronTaskBaseTests(TestCase): """Tests for CronTask base class.""" def setUp(self): """Set up test fixtures.""" self.execution_date = date(2024, 1, 15) def test_task_initialization(self): """Test task initialization.""" task = TestTask(execution_date=self.execution_date) self.assertEqual(task.execution_date, self.execution_date) self.assertEqual(task.options, {}) self.assertIsNone(task.execution) def test_task_initialization_with_options(self): """Test task initialization with options.""" task = TestTask( execution_date=self.execution_date, force=True, rerun=True ) self.assertTrue(task.options.get('force')) self.assertTrue(task.options.get('rerun')) def test_validate_success(self): """Test validation with valid date.""" task = TestTask(execution_date=self.execution_date) task.validate() # Should not raise def test_validate_invalid_date(self): """Test validation with invalid date type.""" task = TestTask(execution_date="2024-01-15") # String instead of date with self.assertRaises(ValidationError) as cm: task.validate() self.assertIn("date object", str(cm.exception)) def test_should_run_standard_pattern_not_executed(self): """Test should_run with STANDARD pattern when not executed.""" task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.STANDARD self.assertTrue(task.should_run()) def test_should_run_standard_pattern_already_executed(self): """Test should_run with STANDARD pattern when already executed.""" CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True ) task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.STANDARD self.assertFalse(task.should_run()) def test_should_run_always_pattern(self): """Test should_run with ALWAYS pattern.""" # Even if executed, should return True CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True ) task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.ALWAYS self.assertTrue(task.should_run()) def test_should_run_rerun_on_failure_no_previous(self): """Test should_run with RERUN_ON_FAILURE pattern when no previous execution.""" task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.RERUN_ON_FAILURE self.assertTrue(task.should_run()) def test_should_run_rerun_on_failure_previous_success(self): """Test should_run with RERUN_ON_FAILURE pattern when previous succeeded.""" CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True, success=True ) task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.RERUN_ON_FAILURE self.assertFalse(task.should_run()) def test_should_run_rerun_on_failure_previous_failure(self): """Test should_run with RERUN_ON_FAILURE pattern when previous failed.""" CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True, success=False ) task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.RERUN_ON_FAILURE self.assertTrue(task.should_run()) def test_should_run_rate_limited_pattern(self): """Test should_run with RATE_LIMITED pattern.""" task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.RATE_LIMITED # Should always return True (locking handles concurrency) self.assertTrue(task.should_run()) def test_should_run_force_option(self): """Test should_run with force option.""" CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True ) task = TestTask(execution_date=self.execution_date, force=True) task.execution_pattern = ExecutionPattern.STANDARD self.assertTrue(task.should_run()) def test_should_run_rerun_option(self): """Test should_run with rerun option.""" CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True, success=True ) task = TestTask(execution_date=self.execution_date, rerun=True) task.execution_pattern = ExecutionPattern.RERUN_ON_FAILURE self.assertTrue(task.should_run()) def test_create_execution_record(self): """Test creating execution record.""" task = TestTask(execution_date=self.execution_date) execution = task.create_execution_record() self.assertIsNotNone(execution.id) self.assertEqual(execution.task_code, 'A001') self.assertEqual(execution.task_name, 'test-task') self.assertEqual(execution.execution_date, self.execution_date) self.assertFalse(execution.completed) self.assertFalse(execution.success) self.assertEqual(task.execution, execution) def test_execute_with_timeout_no_timeout(self): """Test execute_with_timeout without timeout.""" task = TestTask(execution_date=self.execution_date) result = task.execute_with_timeout() self.assertTrue(result.success) self.assertEqual(result.message, 'Success') def test_execute_with_timeout_with_timeout(self): """Test execute_with_timeout with timeout.""" task = TestTask(execution_date=self.execution_date) task.timeout = 60 # Mock signal handling for Unix systems with patch('django_cronjob_utils.base.signal') as mock_signal: if hasattr(mock_signal, 'SIGALRM'): mock_signal.SIGALRM = MagicMock() result = task.execute_with_timeout() self.assertTrue(result.success) def test_execute_dict_return(self): """Test execute with dict return value.""" task = TestTask(execution_date=self.execution_date) result = task._execute() self.assertTrue(result.success) self.assertEqual(result.message, 'Success') self.assertFalse(result.skipped) def test_execute_execution_result_return(self): """Test execute with ExecutionResult return value.""" task = SkippedTask(execution_date=self.execution_date) result = task._execute() self.assertTrue(result.skipped) self.assertEqual(result.reason, "Already processed") def test_execute_exception_handling(self): """Test execute exception handling.""" task = ExceptionTask(execution_date=self.execution_date) result = task._execute() self.assertFalse(result.success) self.assertIn("Test exception", result.message) self.assertEqual(result.error_code, "ValueError") def test_get_error_code(self): """Test get_error_code method.""" task = TestTask(execution_date=self.execution_date) error_code = task.get_error_code(ValueError("Test")) self.assertEqual(error_code, "ValueError") error_code = task.get_error_code(RuntimeError("Test")) self.assertEqual(error_code, "RuntimeError") def test_should_retry_no_retry_enabled(self): """Test should_retry when retry is disabled.""" task = TestTask(execution_date=self.execution_date) task.retry_on_failure = False task.create_execution_record() self.assertFalse(task.should_retry()) def test_should_retry_max_retries_reached(self): """Test should_retry when max retries reached.""" task = TestTask(execution_date=self.execution_date) task.retry_on_failure = True task.max_retries = 3 task.create_execution_record() task.execution.retry_count = 3 self.assertFalse(task.should_retry()) def test_should_retry_within_limit(self): """Test should_retry when within retry limit.""" task = TestTask(execution_date=self.execution_date) task.retry_on_failure = True task.max_retries = 3 task.create_execution_record() task.execution.retry_count = 1 self.assertTrue(task.should_retry()) def test_should_retry_no_execution_record(self): """Test should_retry when no execution record exists.""" task = TestTask(execution_date=self.execution_date) task.retry_on_failure = True self.assertFalse(task.should_retry()) def test_schedule_retry(self): """Test schedule_retry method.""" task = TestTask(execution_date=self.execution_date) task.create_execution_record() initial_count = task.execution.retry_count task.schedule_retry() task.execution.refresh_from_db() self.assertEqual(task.execution.retry_count, initial_count + 1) def test_run_successful_execution(self): """Test run method with successful execution.""" task = TestTask(execution_date=self.execution_date) result = task.run() self.assertTrue(result.success) self.assertEqual(result.message, 'Success') self.assertFalse(result.skipped) # Check execution record was created and completed execution = CronExecution.objects.get( task_code='A001', execution_date=self.execution_date ) self.assertTrue(execution.completed) self.assertTrue(execution.success) def test_run_skipped_execution(self): """Test run method when execution is skipped.""" CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=True ) task = TestTask(execution_date=self.execution_date) task.execution_pattern = ExecutionPattern.STANDARD result = task.run() self.assertTrue(result.skipped) self.assertEqual(result.reason, "Already executed") def test_run_failed_execution(self): """Test run method with failed execution.""" task = FailingTask(execution_date=self.execution_date) result = task.run() self.assertFalse(result.success) self.assertEqual(result.message, 'Task failed') self.assertEqual(result.error_code, 'FAIL') # Check execution record execution = CronExecution.objects.get( task_code='A002', execution_date=self.execution_date ) self.assertTrue(execution.completed) self.assertFalse(execution.success) def test_run_exception_handling(self): """Test run method exception handling.""" task = ExceptionTask(execution_date=self.execution_date) with self.assertRaises(ValueError): task.run() # Check execution record was marked as failed execution = CronExecution.objects.get( task_code='A003', execution_date=self.execution_date ) self.assertTrue(execution.completed) self.assertFalse(execution.success) self.assertEqual(execution.error_code, "ValueError") def test_run_concurrent_execution_error(self): """Test run method with concurrent execution error.""" # Create running execution CronExecution.objects.create( task_code='A001', task_name='test-task', execution_date=self.execution_date, completed=False ) task = TestTask(execution_date=self.execution_date) with self.assertRaises(ConcurrentExecutionError): task.run() def test_run_with_retry_on_failure(self): """Test run method with retry on failure enabled.""" task = FailingTask(execution_date=self.execution_date) task.retry_on_failure = True task.max_retries = 3 result = task.run() self.assertFalse(result.success) # Check retry was scheduled execution = CronExecution.objects.get( task_code='A002', execution_date=self.execution_date ) self.assertEqual(execution.retry_count, 1) def test_run_with_timeout(self): """Test run method with timeout.""" task = TestTask(execution_date=self.execution_date) task.timeout = 1 # 1 second timeout # Mock a task that takes longer than timeout def slow_execute(date): import time time.sleep(2) return {'error': False} task.execute = slow_execute # On Unix systems with SIGALRM, this should raise TimeoutError # On Windows or systems without SIGALRM, it will just run try: result = task.run() # If no timeout error, execution should complete self.assertIsNotNone(result) except TimeoutError: # Expected on Unix systems execution = CronExecution.objects.get( task_code='A001', execution_date=self.execution_date ) self.assertTrue(execution.completed) self.assertFalse(execution.success) self.assertEqual(execution.error_code, "TIMEOUT") def test_notify_failure(self): """Test notify_failure method.""" task = FailingTask(execution_date=self.execution_date) task.create_execution_record() result = ExecutionResult(success=False, message="Test failure") # Mock notification manager with patch.object(task, '_notification_manager') as mock_notifier: task.notify_failure(result) mock_notifier.notify_failure.assert_called_once() def test_execution_result_creation(self): """Test ExecutionResult creation.""" result = ExecutionResult( success=True, message='Success', error_code='', skipped=False, reason='' ) self.assertTrue(result.success) self.assertEqual(result.message, 'Success') self.assertFalse(result.skipped) def test_execution_result_defaults(self): """Test ExecutionResult default values.""" result = ExecutionResult() self.assertTrue(result.success) # Default is True self.assertEqual(result.message, '') self.assertEqual(result.error_code, '') self.assertFalse(result.skipped) self.assertEqual(result.reason, '')