NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-25 16:56:10 -06:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

Rename Change to StagedChange

Browse Source
This commit is contained in:
jeremystretch 2022-11-14 10:43:31 -05:00
parent d6f1e18192
commit 18aafcef18
5 changed files with 38 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
netbox/extras/migrations/0084_staging.py
View File

@ -1,5 +1,3 @@
# Generated by Django 4.1.2 on 2022-11-08 16:25
from django.conf import settings from django.conf import settings
from django.db import migrations, models from django.db import migrations, models
import django.db.models.deletion import django.db.models.deletion
@ -29,7 +27,7 @@ class Migration(migrations.Migration):
}, },
), ),
migrations.CreateModel( migrations.CreateModel(
name='Change', name='StagedChange',
fields=[ fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)), ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
('created', models.DateTimeField(auto_now_add=True, null=True)), ('created', models.DateTimeField(auto_now_add=True, null=True)),
@ -37,7 +35,7 @@ class Migration(migrations.Migration):
('action', models.CharField(max_length=20)), ('action', models.CharField(max_length=20)),
('object_id', models.PositiveBigIntegerField(blank=True, null=True)), ('object_id', models.PositiveBigIntegerField(blank=True, null=True)),
('data', models.JSONField(blank=True, null=True)), ('data', models.JSONField(blank=True, null=True)),
('branch', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='changes', to='extras.branch')), ('branch', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='staged_changes', to='extras.branch')),
('object_type', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='+', to='contenttypes.contenttype')), ('object_type', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='+', to='contenttypes.contenttype')),
], ],
options={ options={

4
netbox/extras/models/__init__.py
View File

@ -7,9 +7,8 @@ from .staging import *
from .tags import Tag, TaggedItem from .tags import Tag, TaggedItem
__all__ = ( __all__ = (
'CachedValue',
'Change',
'Branch', 'Branch',
'CachedValue',
'ConfigContext', 'ConfigContext',
'ConfigContextModel', 'ConfigContextModel',
'ConfigRevision', 'ConfigRevision',
@ -23,6 +22,7 @@ __all__ = (
'Report', 'Report',
'SavedFilter', 'SavedFilter',
'Script', 'Script',
'StagedChange',
'Tag', 'Tag',
'TaggedItem', 'TaggedItem',
'Webhook', 'Webhook',

21
netbox/extras/models/staging.py
View File

@ -11,13 +11,16 @@ from utilities.utils import deserialize_object
__all__ = ( __all__ = (
'Branch', 'Branch',
'Change', 'StagedChange',
) )
logger = logging.getLogger('netbox.staging') logger = logging.getLogger('netbox.staging')
class Branch(ChangeLoggedModel): class Branch(ChangeLoggedModel):
"""
A collection of related StagedChanges.
"""
name = models.CharField( name = models.CharField(
max_length=100, max_length=100,
unique=True unique=True
@ -42,16 +45,20 @@ class Branch(ChangeLoggedModel):
def merge(self): def merge(self):
logger.info(f'Merging changes in branch {self}') logger.info(f'Merging changes in branch {self}')
with transaction.atomic(): with transaction.atomic():
for change in self.changes.all(): for change in self.staged_changes.all():
change.apply() change.apply()
self.changes.all().delete() self.staged_changes.all().delete()
class Change(ChangeLoggedModel): class StagedChange(ChangeLoggedModel):
"""
The prepared creation, modification, or deletion of an object to be applied to the active database at a
future point.
"""
branch = models.ForeignKey( branch = models.ForeignKey(
to=Branch, to=Branch,
on_delete=models.CASCADE, on_delete=models.CASCADE,
related_name='changes' related_name='staged_changes'
) )
action = models.CharField( action = models.CharField(
max_length=20, max_length=20,
@ -79,7 +86,9 @@ class Change(ChangeLoggedModel):
ordering = ('pk',) ordering = ('pk',)
def __str__(self): def __str__(self):
return f"{self.get_action_display()} {self.model}" action = self.get_action_display()
app_label, model_name = self.object_type.natural_key()
return f"{action} {app_label}.{model_name} ({self.object_id})"
@property @property
def model(self): def model(self):

12
netbox/netbox/staging.py
View File

@ -5,7 +5,7 @@ from django.db import transaction
from django.db.models.signals import m2m_changed, pre_delete, post_save from django.db.models.signals import m2m_changed, pre_delete, post_save
from extras.choices import ChangeActionChoices from extras.choices import ChangeActionChoices
from extras.models import Change from extras.models import StagedChange
from utilities.utils import serialize_object from utilities.utils import serialize_object
logger = logging.getLogger('netbox.staging') logger = logging.getLogger('netbox.staging')
@ -36,10 +36,10 @@ class checkout:
transaction.set_autocommit(False) transaction.set_autocommit(False)
# Apply any existing Changes assigned to this Branch # Apply any existing Changes assigned to this Branch
changes = self.branch.changes.all() staged_changes = self.branch.staged_changes.all()
if change_count := changes.count(): if change_count := staged_changes.count():
logger.debug(f"Applying {change_count} pre-staged changes...") logger.debug(f"Applying {change_count} pre-staged changes...")
for change in changes: for change in staged_changes:
change.apply() change.apply()
else: else:
logger.debug("No pre-staged changes found") logger.debug("No pre-staged changes found")
@ -91,7 +91,7 @@ class checkout:
object_type, pk = key object_type, pk = key
action, data = change action, data = change
changes.append(Change( changes.append(StagedChange(
branch=self.branch, branch=self.branch,
action=action, action=action,
object_type=object_type, object_type=object_type,
@ -100,7 +100,7 @@ class checkout:
)) ))
# Save all Change instances to the database # Save all Change instances to the database
Change.objects.bulk_create(changes) StagedChange.objects.bulk_create(changes)
# #
# Signal handlers # Signal handlers

26
netbox/netbox/tests/test_staging.py
View File

@ -2,7 +2,7 @@ from django.test import TransactionTestCase
from circuits.models import Provider, Circuit, CircuitType from circuits.models import Provider, Circuit, CircuitType
from extras.choices import ChangeActionChoices from extras.choices import ChangeActionChoices
from extras.models import Branch, Change, Tag from extras.models import Branch, StagedChange, Tag
from ipam.models import ASN, RIR from ipam.models import ASN, RIR
from netbox.staging import checkout from netbox.staging import checkout
from utilities.testing import create_tags from utilities.testing import create_tags
@ -62,7 +62,7 @@ class StagingTestCase(TransactionTestCase):
# Verify that changes have been rolled back after exiting the context # Verify that changes have been rolled back after exiting the context
self.assertEqual(Provider.objects.count(), 3) self.assertEqual(Provider.objects.count(), 3)
self.assertEqual(Circuit.objects.count(), 9) self.assertEqual(Circuit.objects.count(), 9)
self.assertEqual(Change.objects.count(), 5) self.assertEqual(StagedChange.objects.count(), 5)
# Verify that changes are replayed upon entering the context # Verify that changes are replayed upon entering the context
with checkout(branch): with checkout(branch):
@ -81,7 +81,7 @@ class StagingTestCase(TransactionTestCase):
self.assertListEqual(list(provider.asns.all()), list(asns)) self.assertListEqual(list(provider.asns.all()), list(asns))
circuit = Circuit.objects.get(cid='Circuit D1') circuit = Circuit.objects.get(cid='Circuit D1')
self.assertListEqual(list(circuit.tags.all()), list(tags)) self.assertListEqual(list(circuit.tags.all()), list(tags))
self.assertEqual(Change.objects.count(), 0) self.assertEqual(StagedChange.objects.count(), 0)
def test_object_modification(self): def test_object_modification(self):
branch = Branch.objects.create(name='Branch 1') branch = Branch.objects.create(name='Branch 1')
@ -115,7 +115,7 @@ class StagingTestCase(TransactionTestCase):
circuit = Circuit.objects.get(pk=circuit.pk) circuit = Circuit.objects.get(pk=circuit.pk)
self.assertEqual(circuit.cid, 'Circuit A1') self.assertEqual(circuit.cid, 'Circuit A1')
self.assertListEqual(list(circuit.tags.all()), []) self.assertListEqual(list(circuit.tags.all()), [])
self.assertEqual(Change.objects.count(), 5) self.assertEqual(StagedChange.objects.count(), 5)
# Verify that changes are replayed upon entering the context # Verify that changes are replayed upon entering the context
with checkout(branch): with checkout(branch):
@ -138,7 +138,7 @@ class StagingTestCase(TransactionTestCase):
circuit = Circuit.objects.get(pk=circuit.pk) circuit = Circuit.objects.get(pk=circuit.pk)
self.assertEqual(circuit.cid, 'Circuit X') self.assertEqual(circuit.cid, 'Circuit X')
self.assertListEqual(list(circuit.tags.all()), list(tags)) self.assertListEqual(list(circuit.tags.all()), list(tags))
self.assertEqual(Change.objects.count(), 0) self.assertEqual(StagedChange.objects.count(), 0)
def test_object_deletion(self): def test_object_deletion(self):
branch = Branch.objects.create(name='Branch 1') branch = Branch.objects.create(name='Branch 1')
@ -155,7 +155,7 @@ class StagingTestCase(TransactionTestCase):
# Verify that changes have been rolled back after exiting the context # Verify that changes have been rolled back after exiting the context
self.assertEqual(Provider.objects.count(), 3) self.assertEqual(Provider.objects.count(), 3)
self.assertEqual(Circuit.objects.count(), 9) self.assertEqual(Circuit.objects.count(), 9)
self.assertEqual(Change.objects.count(), 4) self.assertEqual(StagedChange.objects.count(), 4)
# Verify that changes are replayed upon entering the context # Verify that changes are replayed upon entering the context
with checkout(branch): with checkout(branch):
@ -166,7 +166,7 @@ class StagingTestCase(TransactionTestCase):
branch.merge() branch.merge()
self.assertEqual(Provider.objects.count(), 2) self.assertEqual(Provider.objects.count(), 2)
self.assertEqual(Circuit.objects.count(), 6) self.assertEqual(Circuit.objects.count(), 6)
self.assertEqual(Change.objects.count(), 0) self.assertEqual(StagedChange.objects.count(), 0)
def test_exit_enter_context(self): def test_exit_enter_context(self):
branch = Branch.objects.create(name='Branch 1') branch = Branch.objects.create(name='Branch 1')
@ -178,8 +178,8 @@ class StagingTestCase(TransactionTestCase):
provider.save() provider.save()
# Check that a create Change was recorded # Check that a create Change was recorded
self.assertEqual(Change.objects.count(), 1) self.assertEqual(StagedChange.objects.count(), 1)
change = Change.objects.first() change = StagedChange.objects.first()
self.assertEqual(change.action, ChangeActionChoices.ACTION_CREATE) self.assertEqual(change.action, ChangeActionChoices.ACTION_CREATE)
self.assertEqual(change.data['name'], provider.name) self.assertEqual(change.data['name'], provider.name)
@ -191,8 +191,8 @@ class StagingTestCase(TransactionTestCase):
provider.save() provider.save()
# Check that a second Change object has been created for the object # Check that a second Change object has been created for the object
self.assertEqual(Change.objects.count(), 2) self.assertEqual(StagedChange.objects.count(), 2)
change = Change.objects.last() change = StagedChange.objects.last()
self.assertEqual(change.action, ChangeActionChoices.ACTION_UPDATE) self.assertEqual(change.action, ChangeActionChoices.ACTION_UPDATE)
self.assertEqual(change.data['name'], provider.name) self.assertEqual(change.data['name'], provider.name)
self.assertEqual(change.data['comments'], provider.comments) self.assertEqual(change.data['comments'], provider.comments)
@ -204,7 +204,7 @@ class StagingTestCase(TransactionTestCase):
provider.delete() provider.delete()
# Check that a third Change has recorded the object's deletion # Check that a third Change has recorded the object's deletion
self.assertEqual(Change.objects.count(), 3) self.assertEqual(StagedChange.objects.count(), 3)
change = Change.objects.last() change = StagedChange.objects.last()
self.assertEqual(change.action, ChangeActionChoices.ACTION_DELETE) self.assertEqual(change.action, ChangeActionChoices.ACTION_DELETE)
self.assertIsNone(change.data) self.assertIsNone(change.data)