import uuid from datetime import datetime, timedelta, timezone from unittest.mock import PropertyMock, patch from django.contrib.auth.models import User from django.test import TestCase, override_settings from django.test.client import Client from django.urls import reverse from rq.job import Job, JobStatus from rq.registry import ( DeferredJobRegistry, FailedJobRegistry, FinishedJobRegistry, ScheduledJobRegistry, StartedJobRegistry, ) from django_rq import get_queue from django_rq.queues import get_scheduler from django_rq.workers import get_worker from .fixtures import access_self, failing_job from .utils import get_queue_index @override_settings(RQ={'AUTOCOMMIT': True}) class ViewTest(TestCase): def setUp(self): self.user = User.objects.create_user('foo', password='pass') self.user.is_staff = True self.user.is_active = True self.user.save() self.client = Client() self.client.login(username=self.user.username, password='pass') get_queue('django_rq_test').connection.flushall() def test_jobs(self): """Jobs in queue are displayed properly""" queue = get_queue('default') job = queue.enqueue(access_self) queue_index = get_queue_index('default') response = self.client.get(reverse('rq_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], [job]) def test_job_details(self): """Job data is displayed properly""" queue = get_queue('default') job = queue.enqueue(access_self) queue_index = get_queue_index('default') url = reverse('rq_job_detail', args=[queue_index, job.id]) response = self.client.get(url) self.assertEqual(response.context['job'], job) # This page shouldn't fail when job.data is corrupt queue.connection.hset(job.key, 'data', 'unpickleable data') response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertIn('DeserializationError', response.content.decode()) def test_job_details_with_results(self): """Job with results is displayed properly""" queue = get_queue('default') job = queue.enqueue(access_self) queue_index = get_queue_index('default') worker = get_worker('default') worker.work(burst=True) result = job.results()[0] url = reverse('rq_job_detail', args=[queue_index, job.id]) response = self.client.get(url) assert result.id self.assertContains(response, result.id) def test_job_details_on_deleted_dependency(self): """Page doesn't crash even if job.dependency has been deleted""" queue = get_queue('default') queue_index = get_queue_index('default') job = queue.enqueue(access_self) second_job = queue.enqueue(access_self, depends_on=job) job.delete() url = reverse('rq_job_detail', args=[queue_index, second_job.id]) response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertIn(second_job._dependency_id, response.content.decode()) def test_requeue_job(self): """ Ensure that a failed job gets requeued when rq_requeue_job is called """ queue = get_queue('default') queue_index = get_queue_index('default') job = queue.enqueue(failing_job) worker = get_worker('default') worker.work(burst=True) job.refresh() self.assertTrue(job.is_failed) self.client.post(reverse('rq_requeue_job', args=[queue_index, job.id]), {'requeue': 'Requeue'}) self.assertIn(job, queue.jobs) job.delete() def test_requeue_all(self): """ Ensure that requeuing all failed job work properly """ queue = get_queue('default') queue_index = get_queue_index('default') queue.enqueue(failing_job) queue.enqueue(failing_job) worker = get_worker('default') worker.work(burst=True) response = self.client.get(reverse('rq_requeue_all', args=[queue_index])) self.assertEqual(response.context['total_jobs'], 2) # After requeue_all is called, jobs are enqueued response = self.client.post(reverse('rq_requeue_all', args=[queue_index])) self.assertEqual(len(queue), 2) def test_requeue_all_if_deleted_job(self): """ Ensure that requeuing all failed job work properly """ queue = get_queue('default') queue_index = get_queue_index('default') job = queue.enqueue(failing_job) queue.enqueue(failing_job) worker = get_worker('default') worker.work(burst=True) response = self.client.get(reverse('rq_requeue_all', args=[queue_index])) self.assertEqual(response.context['total_jobs'], 2) job.delete() # After requeue_all is called, jobs are enqueued response = self.client.post(reverse('rq_requeue_all', args=[queue_index])) self.assertEqual(len(queue), 1) def test_delete_job(self): """ In addition to deleting job from Redis, the job id also needs to be deleted from Queue. """ queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') job = queue.enqueue(access_self) self.client.post(reverse('rq_delete_job', args=[queue_index, job.id]), {'post': 'yes'}) self.assertFalse(Job.exists(job.id, connection=queue.connection)) self.assertNotIn(job.id, queue.job_ids) def test_action_delete_jobs(self): queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') # enqueue some jobs job_ids = [] for _ in range(0, 3): job = queue.enqueue(access_self) job_ids.append(job.id) # remove those jobs using view self.client.post(reverse('rq_actions', args=[queue_index]), {'action': 'delete', 'job_ids': job_ids}) # check if jobs are removed for job_id in job_ids: self.assertFalse(Job.exists(job_id, connection=queue.connection)) self.assertNotIn(job_id, queue.job_ids) def test_enqueue_jobs(self): queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') # enqueue some jobs that depends on other previous_job = None for _ in range(0, 3): job = queue.enqueue(access_self, depends_on=previous_job) previous_job = job # This job is deferred last_job = job self.assertEqual(last_job.get_status(), JobStatus.DEFERRED) self.assertIsNone(last_job.enqueued_at) # We want to force-enqueue this job response = self.client.post(reverse('rq_enqueue_job', args=[queue_index, last_job.id])) # Check that job is updated correctly last_job = queue.fetch_job(last_job.id) assert last_job self.assertEqual(last_job.get_status(), JobStatus.QUEUED) self.assertIsNotNone(last_job.enqueued_at) def test_action_requeue_jobs(self): queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') # enqueue some jobs that will fail jobs = [] job_ids = [] for _ in range(0, 3): job = queue.enqueue(failing_job) jobs.append(job) job_ids.append(job.id) # do those jobs = fail them worker = get_worker('django_rq_test') worker.work(burst=True) # check if all jobs are really failed for job in jobs: self.assertTrue(job.is_failed) # renqueue failed jobs from failed queue self.client.post(reverse('rq_actions', args=[queue_index]), {'action': 'requeue', 'job_ids': job_ids}) # check if we requeue all failed jobs for job in jobs: self.assertFalse(job.is_failed) def test_clear_queue(self): """Test that the queue clear actually clears the queue.""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') job = queue.enqueue(access_self) self.client.post(reverse('rq_clear', args=[queue_index]), {'post': 'yes'}) self.assertFalse(Job.exists(job.id, connection=queue.connection)) self.assertNotIn(job.id, queue.job_ids) def test_finished_jobs(self): """Ensure that finished jobs page works properly.""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') job = queue.enqueue(access_self) registry = FinishedJobRegistry(queue.name, queue.connection) registry.add(job, 2) response = self.client.get(reverse('rq_finished_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], [job]) def test_failed_jobs(self): """Ensure that failed jobs page works properly.""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') # Test that page doesn't fail when FailedJobRegistry is empty response = self.client.get(reverse('rq_failed_jobs', args=[queue_index])) self.assertEqual(response.status_code, 200) job = queue.enqueue(access_self) registry = FailedJobRegistry(queue.name, queue.connection) registry.add(job, 2) response = self.client.get(reverse('rq_failed_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], [job]) def test_scheduled_jobs(self): """Ensure that scheduled jobs page works properly.""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') # Test that page doesn't fail when ScheduledJobRegistry is empty response = self.client.get(reverse('rq_scheduled_jobs', args=[queue_index])) self.assertEqual(response.status_code, 200) job = queue.enqueue_at(datetime.now(), access_self) response = self.client.get(reverse('rq_scheduled_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], [job]) # Test that page doesn't crash when job_id has special characters (exclude :) queue.enqueue_at(datetime.now(), access_self, job_id="job-!@#$%^&*()_=+[]{};',.<>?|`~") response = self.client.get(reverse('rq_scheduled_jobs', args=[queue_index])) self.assertEqual(response.status_code, 200) def test_scheduled_jobs_registry_removal(self): """Ensure that non existing job is being deleted from registry by view""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') registry = ScheduledJobRegistry(queue.name, queue.connection) job = queue.enqueue_at(datetime.now(), access_self) self.assertEqual(len(registry), 1) queue.connection.delete(job.key) response = self.client.get(reverse('rq_scheduled_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], []) self.assertEqual(len(registry), 0) def test_started_jobs(self): """Ensure that active jobs page works properly.""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') worker = get_worker('django_rq_test') job = queue.enqueue(access_self) worker.prepare_execution(job) response = self.client.get(reverse('rq_started_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], [job]) def test_deferred_jobs(self): """Ensure that active jobs page works properly.""" queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') job = queue.enqueue(access_self) registry = DeferredJobRegistry(queue.name, queue.connection) registry.add(job, 2) response = self.client.get(reverse('rq_deferred_jobs', args=[queue_index])) self.assertEqual(response.context['jobs'], [job]) def test_workers(self): """Worker index page should show workers for a specific queue""" queue_index = get_queue_index('django_rq_test') worker1 = get_worker('django_rq_test', name=uuid.uuid4().hex) worker1.register_birth() worker2 = get_worker('test3') worker2.register_birth() response = self.client.get(reverse('rq_workers', args=[queue_index])) self.assertEqual(response.context['workers'], [worker1]) def test_worker_details(self): """Worker index page should show workers for a specific queue""" queue_index = get_queue_index('django_rq_test') worker = get_worker('django_rq_test', name=uuid.uuid4().hex) worker.register_birth() response = self.client.get(reverse('rq_worker_details', args=[queue_index, worker.key])) self.assertEqual(response.context['worker'], worker) def test_statistics_json_view(self): """ Django-RQ's statistic as JSON only viewable by staff or with API_TOKEN """ # Override testing RQ_QUEUES queues = [ { 'connection_config': { 'DB': 0, 'HOST': 'localhost', 'PORT': 6379, }, 'name': 'default', } ] with patch('django_rq.utils.QUEUES_LIST', new_callable=PropertyMock(return_value=queues)): response = self.client.get(reverse('rq_home')) self.assertEqual(response.status_code, 200) response = self.client.get(reverse('rq_home_json')) self.assertEqual(response.status_code, 200) # Not staff, only token self.user.is_staff = False self.user.save() response = self.client.get(reverse('rq_home')) self.assertEqual(response.status_code, 302) # Error, but with 200 code response = self.client.get(reverse('rq_home_json')) self.assertEqual(response.status_code, 200) self.assertIn("error", response.content.decode('utf-8')) # With token, token = '12345abcde' with patch('django_rq.stats_views.API_TOKEN', new_callable=PropertyMock(return_value=token)): response = self.client.get(reverse('rq_home_json', args=[token])) self.assertEqual(response.status_code, 200) self.assertIn("name", response.content.decode('utf-8')) self.assertNotIn('"error": true', response.content.decode('utf-8')) # Wrong token response = self.client.get(reverse('rq_home_json', args=["wrong_token"])) self.assertEqual(response.status_code, 200) self.assertNotIn("name", response.content.decode('utf-8')) self.assertIn('"error": true', response.content.decode('utf-8')) def test_action_stop_jobs(self): queue = get_queue('django_rq_test') queue_index = get_queue_index('django_rq_test') # Enqueue some jobs job_ids, jobs = [], [] worker = get_worker('django_rq_test') # Due to implementation details in RQ v2.x, this test only works # with a single job. This test should be changed to use mocks for _ in range(1): job = queue.enqueue(access_self) job_ids.append(job.id) jobs.append(job) worker.prepare_job_execution(job) worker.prepare_execution(job) # Check if the jobs are started for job_id in job_ids: job = Job.fetch(job_id, connection=queue.connection) self.assertEqual(job.get_status(), JobStatus.STARTED) # Stop those jobs using the view started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection) self.assertEqual(len(started_job_registry), len(job_ids)) self.client.post(reverse('rq_actions', args=[queue_index]), {'action': 'stop', 'job_ids': job_ids}) for job in jobs: worker.monitor_work_horse(job, queue) # Sets the job as Failed and removes from Started self.assertEqual(len(started_job_registry), 0) canceled_job_registry = FailedJobRegistry(queue.name, connection=queue.connection) self.assertEqual(len(canceled_job_registry), len(job_ids)) for job_id in job_ids: self.assertTrue(job_id in canceled_job_registry) # def test_scheduler_jobs(self): # # Override testing RQ_QUEUES # queues = [ # { # "connection_config": { # "DB": 0, # "HOST": "localhost", # "PORT": 6379, # }, # "name": "default", # } # ] # with patch( # "django_rq.utils.QUEUES_LIST", # new_callable=PropertyMock(return_value=queues), # ): # scheduler = get_scheduler("default") # scheduler_index = get_queue_index("default") # # Enqueue some jobs # cron_job = scheduler.cron("10 9 * * *", func=access_self, id="cron-job") # forever_job = scheduler.schedule( # scheduled_time=datetime.now() + timedelta(minutes=10), # interval=600, # func=access_self, # id="forever-repeat", # ) # repeat_job = scheduler.schedule( # scheduled_time=datetime.now() + timedelta(minutes=30), # repeat=30, # func=access_self, # interval=600, # id="thirty-repeat", # ) # response = self.client.get( # reverse("rq_scheduler_jobs", args=[scheduler_index]) # ) # self.assertEqual(response.context["num_jobs"], 3) # context_jobs = {job.id: job for job in response.context["jobs"]} # self.assertEqual(context_jobs["cron-job"].schedule, "cron: '10 9 * * *'") # self.assertEqual(context_jobs["forever-repeat"].schedule, "interval: 600") # self.assertEqual( # context_jobs["thirty-repeat"].schedule, "interval: 600 repeat: 30" # ) # index_response = self.client.get(reverse("rq_home")) # self.assertEqual( # index_response.context["schedulers"], # {"localhost:6379/1": {"count": 3, "index": 0}}, # )