NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-29 11:56:25 -06:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

7848 cleanup

Browse Source
This commit is contained in:
Arthur Hanson 2024-11-19 10:13:40 -08:00
parent cfbe1c3ad0
commit a8eee91463
4 changed files with 7 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
netbox/core/api/serializers_/tasks.py
View File

@ -47,7 +47,7 @@ class BackgroundTaskSerializer(serializers.Serializer):
class BackgroundQueueSerializer(serializers.Serializer):
name = serializers.CharField()
url = serializers.SerializerMethodField()
# Note: Queue names are not url safe so not returning a url field
jobs = serializers.IntegerField()
oldest_job_timestamp = serializers.CharField()
index = serializers.IntegerField()

1
netbox/core/api/urls.py
View File

@ -12,7 +12,6 @@ router.register('jobs', views.JobViewSet)
router.register('object-changes', views.ObjectChangeViewSet)
router.register('background-queues', views.QueueViewSet, basename='rqqueue')
router.register('background-workers', views.WorkerViewSet, basename='rqworker')
# router.register('background-tasks/(?P<queue_name>.+)/', views.TaskViewSet, basename='rqtask')
router.register('background-tasks', views.TaskViewSet, basename='rqtask')
urlpatterns = router.urls

12
netbox/core/api/views.py
View File

@ -108,6 +108,7 @@ class BaseRQListView(viewsets.ViewSet):
class QueueViewSet(BaseRQListView):
"""
Retrieve a list of RQ Queues.
Note: Queue names are not URL safe so not returning a detail view.
"""
serializer_class = serializers.BackgroundQueueSerializer
lookup_field = 'name'
@ -118,17 +119,6 @@ class QueueViewSet(BaseRQListView):
def get_data(self):
return get_statistics(run_maintenance_tasks=True)["queues"]
def retrieve(self, request, name):
queues = self.get_data()
if not queues:
raise Http404
for queue in queues:
if queue['name'] == name:
return Response(self.serializer_class(queue, context={'request': request}).data)
raise Http404
class WorkerViewSet(BaseRQListView):
"""

49
netbox/core/tests/test_api.py
View File

@ -5,7 +5,6 @@ from django_rq.workers import get_worker
from django.urls import reverse
from django.utils import timezone
from rq.job import Job as RQ_Job, JobStatus
from rq.registry import DeferredJobRegistry, FailedJobRegistry, FinishedJobRegistry, StartedJobRegistry
from users.models import Token, User
from utilities.testing import APITestCase, APIViewTestCases, TestCase
@ -151,49 +150,11 @@ class BackgroundTaskTestCase(TestCase):
self.assertIn('high', str(response.content))
self.assertIn('low', str(response.content))
def test_background_tasks_list_default(self):
queue = get_queue('default')
queue.enqueue(self.dummy_job_default)
response = self.client.get(reverse('core-api:rqtask-list', args=["default",]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn('BackgroundTaskTestCase.dummy_job_default', str(response.content))
def test_background_tasks_list_finished(self):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
registry = FinishedJobRegistry(queue.name, queue.connection)
registry.add(job, 2)
response = self.client.get(reverse('core-api:rqtask-finished', args=["default",]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(job.id, str(response.content))
def test_background_tasks_list_failed(self):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
registry = FailedJobRegistry(queue.name, queue.connection)
registry.add(job, 2)
response = self.client.get(reverse('core-api:rqtask-failed', args=["default"]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(job.id, str(response.content))
def test_background_tasks_list_deferred(self):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
registry = DeferredJobRegistry(queue.name, queue.connection)
registry.add(job, 2)
response = self.client.get(reverse('core-api:rqtask-deferred', args=["default",]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(job.id, str(response.content))
def test_background_task(self):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
response = self.client.get(reverse('core-api:rqtaskdetail-detail', args=[job.id]), **self.header)
response = self.client.get(reverse('core-api:rqtask-detail', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(str(job.id), str(response.content))
self.assertIn('origin', str(response.content))
@ -204,7 +165,7 @@ class BackgroundTaskTestCase(TestCase):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
response = self.client.post(reverse('core-api:rqtaskdetail-delete', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtask-delete', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertFalse(RQ_Job.exists(job.id, connection=queue.connection))
queue = get_queue('default')
@ -220,7 +181,7 @@ class BackgroundTaskTestCase(TestCase):
self.assertTrue(job.is_failed)
# Re-enqueue the failed job and check that its status has been reset
response = self.client.post(reverse('core-api:rqtaskdetail-requeue', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtask-requeue', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
job = RQ_Job.fetch(job.id, queue.connection)
self.assertFalse(job.is_failed)
@ -240,7 +201,7 @@ class BackgroundTaskTestCase(TestCase):
self.assertIsNone(job.enqueued_at)
# Force-enqueue the deferred job
response = self.client.post(reverse('core-api:rqtaskdetail-enqueue', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtask-enqueue', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
# Check that job's status is updated correctly
@ -256,7 +217,7 @@ class BackgroundTaskTestCase(TestCase):
worker.prepare_job_execution(job)
self.assertEqual(job.get_status(), JobStatus.STARTED)
response = self.client.post(reverse('core-api:rqtaskdetail-stop', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtask-stop', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
worker.monitor_work_horse(job, queue) # Sets the job as Failed and removes from Started
started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection)