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 11 Packages Projects Releases Wiki Activity

7848 fix the spectacular test failure

Browse Source
This commit is contained in:
Arthur Hanson 2024-11-22 10:46:28 -08:00
parent 750e19ce88
commit 3e72d7ef07
1 changed files with 16 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -162,7 +162,7 @@ class BackgroundWorkerViewSet(BaseRQViewSet):
class BackgroundTaskViewSet(BaseRQViewSet): class BackgroundTaskViewSet(BaseRQViewSet):
""" """
Retrieve the details of the specified RQ Task. Retrieve a list of RQ Tasks.
""" """
serializer_class = serializers.BackgroundTaskSerializer serializer_class = serializers.BackgroundTaskSerializer
@ -182,27 +182,42 @@ class BackgroundTaskViewSet(BaseRQViewSet):
@extend_schema(responses={200: OpenApiTypes.OBJECT}) @extend_schema(responses={200: OpenApiTypes.OBJECT})
def retrieve(self, request, pk): def retrieve(self, request, pk):
"""
Retrieve the details of the specified RQ Task.
"""
task = self.get_task_from_id(pk) task = self.get_task_from_id(pk)
serializer = self.serializer_class(task, context={'request': request}) serializer = self.serializer_class(task, context={'request': request})
return Response(serializer.data) return Response(serializer.data)
@action(methods=["POST"], detail=True) @action(methods=["POST"], detail=True)
def delete(self, request, pk): def delete(self, request, pk):
"""
Delete the specified RQ Task.
"""
delete_rq_job(pk) delete_rq_job(pk)
return HttpResponse(status=200) return HttpResponse(status=200)
@action(methods=["POST"], detail=True) @action(methods=["POST"], detail=True)
def requeue(self, request, pk): def requeue(self, request, pk):
"""
Requeues the specified RQ Task.
"""
requeue_rq_job(pk) requeue_rq_job(pk)
return HttpResponse(status=200) return HttpResponse(status=200)
@action(methods=["POST"], detail=True) @action(methods=["POST"], detail=True)
def enqueue(self, request, pk): def enqueue(self, request, pk):
"""
Enqueues the specified RQ Task.
"""
enqueue_rq_job(pk) enqueue_rq_job(pk)
return HttpResponse(status=200) return HttpResponse(status=200)
@action(methods=["POST"], detail=True) @action(methods=["POST"], detail=True)
def stop(self, request, pk): def stop(self, request, pk):
"""
Stops the specified RQ Task.
"""
stopped_jobs = stop_rq_job(pk) stopped_jobs = stop_rq_job(pk)
if len(stopped_jobs) == 1: if len(stopped_jobs) == 1:
return HttpResponse(status=200) return HttpResponse(status=200)