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

Misc cleanup

Browse Source
This commit is contained in:
Jeremy Stretch 2024-11-20 09:18:26 -05:00
parent cf627d8c47
commit db94671c58
3 changed files with 17 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,7 +9,11 @@ __all__ = (
class BackgroundTaskSerializer(serializers.Serializer): class BackgroundTaskSerializer(serializers.Serializer):
id = serializers.CharField() id = serializers.CharField()
url = serializers.HyperlinkedIdentityField(view_name='core-api:rqtask-detail', lookup_field='id', lookup_url_kwarg='pk') url = serializers.HyperlinkedIdentityField(
view_name='core-api:rqtask-detail',
lookup_field='id',
lookup_url_kwarg='pk'
)
description = serializers.CharField() description = serializers.CharField()
origin = serializers.CharField() origin = serializers.CharField()
func_name = serializers.CharField() func_name = serializers.CharField()
@ -62,10 +66,15 @@ class BackgroundQueueSerializer(serializers.Serializer):
class BackgroundWorkerSerializer(serializers.Serializer): class BackgroundWorkerSerializer(serializers.Serializer):
name = serializers.CharField() name = serializers.CharField()
url = serializers.HyperlinkedIdentityField(view_name='core-api:rqworker-detail', lookup_field='name') url = serializers.HyperlinkedIdentityField(
view_name='core-api:rqworker-detail',
lookup_field='name'
)
state = serializers.SerializerMethodField() state = serializers.SerializerMethodField()
birth_date = serializers.CharField() birth_date = serializers.CharField()
queue_names = serializers.ListField(child=serializers.CharField()) queue_names = serializers.ListField(
child=serializers.CharField()
)
pid = serializers.CharField() pid = serializers.CharField()
successful_job_count = serializers.IntegerField() successful_job_count = serializers.IntegerField()
failed_job_count = serializers.IntegerField() failed_job_count = serializers.IntegerField()

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

@ -85,9 +85,9 @@ class ObjectChangeViewSet(ReadOnlyModelViewSet):
filterset_class = filtersets.ObjectChangeFilterSet filterset_class = filtersets.ObjectChangeFilterSet
class BaseRQListView(viewsets.ViewSet): class BaseRQViewSet(viewsets.ViewSet):
""" """
Retrieve a list of RQ Queues. Base class for RQ view sets. Provides a list() method. Subclasses must implement get_data().
""" """
permission_classes = [IsAdminUser] permission_classes = [IsAdminUser]
serializer_class = None serializer_class = None
@ -105,7 +105,7 @@ class BaseRQListView(viewsets.ViewSet):
return paginator.get_paginated_response(serializer.data) return paginator.get_paginated_response(serializer.data)
class BackgroundQueueViewSet(BaseRQListView): class BackgroundQueueViewSet(BaseRQViewSet):
""" """
Retrieve a list of RQ Queues. Retrieve a list of RQ Queues.
Note: Queue names are not URL safe so not returning a detail view. Note: Queue names are not URL safe so not returning a detail view.
@ -120,7 +120,7 @@ class BackgroundQueueViewSet(BaseRQListView):
return get_statistics(run_maintenance_tasks=True)["queues"] return get_statistics(run_maintenance_tasks=True)["queues"]
class BackgroundWorkerViewSet(BaseRQListView): class BackgroundWorkerViewSet(BaseRQViewSet):
""" """
Retrieve a list of RQ Workers. Retrieve a list of RQ Workers.
""" """
@ -146,7 +146,7 @@ class BackgroundWorkerViewSet(BaseRQListView):
return Response(serializer.data) return Response(serializer.data)
class BackgroundTaskViewSet(BaseRQListView): class BackgroundTaskViewSet(BaseRQViewSet):
""" """
Retrieve the details of the specified RQ Task. Retrieve the details of the specified RQ Task.
""" """

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

@ -103,15 +103,10 @@ class DataFileTest(
class BackgroundTaskTestCase(TestCase): class BackgroundTaskTestCase(TestCase):
user_permissions = () user_permissions = ()
# Dummy worker functions
@staticmethod @staticmethod
def dummy_job_default(): def dummy_job_default():
return "Job finished" return "Job finished"
@staticmethod
def dummy_job_high():
return "Job finished"
@staticmethod @staticmethod
def dummy_job_failing(): def dummy_job_failing():
raise Exception("Job failed") raise Exception("Job failed")