NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-16 04:02:52 -06:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

Suppress extraneous logging in BackgroundTaskTestCase

Browse Source
This commit is contained in:
Jeremy Stretch 2025-05-16 09:51:50 -04:00
parent b6c8502408
commit 26091cfa87
1 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,6 +9,7 @@ from rq.registry import FailedJobRegistry, StartedJobRegistry
from users.models import Token, User from users.models import Token, User
from utilities.testing import APITestCase, APIViewTestCases, TestCase from utilities.testing import APITestCase, APIViewTestCases, TestCase
from utilities.testing.utils import disable_logging
from ..models import * from ..models import *
@ -189,6 +190,7 @@ class BackgroundTaskTestCase(TestCase):
# Enqueue & run a job that will fail # Enqueue & run a job that will fail
job = queue.enqueue(self.dummy_job_failing) job = queue.enqueue(self.dummy_job_failing)
worker = get_worker('default') worker = get_worker('default')
with disable_logging():
worker.work(burst=True) worker.work(burst=True)
self.assertTrue(job.is_failed) self.assertTrue(job.is_failed)
@ -231,6 +233,7 @@ class BackgroundTaskTestCase(TestCase):
self.assertEqual(job.get_status(), JobStatus.STARTED) self.assertEqual(job.get_status(), JobStatus.STARTED)
response = self.client.post(reverse('core-api:rqtask-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) self.assertEqual(response.status_code, 200)
with disable_logging():
worker.monitor_work_horse(job, queue) # Sets the job as Failed and removes from Started worker.monitor_work_horse(job, queue) # Sets the job as Failed and removes from Started
started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection) started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection)
self.assertEqual(len(started_job_registry), 0) self.assertEqual(len(started_job_registry), 0)