Method test_cron_scheduler_detail_view

tests/test_cron.py:180–236

Test cron scheduler detail view with various scenarios.

(self)

        self.client.login(username=self.user.username, password='pass')

    def test_cron_scheduler_detail_view(self):
        """Test cron scheduler detail view with various scenarios."""
        # Create a real scheduler and register it
        scheduler = DjangoCronScheduler(name='test-scheduler')
        scheduler.register(
            say_hello,
            "default",
            interval=60,
            args=("hello",),
            kwargs={"target": "world"},
            meta={"source": "test"},
            job_timeout=20,
            result_ttl=500,
            ttl=60,
            failure_ttl=120,
        )
        scheduler.register(say_hello, "default", cron="*/5 * * * *")
        scheduler.register_birth()

        for prefix in ('admin:django_rq_', 'django_rq:'):
            # Test 1: Successful view of existing scheduler
            connection_index = scheduler.connection_index
            url = reverse(f'{prefix}cron_scheduler_detail', args=[connection_index, 'test-scheduler'])
            response = self.client.get(url)
            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.context['scheduler'].name, 'test-scheduler')
            self.assertEqual(len(response.context['cron_jobs']), 2)
            first_cron_job = response.context['cron_jobs'][0]
            self.assertEqual(first_cron_job['func_name'], 'tests.fixtures.say_hello')
            self.assertEqual(first_cron_job['queue_name'], 'default')
            self.assertEqual(first_cron_job['schedule'], 'every 60 seconds')
            self.assertContains(response, 'test-scheduler')
            self.assertContains(response, 'Cron Jobs')
            self.assertContains(response, 'tests.fixtures.say_hello')
            self.assertContains(response, 'default')
            self.assertContains(response, 'every 60 seconds')
            self.assertContains(response, 'cron: */5 * * * *')
            self.assertContains(response, 'hello')
            self.assertContains(response, 'world')
            self.assertContains(response, 'source')
            self.assertContains(response, 'job_timeout=20')
            self.assertContains(response, 'result_ttl=500')
            self.assertContains(response, 'ttl=60')
            self.assertContains(response, 'failure_ttl=120')

            # Test 2: Non-existent scheduler returns 404
            url = reverse(f'{prefix}cron_scheduler_detail', args=[connection_index, 'nonexistent-scheduler'])
            response = self.client.get(url)
            self.assertEqual(response.status_code, 404)

            # Test 3: Invalid connection index returns 404
            url = reverse(f'{prefix}cron_scheduler_detail', args=[999, 'test-scheduler'])
            response = self.client.get(url)
            self.assertEqual(response.status_code, 404)

        # Clean up
        scheduler.register_death()