RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge ce3aa55241 into 2f79d987a8

This commit is contained in:
Martin Hauser 2026-09-16 15:09:04 +02:00 committed by GitHub
parent 2f79d987a8 ce3aa55241
commit a75474c6b5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 272 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
netbox/core/models/data.py
View File

@ -6,7 +6,6 @@ from urllib.parse import urlparse
import yaml
from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey
from django.core.exceptions import ValidationError
from django.core.validators import RegexValidator
from django.db import models
@ -18,6 +17,7 @@ from netbox.constants import CENSOR_TOKEN, CENSOR_TOKEN_CHANGED
from netbox.models import PrimaryModel
from netbox.models.features import JobsMixin
from netbox.registry import registry
from utilities.fields import RestrictedGenericForeignKey
from utilities.querysets import RestrictedQuerySet
from ..choices import *
@ -393,7 +393,7 @@ class AutoSyncRecord(models.Model):
related_name='+'
)
object_id = models.PositiveBigIntegerField()
object = GenericForeignKey(
object = RestrictedGenericForeignKey(
ct_field='object_type',
fk_field='object_id'
)

62
netbox/core/signals.py
View File

@ -4,15 +4,19 @@ from threading import local
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.core.signals import request_finished
from django.db import transaction
from django.db.models import CASCADE, RESTRICT
from django.db.models.fields.reverse_related import ManyToManyRel, ManyToOneRel
from django.db.models.signals import m2m_changed, post_migrate, post_save, pre_delete
from django.dispatch import Signal, receiver
from django.utils.translation import gettext_lazy as _
from django.utils.translation import ngettext
from django_prometheus.models import model_deletes, model_inserts, model_updates
from rq.timeouts import JobTimeoutException
from core.choices import JobStatusChoices, ObjectChangeActionChoices
from core.events import *
from core.exceptions import SyncError
from core.models import ObjectType
from extras.events import enqueue_event
from extras.models import Tag
@ -25,6 +29,8 @@ from utilities.exceptions import AbortRequest
from .models import ConfigRevision, DataSource, ObjectChange
logger = logging.getLogger('netbox.core.signals')
__all__ = (
'clear_events',
'job_end',
@ -299,6 +305,10 @@ def enqueue_sync_job(instance, created, **kwargs):
job.delete()
# Keeps the aggregated error readable when a whole source fails at once
_AUTO_SYNC_DETAIL_LIMIT = 10
@receiver(post_sync)
def auto_sync(instance, **kwargs):
"""
@ -306,8 +316,56 @@ def auto_sync(instance, **kwargs):
"""
from .models import AutoSyncRecord
for autosync in AutoSyncRecord.objects.filter(datafile__source=instance).prefetch_related('object'):
autosync.object.sync(save=True)
failure_count = 0
details = []
first_error = None
records = AutoSyncRecord.objects.filter(datafile__source=instance).order_by('pk')
for autosync in records.select_related('object_type').prefetch_related('object'):
# The object may be unresolvable or mid-failure, so identify by keys
target = f'{autosync.object_type.app_label}.{autosync.object_type.model} ID {autosync.object_id}'
if autosync.object_type.model_class() is None:
# Not an orphaned row, so leave it for remove_stale_contenttypes
logger.warning(f"Skipping AutoSyncRecord for uninstalled model {target}")
continue
try:
# The try must stay outside this, so the savepoint is rolled back before the handler runs
with transaction.atomic():
obj = autosync.object
if obj is None:
# The prefetch resolves through the default manager, so recheck with the base manager
try:
obj = autosync.object_type.get_object_for_this_type(pk=autosync.object_id)
except ObjectDoesNotExist:
logger.warning(f"Deleting stale AutoSyncRecord for {target}")
autosync.delete()
continue
obj.sync(save=True)
except JobTimeoutException:
# rq arms one alarm per job, so a timeout is not an ordinary per-object failure
raise
except Exception as e:
failure_count += 1
if first_error is None:
first_error = e
# Not capped, unlike the raised message below
logger.error(f"Error auto-syncing {target}: {e}", exc_info=True)
if len(details) < _AUTO_SYNC_DETAIL_LIMIT:
details.append(f'- {target}: {type(e).__name__}: {e}')
if first_error is not None:
summary = ngettext(
'Automatic synchronization failed for {count} object:',
'Automatic synchronization failed for {count} objects:',
failure_count,
).format(count=failure_count)
if omitted := failure_count - len(details):
details.append(ngettext(
'and {count} more failure.',
'and {count} more failures.',
omitted,
).format(count=omitted))
raise SyncError('\n'.join([summary, *details])) from first_error
@receiver(post_save, sender=ConfigRevision)

222
netbox/core/tests/test_signals.py
View File

@ -5,16 +5,19 @@ from unittest.mock import MagicMock, Mock, patch
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.core.signals import request_finished
from django.db import transaction
from django.db import IntegrityError, transaction
from django.test import RequestFactory, SimpleTestCase, TestCase, override_settings
from django.utils import timezone
from rq.timeouts import JobTimeoutException
from core import signals
from core.choices import DataSourceStatusChoices, JobStatusChoices, ObjectChangeActionChoices
from core.models import ConfigRevision, DataSource, ObjectChange, ObjectType
from core.exceptions import SyncError
from core.models import AutoSyncRecord, ConfigRevision, DataFile, DataSource, ObjectChange, ObjectType
from core.signals import _signals_received, clear_events, post_sync
from dcim.choices import InterfaceTypeChoices
from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site, SiteGroup
from extras.models import Tag
from extras.models import ConfigContext, Tag
from extras.validators import CustomValidator
from netbox.context import events_queue
from netbox.context_managers import event_tracking
@ -348,28 +351,223 @@ class AutoSyncSignalTestCase(TestCase):
DataSource when post_sync fires.
"""
def test_post_sync_resyncs_dependent_records(self):
ds = DataSource.objects.create(
@classmethod
def setUpTestData(cls):
cls.datasource = DataSource.objects.create(
name='DS 1',
type='local',
source_url='/tmp/ds1',
status=DataSourceStatusChoices.COMPLETED,
)
record_a = SimpleNamespace(object=SimpleNamespace(synced=False))
cls.object_type = ObjectType.objects.get_for_model(ConfigContext)
def make_record(self, object_id, obj):
"""Stand in for an AutoSyncRecord, carrying the attributes the receiver reads."""
return SimpleNamespace(object=obj, object_type=self.object_type, object_id=object_id)
def patch_records(self, autosync_model, records):
"""Point the patched manager's queryset chain at the given stand-in records."""
queryset = autosync_model.objects.filter.return_value.order_by.return_value.select_related.return_value
queryset.prefetch_related.return_value = records
def test_post_sync_resyncs_dependent_records(self):
record_a = self.make_record(1, SimpleNamespace(synced=False))
record_a.object.sync = lambda save: setattr(record_a.object, 'synced', save)
record_b = SimpleNamespace(object=SimpleNamespace(synced=False))
record_b = self.make_record(2, SimpleNamespace(synced=False))
record_b.object.sync = lambda save: setattr(record_b.object, 'synced', save)
with patch('core.models.AutoSyncRecord') as autosync_model:
autosync_model.objects.filter.return_value.prefetch_related.return_value = [
record_a,
record_b,
]
post_sync.send(sender=ds.__class__, instance=ds)
self.patch_records(autosync_model, [record_a, record_b])
post_sync.send(sender=DataSource, instance=self.datasource)
self.assertTrue(record_a.object.synced)
self.assertTrue(record_b.object.synced)
def test_post_sync_continues_after_failed_record(self):
"""A failing record leaves the connection usable for the records after it."""
record_a = self.make_record(1, MagicMock())
record_b = self.make_record(2, MagicMock())
def create_duplicate_datasource(save):
# Violates the unique constraint on DataSource.name
DataSource.objects.create(name=self.datasource.name, type='local', source_url='/tmp/duplicate')
def create_datasource(save):
DataSource.objects.create(name='DS 2', type='local', source_url='/tmp/ds2')
record_a.object.sync.side_effect = create_duplicate_datasource
record_b.object.sync.side_effect = create_datasource
with patch('core.models.AutoSyncRecord') as autosync_model:
self.patch_records(autosync_model, [record_a, record_b])
with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm:
post_sync.send(sender=DataSource, instance=self.datasource)
record_a.object.sync.assert_called_once_with(save=True)
record_b.object.sync.assert_called_once_with(save=True)
self.assertIsInstance(cm.exception.__cause__, IntegrityError)
self.assertIn('Automatic synchronization failed for 1 object:', str(cm.exception))
self.assertIn('- extras.configcontext ID 1: IntegrityError:', str(cm.exception))
self.assertTrue(DataSource.objects.filter(name='DS 2').exists())
def test_post_sync_reports_every_failure(self):
"""Each failed object is named in the aggregated error."""
record_a = self.make_record(1, MagicMock())
record_b = self.make_record(2, MagicMock())
record_a.object.sync.side_effect = ValueError('First failure')
record_b.object.sync.side_effect = RuntimeError('Second failure')
with patch('core.models.AutoSyncRecord') as autosync_model:
self.patch_records(autosync_model, [record_a, record_b])
with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm:
post_sync.send(sender=DataSource, instance=self.datasource)
record_a.object.sync.assert_called_once_with(save=True)
record_b.object.sync.assert_called_once_with(save=True)
self.assertEqual(
str(cm.exception),
'Automatic synchronization failed for 2 objects:\n'
'- extras.configcontext ID 1: ValueError: First failure\n'
'- extras.configcontext ID 2: RuntimeError: Second failure',
)
self.assertIsInstance(cm.exception.__cause__, ValueError)
def test_post_sync_deletes_dangling_record(self):
"""A record whose generic relation no longer resolves is removed instead of failing the sync."""
dangling = MagicMock(object=None, object_type=self.object_type, object_id=7)
record = self.make_record(1, MagicMock())
with patch('core.models.AutoSyncRecord') as autosync_model:
self.patch_records(autosync_model, [dangling, record])
with self.assertLogs('netbox.core.signals', 'WARNING'):
post_sync.send(sender=DataSource, instance=self.datasource)
dangling.delete.assert_called_once_with()
record.object.sync.assert_called_once_with(save=True)
def make_datafile(self, path='dir1/context.yaml'):
"""Create a DataFile on the source the receiver filters by."""
return DataFile.objects.create(
source=self.datasource,
path=path,
last_updated=timezone.now(),
size=1000,
hash='442da078f0111cbdf42f21903724f6597c692535f55bdfbbea758a1ae99ad9e1',
data=b'value: original',
)
def test_post_sync_skips_record_for_uninstalled_model(self):
"""A record whose model is no longer installed is skipped rather than deleted."""
# get_for_id() caches the ghost type in the manager, which the rollback does not undo
self.addCleanup(ContentType.objects.clear_cache)
stale_type = ContentType.objects.create(app_label='ghost_plugin', model='ghostmodel')
AutoSyncRecord.objects.create(
datafile=self.make_datafile('dir1/ghost.yaml'),
object_type=stale_type,
object_id=1,
)
with self.assertLogs('netbox.core.signals', 'WARNING') as logs:
post_sync.send(sender=DataSource, instance=self.datasource)
# remove_stale_contenttypes owns this cleanup and cascades to the record
self.assertTrue(AutoSyncRecord.objects.filter(object_type=stale_type).exists())
self.assertIn('ghost_plugin.ghostmodel ID 1', logs.output[0])
def test_post_sync_deletes_real_dangling_record(self):
"""The stale record is removed from the database, exercising the unmocked queryset."""
datafile = self.make_datafile()
AutoSyncRecord.objects.create(
datafile=datafile,
object_type=ObjectType.objects.get_for_model(ConfigContext),
object_id=99999,
)
with self.assertLogs('netbox.core.signals', 'WARNING') as logs:
post_sync.send(sender=DataSource, instance=self.datasource)
self.assertFalse(AutoSyncRecord.objects.filter(datafile=datafile).exists())
# The mocked tests must not assert a format the real content type never emits
self.assertIn('extras.configcontext ID 99999', logs.output[0])
def test_post_sync_syncs_record_hidden_by_default_manager(self):
"""A target the default manager excludes is synced through the base manager, not deleted."""
datafile = self.make_datafile('dir1/hidden.yaml')
context = ConfigContext.objects.create(
name='CC 1',
data={},
data_source=self.datasource,
data_file=datafile,
data_path=datafile.path,
auto_sync_enabled=True,
)
hidden = AutoSyncRecord.objects.get(object_type=self.object_type, object_id=context.pk)
missing = AutoSyncRecord.objects.create(
datafile=datafile,
object_type=self.object_type,
object_id=99999,
)
# The prefetch reads ConfigContext.objects, so an empty queryset hides a live target
with patch.object(ConfigContext, 'objects', ConfigContext.objects.none()):
with self.assertLogs('netbox.core.signals', 'WARNING') as logs:
post_sync.send(sender=DataSource, instance=self.datasource)
context.refresh_from_db()
self.assertEqual(context.data, {'value': 'original'})
self.assertTrue(context.is_synced)
self.assertTrue(AutoSyncRecord.objects.filter(pk=hidden.pk).exists())
self.assertFalse(AutoSyncRecord.objects.filter(pk=missing.pk).exists())
self.assertIn('extras.configcontext ID 99999', logs.output[0])
def test_post_sync_reports_dangling_record_when_cleanup_is_refused(self):
"""A stale record is reported by its target keys when its own cleanup is blocked."""
dangling = MagicMock(object=None, object_type=self.object_type, object_id=7)
dangling.delete.side_effect = AbortRequest('Deletion is prevented by a protection rule')
with patch('core.models.AutoSyncRecord') as autosync_model:
self.patch_records(autosync_model, [dangling])
with self.assertLogs('netbox.core.signals', 'ERROR'), self.assertRaises(SyncError) as cm:
post_sync.send(sender=DataSource, instance=self.datasource)
self.assertIn('- extras.configcontext ID 7: AbortRequest:', str(cm.exception))
self.assertNotIn('- None:', str(cm.exception))
def test_post_sync_caps_reported_failures(self):
"""Only the first N failures are detailed, with the remainder counted."""
records = []
for i in range(signals._AUTO_SYNC_DETAIL_LIMIT + 3):
record = self.make_record(i, MagicMock())
record.object.sync.side_effect = ValueError(f'Failure {i}')
records.append(record)
with patch('core.models.AutoSyncRecord') as autosync_model:
self.patch_records(autosync_model, records)
with self.assertLogs('netbox.core.signals', 'ERROR') as logs, self.assertRaises(SyncError) as cm:
post_sync.send(sender=DataSource, instance=self.datasource)
message = str(cm.exception)
self.assertIn(f'failed for {len(records)} objects:', message)
self.assertEqual(message.count('ValueError: Failure'), signals._AUTO_SYNC_DETAIL_LIMIT)
self.assertIn('and 3 more failures.', message)
# Capping the message must not lose an identity, so every failure is still logged
self.assertEqual(len(logs.records), len(records))
# A capped list is only deterministic if the subset is ordered
autosync_model.objects.filter.return_value.order_by.assert_called_once_with('pk')
def test_post_sync_propagates_job_timeout(self):
"""An rq job timeout escapes the receiver instead of being recorded as a failure."""
record_a = self.make_record(1, MagicMock())
record_b = self.make_record(2, MagicMock())
record_a.object.sync.side_effect = JobTimeoutException('Task exceeded maximum timeout value')
with patch('core.models.AutoSyncRecord') as autosync_model:
self.patch_records(autosync_model, [record_a, record_b])
with self.assertRaises(JobTimeoutException):
post_sync.send(sender=DataSource, instance=self.datasource)
record_b.object.sync.assert_not_called()
class UpdateConfigSignalTestCase(TestCase):
"""