parent
05e6e85042
commit
928d07e69f
|
|
@ -371,7 +371,10 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
|
|||
queue = get_queue('default')
|
||||
worker = get_worker('default')
|
||||
job = queue.enqueue(self.dummy_job_default)
|
||||
worker.prepare_job_execution(job)
|
||||
# prepare_job_execution() invokes the worker heartbeat, which logs a "re-registering"
|
||||
# warning for this freshly-created (unregistered) worker; suppress the expected noise.
|
||||
with disable_logging():
|
||||
worker.prepare_job_execution(job)
|
||||
url = reverse('core-api:rqtask-stop', args=[job.id])
|
||||
self.assertEqual(job.get_status(), JobStatus.STARTED)
|
||||
|
||||
|
|
|
|||
|
|
@ -423,8 +423,11 @@ class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
|
|||
|
||||
worker = get_worker('default')
|
||||
job = queue.enqueue(self.dummy_job_default)
|
||||
worker.prepare_job_execution(job)
|
||||
worker.prepare_execution(job)
|
||||
# prepare_job_execution() invokes the worker heartbeat, which logs a "re-registering"
|
||||
# warning for this freshly-created (unregistered) worker; suppress the expected noise.
|
||||
with disable_logging():
|
||||
worker.prepare_job_execution(job)
|
||||
worker.prepare_execution(job)
|
||||
|
||||
self.assertEqual(job.get_status(), JobStatus.STARTED)
|
||||
|
||||
|
|
|
|||
|
|
@ -638,6 +638,13 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
|
|||
Pre-existing non-dict action_data must not cause flush_events() to
|
||||
raise.
|
||||
"""
|
||||
# flush_events() logs a warning about the invalid action_data; mute it so the expected
|
||||
# message doesn't clutter the test runner's output.
|
||||
events_logger = logging.getLogger('netbox.events_processor')
|
||||
original_level = events_logger.level
|
||||
events_logger.setLevel(logging.CRITICAL)
|
||||
self.addCleanup(events_logger.setLevel, original_level)
|
||||
|
||||
site_type = ObjectType.objects.get_for_model(Site)
|
||||
webhook = Webhook.objects.get(name='Webhook 1')
|
||||
webhook_type = ObjectType.objects.get_for_model(Webhook)
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import logging
|
||||
from contextlib import contextmanager
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
|
@ -45,6 +46,16 @@ class DummyScript:
|
|||
|
||||
|
||||
class RunScriptTestCase(TestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
# The failure/abort paths log via the module logger `netbox.scripts.<full_name>`. These
|
||||
# tests deliberately exercise those paths, so mute the logger to keep the expected error
|
||||
# messages and tracebacks out of the test runner's output.
|
||||
logger = logging.getLogger('netbox.scripts')
|
||||
original_level = logger.level
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
self.addCleanup(logger.setLevel, original_level)
|
||||
|
||||
def test_run_script_success_commit_true_sets_output_and_job_data(self):
|
||||
runner = _make_runner()
|
||||
script = DummyScript(run_result='hello')
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import logging
|
||||
import uuid
|
||||
from unittest.mock import PropertyMock, patch
|
||||
|
||||
|
|
@ -467,6 +468,16 @@ class ImageAttachmentTestCase(
|
|||
# placeholder URLs instead of real images on disk.
|
||||
model = ImageAttachment
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
# The fixtures use placeholder image URLs with no file on disk, so rendering the thumbnail
|
||||
# column logs a FileNotFoundError traceback for every attachment. The missing files are
|
||||
# expected here, so mute the sorl-thumbnail logger to keep the test output clean.
|
||||
logger = logging.getLogger('sorl.thumbnail')
|
||||
original_level = logger.level
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
self.addCleanup(logger.setLevel, original_level)
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
ct = ContentType.objects.get_for_model(Site)
|
||||
|
|
|
|||
|
|
@ -44,6 +44,11 @@ class ScaffoldInstanceTest(SimpleTestCase):
|
|||
self.enterContext(patch('netbox.scaffold._config_template', return_value=template))
|
||||
self.enterContext(patch('netbox.scaffold._contrib_dir', return_value=contrib_src))
|
||||
|
||||
# scaffold_instance()/main() print per-file progress to stdout (legitimate `netbox setup`
|
||||
# CLI feedback); swallow it here so it doesn't clutter the test runner's output. Tests that
|
||||
# assert on captured output redirect stdout themselves within the individual test method.
|
||||
self.enterContext(contextlib.redirect_stdout(StringIO()))
|
||||
|
||||
def test_scaffolds_configuration_and_contrib_examples(self):
|
||||
"""A fresh target gets conf/__init__.py, conf/configuration.py, local_requirements.txt, and contrib/."""
|
||||
written = scaffold.scaffold_instance(self.target)
|
||||
|
|
|
|||
|
|
@ -137,9 +137,15 @@ def disable_logging(level=logging.CRITICAL):
|
|||
"""
|
||||
Temporarily suppress log messages at or below the specified level (default: critical).
|
||||
"""
|
||||
# Capture the current disable level so it can be restored on exit (rather than assuming
|
||||
# NOTSET), which keeps nested calls well-behaved. The teardown runs inside a finally block so
|
||||
# logging is always restored even if the wrapped block raises.
|
||||
previous_level = logging.root.manager.disable
|
||||
logging.disable(level)
|
||||
yield
|
||||
logging.disable(logging.NOTSET)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
logging.disable(previous_level)
|
||||
|
||||
|
||||
#
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import logging
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.test import TestCase
|
||||
|
|
@ -33,6 +34,16 @@ class NetBoxRQWorkerHeartbeatTestCase(TestCase):
|
|||
super().heartbeat().
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
# These tests exercise the recovery branches, which log a "re-registering" warning (and,
|
||||
# for the Redis-failure case, an exception traceback). Mute the logger so the expected
|
||||
# messages don't clutter the test runner's output.
|
||||
logger = logging.getLogger('netbox.rqworker')
|
||||
original_level = logger.level
|
||||
logger.setLevel(logging.CRITICAL)
|
||||
self.addCleanup(logger.setLevel, original_level)
|
||||
|
||||
def _make_subject(self, is_member, hash_exists=False, marked_dead=False):
|
||||
worker = NetBoxRQWorker.__new__(NetBoxRQWorker)
|
||||
worker.name = 'test-worker'
|
||||
|
|
|
|||
Loading…
Reference in New Issue