Просмотр исходного кода

Skip report/script execution tests if RQ worker is not running

Jeremy Stretch 5 лет назад
Родитель
Сommit
d5a5a4a8d1
1 измененных файлов с 8 добавлено и 0 удалено
  1. 8 0
      netbox/extras/tests/test_api.py

+ 8 - 0
netbox/extras/tests/test_api.py

@@ -1,9 +1,12 @@
 import datetime
 import datetime
+from unittest import skipIf
 
 
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.contenttypes.models import ContentType
 from django.urls import reverse
 from django.urls import reverse
 from django.utils import timezone
 from django.utils import timezone
+from django_rq.queues import get_connection
 from rest_framework import status
 from rest_framework import status
+from rq import Worker
 
 
 from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, RackGroup, RackRole, Site
 from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, RackGroup, RackRole, Site
 from extras.api.views import ReportViewSet, ScriptViewSet
 from extras.api.views import ReportViewSet, ScriptViewSet
@@ -13,6 +16,9 @@ from extras.scripts import BooleanVar, IntegerVar, Script, StringVar
 from utilities.testing import APITestCase, APIViewTestCases
 from utilities.testing import APITestCase, APIViewTestCases
 
 
 
 
+rq_worker_running = Worker.count(get_connection('default'))
+
+
 class AppTest(APITestCase):
 class AppTest(APITestCase):
 
 
     def test_root(self):
     def test_root(self):
@@ -230,6 +236,7 @@ class ReportTest(APITestCase):
 
 
         self.assertEqual(response.data['name'], self.TestReport.__name__)
         self.assertEqual(response.data['name'], self.TestReport.__name__)
 
 
+    @skipIf(not rq_worker_running, "RQ worker not running")
     def test_run_report(self):
     def test_run_report(self):
         self.add_permissions('extras.run_script')
         self.add_permissions('extras.run_script')
 
 
@@ -279,6 +286,7 @@ class ScriptTest(APITestCase):
         self.assertEqual(response.data['vars']['var2'], 'IntegerVar')
         self.assertEqual(response.data['vars']['var2'], 'IntegerVar')
         self.assertEqual(response.data['vars']['var3'], 'BooleanVar')
         self.assertEqual(response.data['vars']['var3'], 'BooleanVar')
 
 
+    @skipIf(not rq_worker_running, "RQ worker not running")
     def test_run_script(self):
     def test_run_script(self):
 
 
         script_data = {
         script_data = {