|
|
@@ -2,8 +2,10 @@ from io import StringIO
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
from django.core.management import call_command
|
|
|
+from django.core.management.base import CommandError
|
|
|
from django.test import TestCase
|
|
|
|
|
|
+from dcim.models import Region
|
|
|
from utilities.management.commands.calculate_cached_counts import Command
|
|
|
|
|
|
|
|
|
@@ -49,3 +51,57 @@ class CalculateCachedCountsTestCase(TestCase):
|
|
|
ChildModel._meta.get_field.assert_called_once_with('parent')
|
|
|
fk_field.related_query_name.assert_called_once_with()
|
|
|
self.assertEqual(dict(models), {ParentModel: {'child_count': 'children'}})
|
|
|
+
|
|
|
+
|
|
|
+class RebuildLtreePathsTestCase(TestCase):
|
|
|
+ """
|
|
|
+ The command must repair path/sort_path values the triggers did not maintain.
|
|
|
+
|
|
|
+ Corruption is injected by writing the path columns directly: the triggers fire on
|
|
|
+ parent_id and the name column, so a raw UPDATE of path bypasses them, reproducing a
|
|
|
+ database whose cascade trigger went missing across a restore.
|
|
|
+ """
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def setUpTestData(cls):
|
|
|
+ cls.parent = Region.objects.create(name='Alpha', slug='alpha-rlp')
|
|
|
+ cls.child = Region.objects.create(name='Beta', slug='beta-rlp', parent=cls.parent)
|
|
|
+
|
|
|
+ def test_rebuilds_stale_path_and_sort_path(self):
|
|
|
+ Region.objects.filter(pk=self.child.pk).update(
|
|
|
+ path='9999999999999999999', sort_path='stale',
|
|
|
+ )
|
|
|
+
|
|
|
+ call_command('rebuild_ltree_paths', 'dcim.region', stdout=StringIO())
|
|
|
+
|
|
|
+ self.child.refresh_from_db()
|
|
|
+ self.assertEqual(
|
|
|
+ self.child.path,
|
|
|
+ f'{str(self.parent.pk).zfill(19)}.{str(self.child.pk).zfill(19)}',
|
|
|
+ )
|
|
|
+ self.assertEqual(self.child.sort_path, f'Alpha{chr(9)}Beta')
|
|
|
+
|
|
|
+ def test_rebuilds_a_stale_sort_path_alone(self):
|
|
|
+ # What a rename leaves behind: the renamed row's own sort_path is rewritten by the
|
|
|
+ # BEFORE trigger, its descendants' are not, and no path changes.
|
|
|
+ Region.objects.filter(pk=self.child.pk).update(sort_path='stale')
|
|
|
+
|
|
|
+ call_command('rebuild_ltree_paths', 'dcim.region', stdout=StringIO())
|
|
|
+
|
|
|
+ self.child.refresh_from_db()
|
|
|
+ self.assertEqual(self.child.sort_path, f'Alpha{chr(9)}Beta')
|
|
|
+
|
|
|
+ def test_rebuilds_every_core_hierarchical_model_by_default(self):
|
|
|
+ out = StringIO()
|
|
|
+
|
|
|
+ call_command('rebuild_ltree_paths', stdout=out)
|
|
|
+
|
|
|
+ output = out.getvalue()
|
|
|
+ for label in ('dcim.region', 'dcim.inventoryitem', 'dcim.inventoryitemtemplate',
|
|
|
+ 'tenancy.tenantgroup', 'wireless.wirelesslangroup'):
|
|
|
+ self.assertIn(label, output)
|
|
|
+ self.assertIn('Finished.', output)
|
|
|
+
|
|
|
+ def test_rejects_a_model_which_is_not_hierarchical(self):
|
|
|
+ with self.assertRaises(CommandError):
|
|
|
+ call_command('rebuild_ltree_paths', 'dcim.site')
|