""" Tests for django_cronjob_utils registry. """ from datetime import date from django.test import TestCase from django_cronjob_utils.registry import TaskRegistry, ExecutionPattern from django_cronjob_utils.base import CronTask, ExecutionResult from django_cronjob_utils.exceptions import TaskNotFoundError class TestTask(CronTask): """Test task for registry tests.""" def execute(self, date: date) -> dict: return {'error': False, 'message': 'Success'} class TestTask2(CronTask): """Another test task for registry tests.""" def execute(self, date: date) -> dict: return {'error': False, 'message': 'Success 2'} class TaskRegistryTests(TestCase): """Tests for TaskRegistry.""" def setUp(self): """Clear registry before each test.""" TaskRegistry._tasks.clear() TaskRegistry._codes.clear() TaskRegistry._configs.clear() def test_register_task(self): """Test registering a task.""" TaskRegistry.register('test-task', 'A001', TestTask) self.assertTrue(TaskRegistry.is_registered('test-task')) self.assertEqual(TaskRegistry._tasks['test-task'], TestTask) self.assertEqual(TaskRegistry._codes['A001'], 'test-task') self.assertEqual(TestTask.task_name, 'test-task') self.assertEqual(TestTask.task_code, 'A001') def test_register_task_with_config(self): """Test registering a task with configuration.""" TaskRegistry.register( 'test-task', 'A001', TestTask, execution_pattern=ExecutionPattern.ALWAYS, retry_on_failure=True, max_retries=3, retry_delay=600, timeout=3600 ) self.assertEqual(TestTask.execution_pattern, ExecutionPattern.ALWAYS) self.assertTrue(TestTask.retry_on_failure) self.assertEqual(TestTask.max_retries, 3) self.assertEqual(TestTask.retry_delay, 600) self.assertEqual(TestTask.timeout, 3600) def test_register_task_duplicate_name(self): """Test registering duplicate task name raises error.""" TaskRegistry.register('test-task', 'A001', TestTask) with self.assertRaises(ValueError) as cm: TaskRegistry.register('test-task', 'A002', TestTask2) self.assertIn("already registered", str(cm.exception)) def test_register_task_duplicate_code(self): """Test registering duplicate task code raises error.""" TaskRegistry.register('test-task', 'A001', TestTask) with self.assertRaises(ValueError) as cm: TaskRegistry.register('test-task-2', 'A001', TestTask2) self.assertIn("already registered", str(cm.exception)) def test_get_task(self): """Test getting task by name.""" TaskRegistry.register('test-task', 'A001', TestTask) task_class = TaskRegistry.get_task('test-task') self.assertEqual(task_class, TestTask) def test_get_task_not_found(self): """Test getting non-existent task raises error.""" with self.assertRaises(TaskNotFoundError) as cm: TaskRegistry.get_task('non-existent') self.assertIn("not found", str(cm.exception)) self.assertIn("non-existent", str(cm.exception)) def test_get_task_by_code(self): """Test getting task by code.""" TaskRegistry.register('test-task', 'A001', TestTask) task_class = TaskRegistry.get_task_by_code('A001') self.assertEqual(task_class, TestTask) def test_get_task_by_code_not_found(self): """Test getting task by non-existent code raises error.""" with self.assertRaises(TaskNotFoundError) as cm: TaskRegistry.get_task_by_code('Z999') self.assertIn("not found", str(cm.exception)) def test_list_tasks(self): """Test listing all registered tasks.""" self.assertEqual(TaskRegistry.list_tasks(), []) TaskRegistry.register('test-task-1', 'A001', TestTask) TaskRegistry.register('test-task-2', 'A002', TestTask2) tasks = TaskRegistry.list_tasks() self.assertEqual(len(tasks), 2) self.assertIn('test-task-1', tasks) self.assertIn('test-task-2', tasks) def test_get_config(self): """Test getting task configuration.""" config = { 'execution_pattern': ExecutionPattern.ALWAYS, 'retry_on_failure': True } TaskRegistry.register('test-task', 'A001', TestTask, **config) retrieved_config = TaskRegistry.get_config('test-task') self.assertEqual(retrieved_config['execution_pattern'], ExecutionPattern.ALWAYS) self.assertTrue(retrieved_config['retry_on_failure']) def test_get_config_not_found(self): """Test getting config for non-existent task.""" config = TaskRegistry.get_config('non-existent') self.assertEqual(config, {}) def test_is_registered(self): """Test checking if task is registered.""" self.assertFalse(TaskRegistry.is_registered('test-task')) TaskRegistry.register('test-task', 'A001', TestTask) self.assertTrue(TaskRegistry.is_registered('test-task')) def test_register_multiple_tasks(self): """Test registering multiple tasks.""" TaskRegistry.register('task-1', 'A001', TestTask) TaskRegistry.register('task-2', 'A002', TestTask2) self.assertEqual(len(TaskRegistry.list_tasks()), 2) self.assertEqual(TaskRegistry.get_task('task-1'), TestTask) self.assertEqual(TaskRegistry.get_task('task-2'), TestTask2) def test_register_task_default_config(self): """Test registering task with default configuration.""" TaskRegistry.register('test-task', 'A001', TestTask) # Check defaults self.assertEqual(TestTask.execution_pattern, ExecutionPattern.STANDARD) self.assertFalse(TestTask.retry_on_failure) self.assertEqual(TestTask.max_retries, 0) self.assertEqual(TestTask.retry_delay, 300) self.assertIsNone(TestTask.timeout) def test_register_task_partial_config(self): """Test registering task with partial configuration.""" TaskRegistry.register( 'test-task', 'A001', TestTask, execution_pattern=ExecutionPattern.RERUN_ON_FAILURE, timeout=1800 ) self.assertEqual(TestTask.execution_pattern, ExecutionPattern.RERUN_ON_FAILURE) self.assertEqual(TestTask.timeout, 1800) # Should use defaults for others self.assertFalse(TestTask.retry_on_failure) self.assertEqual(TestTask.max_retries, 0) self.assertEqual(TestTask.retry_delay, 300) def test_code_name_mapping(self): """Test code to name mapping.""" TaskRegistry.register('test-task', 'A001', TestTask) # Get task by code should work task_class = TaskRegistry.get_task_by_code('A001') self.assertEqual(task_class, TestTask) # Code should map to name self.assertEqual(TaskRegistry._codes['A001'], 'test-task') def test_empty_registry(self): """Test empty registry behavior.""" self.assertEqual(TaskRegistry.list_tasks(), []) self.assertFalse(TaskRegistry.is_registered('any-task')) with self.assertRaises(TaskNotFoundError): TaskRegistry.get_task('any-task') def test_register_same_class_different_names(self): """Test registering same class with different names/codes.""" # This should work - same class can be registered multiple times # (though unusual in practice) TaskRegistry.register('task-1', 'A001', TestTask) TaskRegistry.register('task-2', 'A002', TestTask) self.assertEqual(len(TaskRegistry.list_tasks()), 2) # Both should point to same class self.assertEqual(TaskRegistry.get_task('task-1'), TestTask) self.assertEqual(TaskRegistry.get_task('task-2'), TestTask) # But class attributes will be overwritten self.assertEqual(TestTask.task_name, 'task-2') # Last registration wins self.assertEqual(TestTask.task_code, 'A002')