|
|
@@ -0,0 +1,911 @@
|
|
|
+"""Tests for the ltree-based hierarchical model infrastructure."""
|
|
|
+from django.contrib.contenttypes.models import ContentType
|
|
|
+from django.db import connection
|
|
|
+from django.test import TestCase
|
|
|
+
|
|
|
+from core.models import ObjectChange
|
|
|
+from dcim.models import Region, Site
|
|
|
+from tenancy.models import Contact, ContactGroup
|
|
|
+
|
|
|
+
|
|
|
+def _path(*pks):
|
|
|
+ """Construct the expected ltree path value from a sequence of PKs.
|
|
|
+
|
|
|
+ Mirrors the trigger's zero-padded label scheme (19 chars per label).
|
|
|
+ """
|
|
|
+ return '.'.join(str(pk).zfill(19) for pk in pks)
|
|
|
+
|
|
|
+
|
|
|
+class LtreeTriggerTests(TestCase):
|
|
|
+ """Verify per-row PostgreSQL triggers maintain `path` correctly."""
|
|
|
+
|
|
|
+ def test_insert_root_path(self):
|
|
|
+ r = Region.objects.create(name='Root', slug='root')
|
|
|
+ self.assertEqual(r.path, _path(r.pk))
|
|
|
+
|
|
|
+ def test_insert_child_path(self):
|
|
|
+ r = Region.objects.create(name='Root', slug='root')
|
|
|
+ c = Region.objects.create(parent=r, name='Child', slug='child')
|
|
|
+ self.assertEqual(c.path, _path(r.pk, c.pk))
|
|
|
+
|
|
|
+ def test_grandchild_path(self):
|
|
|
+ r = Region.objects.create(name='R', slug='r')
|
|
|
+ c = Region.objects.create(parent=r, name='C', slug='c')
|
|
|
+ g = Region.objects.create(parent=c, name='G', slug='g')
|
|
|
+ self.assertEqual(g.path, _path(r.pk, c.pk, g.pk))
|
|
|
+
|
|
|
+ def test_move_cascades_to_descendants(self):
|
|
|
+ r = Region.objects.create(name='R', slug='r')
|
|
|
+ c = Region.objects.create(parent=r, name='C', slug='c')
|
|
|
+ g = Region.objects.create(parent=c, name='G', slug='g')
|
|
|
+ c.parent = None
|
|
|
+ c.save()
|
|
|
+ c.refresh_from_db()
|
|
|
+ g.refresh_from_db()
|
|
|
+ self.assertEqual(c.path, _path(c.pk))
|
|
|
+ self.assertEqual(g.path, _path(c.pk, g.pk))
|
|
|
+
|
|
|
+ def test_bulk_create_populates_paths(self):
|
|
|
+ """BEFORE INSERT trigger fires on bulk_create, populating path."""
|
|
|
+ root = Region.objects.create(name='R', slug='r-bulk')
|
|
|
+ children = Region.objects.bulk_create([
|
|
|
+ Region(parent=root, name=f'C{i}', slug=f'c{i}-bulk') for i in range(3)
|
|
|
+ ])
|
|
|
+ for child in children:
|
|
|
+ child.refresh_from_db()
|
|
|
+ self.assertEqual(child.path, _path(root.pk, child.pk))
|
|
|
+
|
|
|
+ def test_queryset_update_with_parent_id_cascades(self):
|
|
|
+ """Raw .update() that changes parent_id still fires triggers."""
|
|
|
+ r1 = Region.objects.create(name='R1', slug='r1-up')
|
|
|
+ r2 = Region.objects.create(name='R2', slug='r2-up')
|
|
|
+ c = Region.objects.create(parent=r1, name='C', slug='c-up')
|
|
|
+ g = Region.objects.create(parent=c, name='G', slug='g-up')
|
|
|
+
|
|
|
+ Region.objects.filter(pk=c.pk).update(parent=r2)
|
|
|
+ c.refresh_from_db()
|
|
|
+ g.refresh_from_db()
|
|
|
+ self.assertEqual(c.path, _path(r2.pk, c.pk))
|
|
|
+ self.assertEqual(g.path, _path(r2.pk, c.pk, g.pk))
|
|
|
+
|
|
|
+ def test_gist_index_exists(self):
|
|
|
+ """Every ltree-backed table has a GiST index on path."""
|
|
|
+ expected = {
|
|
|
+ 'dcim_region_path_gist',
|
|
|
+ 'dcim_sitegroup_path_gist',
|
|
|
+ 'dcim_location_path_gist',
|
|
|
+ 'dcim_devicerole_path_gist',
|
|
|
+ 'dcim_platform_path_gist',
|
|
|
+ 'dcim_inventoryitem_path_gist',
|
|
|
+ 'dcim_inv_item_tmpl_path_gist',
|
|
|
+ 'dcim_modulebay_path_gist',
|
|
|
+ 'tenancy_tenantgroup_path_gist',
|
|
|
+ 'tenancy_contactgroup_path_gist',
|
|
|
+ 'wireless_lan_grp_path_gist',
|
|
|
+ }
|
|
|
+ with connection.cursor() as cursor:
|
|
|
+ cursor.execute("""
|
|
|
+ SELECT indexname FROM pg_indexes
|
|
|
+ WHERE indexname = ANY(%s) AND indexdef LIKE '%%USING gist%%'
|
|
|
+ """, [list(expected)])
|
|
|
+ found = {row[0] for row in cursor.fetchall()}
|
|
|
+ self.assertSetEqual(found, expected)
|
|
|
+
|
|
|
+
|
|
|
+class AdvisoryLockScopeTests(TestCase):
|
|
|
+ """
|
|
|
+ Pin where the BEFORE trigger takes its per-tree advisory lock.
|
|
|
+
|
|
|
+ A new root (INSERT with parent_id IS NULL) is a race-free singleton tree and
|
|
|
+ must take NO lock, so a bulk import of many top-level objects cannot exhaust
|
|
|
+ the shared lock table. Every other write (child insert, reparent-to-root)
|
|
|
+ must still lock per tree — the companion assertions keep the optimization from
|
|
|
+ silently over-broadening to cases that need serialization.
|
|
|
+
|
|
|
+ `pg_locks` is cluster-wide and advisory *xact* locks are held until the
|
|
|
+ TestCase transaction ends, so each test measures the DELTA in this backend's
|
|
|
+ advisory-lock count across a single operation rather than an absolute count.
|
|
|
+ """
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _advisory_lock_count():
|
|
|
+ with connection.cursor() as cursor:
|
|
|
+ cursor.execute(
|
|
|
+ "SELECT count(*) FROM pg_locks "
|
|
|
+ "WHERE locktype = 'advisory' AND pid = pg_backend_pid()"
|
|
|
+ )
|
|
|
+ return cursor.fetchone()[0]
|
|
|
+
|
|
|
+ def test_root_insert_takes_no_lock(self):
|
|
|
+ before = self._advisory_lock_count()
|
|
|
+ Region.objects.create(name='Root', slug='root-lock')
|
|
|
+ self.assertEqual(self._advisory_lock_count() - before, 0)
|
|
|
+
|
|
|
+ def test_child_insert_takes_one_lock(self):
|
|
|
+ root = Region.objects.create(name='Root', slug='root-lock2')
|
|
|
+ before = self._advisory_lock_count()
|
|
|
+ Region.objects.create(parent=root, name='Child', slug='child-lock2')
|
|
|
+ self.assertEqual(self._advisory_lock_count() - before, 1)
|
|
|
+
|
|
|
+ def test_reparent_to_root_takes_lock(self):
|
|
|
+ root = Region.objects.create(name='Root', slug='root-lock3')
|
|
|
+ child = Region.objects.create(parent=root, name='Child', slug='child-lock3')
|
|
|
+ before = self._advisory_lock_count()
|
|
|
+ child.parent = None
|
|
|
+ child.save()
|
|
|
+ # An existing row promoted to root still has a real subtree to rewrite, so
|
|
|
+ # it must lock (its own new root key, plus the old tree's root for the move).
|
|
|
+ self.assertGreaterEqual(self._advisory_lock_count() - before, 1)
|
|
|
+
|
|
|
+
|
|
|
+class LtreeAPIParityTests(TestCase):
|
|
|
+ """Verify the MPTTModel-compatible API surface."""
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def setUpTestData(cls):
|
|
|
+ # Build: root -> mid -> leaf
|
|
|
+ # -> leaf2 (sibling of mid's child)
|
|
|
+ cls.root = Region.objects.create(name='Root', slug='root-api')
|
|
|
+ cls.mid = Region.objects.create(parent=cls.root, name='Mid', slug='mid-api')
|
|
|
+ cls.leaf = Region.objects.create(parent=cls.mid, name='Leaf', slug='leaf-api')
|
|
|
+ cls.leaf2 = Region.objects.create(parent=cls.mid, name='Leaf2', slug='leaf2-api')
|
|
|
+
|
|
|
+ def test_level(self):
|
|
|
+ self.assertEqual(self.root.level, 0)
|
|
|
+ self.assertEqual(self.mid.level, 1)
|
|
|
+ self.assertEqual(self.leaf.level, 2)
|
|
|
+ self.assertEqual(self.leaf.get_level(), 2)
|
|
|
+
|
|
|
+ def test_get_ancestors(self):
|
|
|
+ ancestors = list(self.leaf.get_ancestors().values_list('name', flat=True))
|
|
|
+ self.assertEqual(ancestors, ['Root', 'Mid'])
|
|
|
+ with_self = list(self.leaf.get_ancestors(include_self=True).values_list('name', flat=True))
|
|
|
+ self.assertEqual(with_self, ['Root', 'Mid', 'Leaf'])
|
|
|
+
|
|
|
+ def test_get_descendants(self):
|
|
|
+ descendants = sorted(self.root.get_descendants().values_list('name', flat=True))
|
|
|
+ self.assertEqual(descendants, ['Leaf', 'Leaf2', 'Mid'])
|
|
|
+
|
|
|
+ def test_get_children(self):
|
|
|
+ children = sorted(self.root.get_children().values_list('name', flat=True))
|
|
|
+ self.assertEqual(children, ['Mid'])
|
|
|
+
|
|
|
+
|
|
|
+class CycleValidationTests(TestCase):
|
|
|
+ """clean() and save() must refuse to assign self or a descendant as parent."""
|
|
|
+
|
|
|
+ def test_cycle_raises(self):
|
|
|
+ from django.core.exceptions import ValidationError
|
|
|
+ a = Region.objects.create(name='A', slug='a-cyc')
|
|
|
+ b = Region.objects.create(parent=a, name='B', slug='b-cyc')
|
|
|
+ Region.objects.create(parent=b, name='C', slug='c-cyc')
|
|
|
+ a.parent = b
|
|
|
+ with self.assertRaises(ValidationError):
|
|
|
+ a.full_clean()
|
|
|
+
|
|
|
+ def test_cycle_raises_on_save_without_clean(self):
|
|
|
+ # The save()-level guard mirrors django-mptt's InvalidMove: a cyclic move
|
|
|
+ # is rejected even when full_clean() is bypassed (scripts, bulk callers).
|
|
|
+ from django.core.exceptions import ValidationError
|
|
|
+ a = Region.objects.create(name='A', slug='a-cyc2')
|
|
|
+ b = Region.objects.create(parent=a, name='B', slug='b-cyc2')
|
|
|
+ Region.objects.create(parent=b, name='C', slug='c-cyc2')
|
|
|
+ a.parent = b
|
|
|
+ with self.assertRaises(ValidationError):
|
|
|
+ a.save()
|
|
|
+ # The rejected move must leave the tree unchanged.
|
|
|
+ a.refresh_from_db()
|
|
|
+ self.assertIsNone(a.parent_id)
|
|
|
+
|
|
|
+ def test_self_parent_raises_on_save(self):
|
|
|
+ from django.core.exceptions import ValidationError
|
|
|
+ a = Region.objects.create(name='A', slug='a-self')
|
|
|
+ a.parent = a
|
|
|
+ with self.assertRaises(ValidationError):
|
|
|
+ a.save()
|
|
|
+
|
|
|
+ def test_move_to_unrelated_parent_is_allowed(self):
|
|
|
+ # A legitimate move (target is neither self nor a descendant) must succeed.
|
|
|
+ a = Region.objects.create(name='A', slug='a-ok')
|
|
|
+ b = Region.objects.create(name='B', slug='b-ok')
|
|
|
+ child = Region.objects.create(parent=a, name='Child', slug='child-ok')
|
|
|
+ child.parent = b
|
|
|
+ child.save()
|
|
|
+ child.refresh_from_db()
|
|
|
+ self.assertEqual(child.parent_id, b.pk)
|
|
|
+ self.assertEqual(child.path, _path(b.pk, child.pk))
|
|
|
+
|
|
|
+
|
|
|
+class SortPathTests(TestCase):
|
|
|
+ """
|
|
|
+ Verify that sort_path produces tree-flatten output with siblings in name
|
|
|
+ order, mirroring MPTT's `order_insertion_by=('name',)` behavior.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_siblings_in_name_order_regardless_of_insertion_order(self):
|
|
|
+ # Create siblings out of name order
|
|
|
+ Region.objects.create(name='Zebra', slug='zebra-sp')
|
|
|
+ Region.objects.create(name='Aardvark', slug='aardvark-sp')
|
|
|
+ buffalo = Region.objects.create(name='Buffalo', slug='buffalo-sp')
|
|
|
+
|
|
|
+ # Children of Buffalo also out of order
|
|
|
+ Region.objects.create(parent=buffalo, name='Zoo', slug='b-zoo-sp')
|
|
|
+ Region.objects.create(parent=buffalo, name='Apex', slug='b-apex-sp')
|
|
|
+
|
|
|
+ ordered = list(
|
|
|
+ Region.objects.filter(slug__endswith='-sp')
|
|
|
+ .order_by('sort_path')
|
|
|
+ .values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ # Tree-flatten with siblings in name order:
|
|
|
+ # Aardvark, Buffalo (parent), Apex (child), Zoo (child), Zebra
|
|
|
+ self.assertEqual(ordered, ['Aardvark', 'Buffalo', 'Apex', 'Zoo', 'Zebra'])
|
|
|
+
|
|
|
+ def test_default_ordering_is_sort_path(self):
|
|
|
+ """Region.objects.all() uses sort_path-based ordering by default."""
|
|
|
+ Region.objects.create(name='B', slug='b-default')
|
|
|
+ Region.objects.create(name='A', slug='a-default')
|
|
|
+ names = list(
|
|
|
+ Region.objects.filter(slug__endswith='-default').values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ self.assertEqual(names, ['A', 'B'])
|
|
|
+
|
|
|
+ def test_get_descendants_returns_siblings_in_name_order(self):
|
|
|
+ """
|
|
|
+ For models with a sort_path column, get_descendants() must return
|
|
|
+ descendants with siblings in name order (matching MPTT's
|
|
|
+ order_insertion_by behavior), not insertion/PK order.
|
|
|
+ """
|
|
|
+ root = Region.objects.create(name='Root', slug='root-gdord')
|
|
|
+ # Insert siblings out of name order
|
|
|
+ Region.objects.create(parent=root, name='Zebra', slug='z-gdord')
|
|
|
+ Region.objects.create(parent=root, name='Aardvark', slug='a-gdord')
|
|
|
+ Region.objects.create(parent=root, name='Buffalo', slug='b-gdord')
|
|
|
+ names = list(root.get_descendants().values_list('name', flat=True))
|
|
|
+ self.assertEqual(names, ['Aardvark', 'Buffalo', 'Zebra'])
|
|
|
+
|
|
|
+ def test_get_children_returns_in_name_order(self):
|
|
|
+ root = Region.objects.create(name='Root', slug='root-gcord')
|
|
|
+ Region.objects.create(parent=root, name='Zebra', slug='z-gcord')
|
|
|
+ Region.objects.create(parent=root, name='Aardvark', slug='a-gcord')
|
|
|
+ names = list(root.get_children().values_list('name', flat=True))
|
|
|
+ self.assertEqual(names, ['Aardvark', 'Zebra'])
|
|
|
+
|
|
|
+
|
|
|
+class AddRelatedCountTests(TestCase):
|
|
|
+ """add_related_count must cumulate across subtrees via path <@."""
|
|
|
+
|
|
|
+ def test_cumulative_fk_count(self):
|
|
|
+ root = Region.objects.create(name='R', slug='r-arc')
|
|
|
+ child = Region.objects.create(parent=root, name='C', slug='c-arc')
|
|
|
+ Site.objects.create(name='S1', slug='s1-arc', region=child)
|
|
|
+ Site.objects.create(name='S2', slug='s2-arc', region=root)
|
|
|
+
|
|
|
+ qs = Region.objects.add_related_count(
|
|
|
+ Region.objects.filter(slug__endswith='-arc'),
|
|
|
+ Site, 'region', 'site_count', cumulative=True,
|
|
|
+ )
|
|
|
+ counts = {r.name: r.site_count for r in qs}
|
|
|
+ # root sees both sites (direct + via child)
|
|
|
+ self.assertEqual(counts['R'], 2)
|
|
|
+ self.assertEqual(counts['C'], 1)
|
|
|
+
|
|
|
+ def test_cumulative_m2m_count(self):
|
|
|
+ """
|
|
|
+ Cumulative count over an M2M relation walks the subtree via
|
|
|
+ path <@, joining the through table by the correct columns.
|
|
|
+
|
|
|
+ Regression: previously the JOINs swapped m2m_column_name() and
|
|
|
+ m2m_reverse_name(), producing wrong counts on M2M relations.
|
|
|
+ """
|
|
|
+ root = ContactGroup.objects.create(name='Root', slug='root-m2m')
|
|
|
+ child = ContactGroup.objects.create(parent=root, name='Child', slug='child-m2m')
|
|
|
+ leaf = ContactGroup.objects.create(parent=child, name='Leaf', slug='leaf-m2m')
|
|
|
+ # 3 contacts spread across the subtree (one per node)
|
|
|
+ c_root = Contact.objects.create(name='CR')
|
|
|
+ c_child = Contact.objects.create(name='CC')
|
|
|
+ c_leaf = Contact.objects.create(name='CL')
|
|
|
+ c_root.groups.add(root)
|
|
|
+ c_child.groups.add(child)
|
|
|
+ c_leaf.groups.add(leaf)
|
|
|
+
|
|
|
+ qs = ContactGroup.objects.add_related_count(
|
|
|
+ ContactGroup.objects.filter(slug__endswith='-m2m'),
|
|
|
+ Contact, 'groups', 'contact_count', cumulative=True,
|
|
|
+ )
|
|
|
+ counts = {g.name: g.contact_count for g in qs}
|
|
|
+ self.assertEqual(counts['Root'], 3)
|
|
|
+ self.assertEqual(counts['Child'], 2)
|
|
|
+ self.assertEqual(counts['Leaf'], 1)
|
|
|
+
|
|
|
+ def test_noncumulative_fk_count(self):
|
|
|
+ """Non-cumulative FK count includes only rows pointing directly at the node."""
|
|
|
+ root = Region.objects.create(name='R', slug='r-ncfk')
|
|
|
+ child = Region.objects.create(parent=root, name='C', slug='c-ncfk')
|
|
|
+ Site.objects.create(name='S1', slug='s1-ncfk', region=child)
|
|
|
+ Site.objects.create(name='S2', slug='s2-ncfk', region=root)
|
|
|
+
|
|
|
+ qs = Region.objects.add_related_count(
|
|
|
+ Region.objects.filter(slug__endswith='-ncfk'),
|
|
|
+ Site, 'region', 'site_count', cumulative=False,
|
|
|
+ )
|
|
|
+ counts = {r.name: r.site_count for r in qs}
|
|
|
+ # Each node counts only its own directly-assigned sites (no subtree rollup).
|
|
|
+ self.assertEqual(counts['R'], 1)
|
|
|
+ self.assertEqual(counts['C'], 1)
|
|
|
+
|
|
|
+ def test_noncumulative_m2m_count(self):
|
|
|
+ """Non-cumulative M2M count includes only directly-assigned rows, not the subtree."""
|
|
|
+ root = ContactGroup.objects.create(name='Root', slug='root-ncm2m')
|
|
|
+ child = ContactGroup.objects.create(parent=root, name='Child', slug='child-ncm2m')
|
|
|
+ c_root = Contact.objects.create(name='CR-nc')
|
|
|
+ c_child = Contact.objects.create(name='CC-nc')
|
|
|
+ c_root.groups.add(root)
|
|
|
+ c_child.groups.add(child)
|
|
|
+
|
|
|
+ qs = ContactGroup.objects.add_related_count(
|
|
|
+ ContactGroup.objects.filter(slug__endswith='-ncm2m'),
|
|
|
+ Contact, 'groups', 'contact_count', cumulative=False,
|
|
|
+ )
|
|
|
+ counts = {g.name: g.contact_count for g in qs}
|
|
|
+ self.assertEqual(counts['Root'], 1)
|
|
|
+ self.assertEqual(counts['Child'], 1)
|
|
|
+
|
|
|
+
|
|
|
+class SaveUpdateFieldsTests(TestCase):
|
|
|
+ """
|
|
|
+ Regression: when save(update_fields=...) excludes parent, _loaded_parent_id
|
|
|
+ must not advance, otherwise a subsequent full save() will mis-detect the
|
|
|
+ parent change as already-applied and leave path stale in memory.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_partial_save_then_full_save_refreshes_path(self):
|
|
|
+ r1 = Region.objects.create(name='R1', slug='r1-uf')
|
|
|
+ r2 = Region.objects.create(name='R2', slug='r2-uf')
|
|
|
+ obj = Region.objects.create(name='Obj', slug='obj-uf')
|
|
|
+ original_path = obj.path
|
|
|
+
|
|
|
+ # Reparent in memory but persist a different field only:
|
|
|
+ obj.parent = r1
|
|
|
+ obj.name = 'Obj-renamed'
|
|
|
+ obj.save(update_fields=['name'])
|
|
|
+
|
|
|
+ # DB parent_id is still NULL — confirm:
|
|
|
+ db_parent = Region.objects.values_list('parent_id', flat=True).get(pk=obj.pk)
|
|
|
+ self.assertIsNone(db_parent)
|
|
|
+ self.assertEqual(
|
|
|
+ Region.objects.values_list('path', flat=True).get(pk=obj.pk),
|
|
|
+ original_path,
|
|
|
+ )
|
|
|
+
|
|
|
+ # Now a full save persists the new parent — path must refresh:
|
|
|
+ obj.parent = r2
|
|
|
+ obj.save()
|
|
|
+ db_path = Region.objects.values_list('path', flat=True).get(pk=obj.pk)
|
|
|
+ self.assertEqual(db_path, _path(r2.pk, obj.pk))
|
|
|
+ self.assertEqual(obj.path, db_path, "in-memory path is stale after full save")
|
|
|
+
|
|
|
+
|
|
|
+class BulkCreateOrderingTests(TestCase):
|
|
|
+ """
|
|
|
+ The BEFORE INSERT trigger looks up parent.path via subquery per row.
|
|
|
+ In bulk_create the trigger fires per row in list order, so a parent
|
|
|
+ placed in the same batch must precede its children.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_parent_before_child_in_same_batch(self):
|
|
|
+ root = Region.objects.create(name='R', slug='r-bcord')
|
|
|
+ # Parent BEFORE child in the list — both get correct paths
|
|
|
+ parent_pending = Region(parent=root, name='Mid', slug='mid-bcord')
|
|
|
+ child_pending = Region(parent=parent_pending, name='Leaf', slug='leaf-bcord')
|
|
|
+ # parent_pending isn't yet saved, so child_pending.parent_id is None;
|
|
|
+ # set the parent reference after the parent is saved.
|
|
|
+ Region.objects.bulk_create([parent_pending])
|
|
|
+ parent_pending.refresh_from_db()
|
|
|
+ child_pending.parent = parent_pending
|
|
|
+ Region.objects.bulk_create([child_pending])
|
|
|
+ child_pending.refresh_from_db()
|
|
|
+ self.assertEqual(child_pending.path, _path(root.pk, parent_pending.pk, child_pending.pk))
|
|
|
+
|
|
|
+ def test_bulk_create_rejects_child_before_parent_in_same_batch(self):
|
|
|
+ """The guard refuses misordered batches instead of writing bad paths."""
|
|
|
+ unsaved_parent = Region(name='P', slug='p-bcrej')
|
|
|
+ unsaved_child = Region(parent=unsaved_parent, name='C', slug='c-bcrej')
|
|
|
+ with self.assertRaises(ValueError) as ctx:
|
|
|
+ Region.objects.bulk_create([unsaved_child, unsaved_parent])
|
|
|
+ self.assertIn('unsaved parent', str(ctx.exception))
|
|
|
+
|
|
|
+ def test_bulk_create_rejects_unsaved_parent_earlier_in_batch(self):
|
|
|
+ """
|
|
|
+ An unsaved parent placed earlier in the batch must also be rejected:
|
|
|
+ Django binds VALUES from the child's parent_id BEFORE the parent's
|
|
|
+ RETURNING-assigned pk lands, so the child would be inserted with
|
|
|
+ parent_id=NULL and silently stored as a root.
|
|
|
+ """
|
|
|
+ unsaved_parent = Region(name='P', slug='p-bcearly')
|
|
|
+ unsaved_child = Region(parent=unsaved_parent, name='C', slug='c-bcearly')
|
|
|
+ with self.assertRaises(ValueError) as ctx:
|
|
|
+ Region.objects.bulk_create([unsaved_parent, unsaved_child])
|
|
|
+ self.assertIn('unsaved parent', str(ctx.exception))
|
|
|
+ # Nothing should have been written.
|
|
|
+ self.assertFalse(Region.objects.filter(slug__in=('p-bcearly', 'c-bcearly')).exists())
|
|
|
+
|
|
|
+ def test_bulk_create_rejects_unsaved_parent_not_in_batch(self):
|
|
|
+ """
|
|
|
+ An unsaved parent that isn't in the batch must be rejected with a clear
|
|
|
+ ValueError; otherwise the child's parent_id is None at INSERT time and
|
|
|
+ the BEFORE trigger silently stores the row as a root.
|
|
|
+ """
|
|
|
+ external_parent = Region(name='X', slug='x-bcok') # not in the batch
|
|
|
+ with self.assertRaises(ValueError) as ctx:
|
|
|
+ Region.objects.bulk_create([Region(parent=external_parent, name='Y', slug='y-bcok')])
|
|
|
+ self.assertIn('unsaved parent', str(ctx.exception))
|
|
|
+ # Nothing should have been written.
|
|
|
+ self.assertFalse(Region.objects.filter(slug='y-bcok').exists())
|
|
|
+
|
|
|
+
|
|
|
+class TreeNodeFilterTests(TestCase):
|
|
|
+ """
|
|
|
+ Regression: the FK branch of TreeNodeFilter.filter() previously destructured
|
|
|
+ q_filter.children as (str, value) tuples, which crashed for compound Q
|
|
|
+ match types (DESCENDANTS, ANCESTORS, SIBLINGS, SELF_AND_DESCENDANTS). The
|
|
|
+ FK branch must resolve via __in like the M2M / M2O branches.
|
|
|
+ """
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def setUpTestData(cls):
|
|
|
+ from tenancy.models import Tenant, TenantGroup
|
|
|
+ cls.Tenant = Tenant
|
|
|
+ cls.TenantGroup = TenantGroup
|
|
|
+ cls.root = TenantGroup.objects.create(name='Root', slug='root-tnf')
|
|
|
+ cls.mid = TenantGroup.objects.create(parent=cls.root, name='Mid', slug='mid-tnf')
|
|
|
+ cls.leaf = TenantGroup.objects.create(parent=cls.mid, name='Leaf', slug='leaf-tnf')
|
|
|
+ cls.sibling = TenantGroup.objects.create(parent=cls.root, name='Sibling', slug='sibling-tnf')
|
|
|
+
|
|
|
+ cls.t_root = Tenant.objects.create(name='TRoot', slug='troot-tnf', group=cls.root)
|
|
|
+ cls.t_mid = Tenant.objects.create(name='TMid', slug='tmid-tnf', group=cls.mid)
|
|
|
+ cls.t_leaf = Tenant.objects.create(name='TLeaf', slug='tleaf-tnf', group=cls.leaf)
|
|
|
+ cls.t_sibling = Tenant.objects.create(name='TSib', slug='tsib-tnf', group=cls.sibling)
|
|
|
+
|
|
|
+ def _filter(self, match_type):
|
|
|
+ from netbox.graphql.filter_lookups import TreeNodeFilter, TreeNodeMatch
|
|
|
+ tnf = TreeNodeFilter(id=self.mid.pk, match_type=getattr(TreeNodeMatch, match_type))
|
|
|
+ # `filter` is decorated by @strawberry_django.filter_field; its wrapper
|
|
|
+ # asserts info is not None. The undecorated body is on `_unbound_wrapped_func`.
|
|
|
+ inner = TreeNodeFilter.filter._unbound_wrapped_func
|
|
|
+ qs, q = inner(tnf, info=None, queryset=self.Tenant.objects.all(), prefix='group__')
|
|
|
+ return list(qs.filter(q).values_list('name', flat=True))
|
|
|
+
|
|
|
+ def test_descendants_strict(self):
|
|
|
+ # DESCENDANTS of `mid` = leaf only (Mid itself excluded)
|
|
|
+ names = self._filter('DESCENDANTS')
|
|
|
+ self.assertEqual(sorted(names), ['TLeaf'])
|
|
|
+
|
|
|
+ def test_self_and_descendants(self):
|
|
|
+ names = self._filter('SELF_AND_DESCENDANTS')
|
|
|
+ self.assertEqual(sorted(names), ['TLeaf', 'TMid'])
|
|
|
+
|
|
|
+ def test_ancestors(self):
|
|
|
+ names = self._filter('ANCESTORS')
|
|
|
+ # Ancestors of Mid = Root only (Mid itself excluded)
|
|
|
+ self.assertEqual(sorted(names), ['TRoot'])
|
|
|
+
|
|
|
+ def test_siblings(self):
|
|
|
+ # Siblings of Mid (within Root) = Sibling
|
|
|
+ names = self._filter('SIBLINGS')
|
|
|
+ self.assertEqual(sorted(names), ['TSib'])
|
|
|
+
|
|
|
+
|
|
|
+class DescendantLookupSemanticsTests(TestCase):
|
|
|
+ """
|
|
|
+ path__descendant is strict (path <@ rhs AND path != rhs); the inclusive
|
|
|
+ form is path__descendant_or_equal. Previously both were inclusive.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_strict_descendant_excludes_self(self):
|
|
|
+ root = Region.objects.create(name='Root', slug='root-dls')
|
|
|
+ Region.objects.create(parent=root, name='Kid', slug='kid-dls')
|
|
|
+ strict = list(
|
|
|
+ Region.objects.filter(path__descendant=root.path)
|
|
|
+ .values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ self.assertEqual(sorted(strict), ['Kid'])
|
|
|
+ inclusive = list(
|
|
|
+ Region.objects.filter(path__descendant_or_equal=root.path)
|
|
|
+ .values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ self.assertEqual(sorted(inclusive), ['Kid', 'Root'])
|
|
|
+
|
|
|
+
|
|
|
+class AncestorLookupSemanticsTests(TestCase):
|
|
|
+ """
|
|
|
+ path__ancestor is strict (path @> rhs AND path != rhs); the inclusive form
|
|
|
+ is path__ancestor_or_equal.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_strict_ancestor_excludes_self(self):
|
|
|
+ root = Region.objects.create(name='Root', slug='root-als')
|
|
|
+ kid = Region.objects.create(parent=root, name='Kid', slug='kid-als')
|
|
|
+ strict = list(
|
|
|
+ Region.objects.filter(path__ancestor=kid.path)
|
|
|
+ .values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ self.assertEqual(sorted(strict), ['Root'])
|
|
|
+ inclusive = list(
|
|
|
+ Region.objects.filter(path__ancestor_or_equal=kid.path)
|
|
|
+ .values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ self.assertEqual(sorted(inclusive), ['Kid', 'Root'])
|
|
|
+
|
|
|
+
|
|
|
+class RenameCascadesSortPathTests(TestCase):
|
|
|
+ """
|
|
|
+ Renaming a node updates its own sort_path AND cascades into descendants'
|
|
|
+ sort_paths via the AFTER trigger. (Diverges from MPTT order_insertion_by,
|
|
|
+ which leaves descendants stale until manual rebuild.)
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_rename_cascades_into_descendants(self):
|
|
|
+ parent = Region.objects.create(name='Bravo', slug='bravo-rcsp')
|
|
|
+ mid = Region.objects.create(parent=parent, name='Mid', slug='mid-rcsp')
|
|
|
+ leaf = Region.objects.create(parent=mid, name='Leaf', slug='leaf-rcsp')
|
|
|
+
|
|
|
+ parent.name = 'Zulu'
|
|
|
+ parent.save()
|
|
|
+
|
|
|
+ parent.refresh_from_db()
|
|
|
+ mid.refresh_from_db()
|
|
|
+ leaf.refresh_from_db()
|
|
|
+ self.assertEqual(parent.sort_path, 'Zulu')
|
|
|
+ self.assertEqual(mid.sort_path, f'Zulu{chr(9)}Mid')
|
|
|
+ self.assertEqual(leaf.sort_path, f'Zulu{chr(9)}Mid{chr(9)}Leaf')
|
|
|
+ # Paths unchanged — only sort_path moved.
|
|
|
+ self.assertEqual(mid.path, _path(parent.pk, mid.pk))
|
|
|
+ self.assertEqual(leaf.path, _path(parent.pk, mid.pk, leaf.pk))
|
|
|
+
|
|
|
+ def test_rename_does_not_affect_unrelated_subtree(self):
|
|
|
+ # Two roots; renaming one must not touch the other's sort_path.
|
|
|
+ a = Region.objects.create(name='AA', slug='aa-iso')
|
|
|
+ Region.objects.create(parent=a, name='AKid', slug='akid-iso')
|
|
|
+ b = Region.objects.create(name='BB', slug='bb-iso')
|
|
|
+ b_kid = Region.objects.create(parent=b, name='BKid', slug='bkid-iso')
|
|
|
+
|
|
|
+ a.name = 'AAren'
|
|
|
+ a.save()
|
|
|
+
|
|
|
+ b_kid.refresh_from_db()
|
|
|
+ self.assertEqual(b_kid.sort_path, f'BB{chr(9)}BKid')
|
|
|
+
|
|
|
+
|
|
|
+class RebuildSortPathsTests(TestCase):
|
|
|
+ """rebuild_sort_paths() is still available for repair after raw SQL writes."""
|
|
|
+
|
|
|
+ def test_rebuild_after_raw_update(self):
|
|
|
+ parent = Region.objects.create(name='Bravo', slug='bravo-rsp')
|
|
|
+ mid = Region.objects.create(parent=parent, name='Mid', slug='mid-rsp')
|
|
|
+ # Raw .update() with only the name column bypasses the BEFORE trigger
|
|
|
+ # (its column list is parent_id + name, but the trigger is keyed on
|
|
|
+ # name in the SET clause). Actually update() on `name` DOES fire the
|
|
|
+ # trigger now — so to simulate a bypass we corrupt sort_path directly.
|
|
|
+ Region.objects.filter(pk=parent.pk).update(sort_path='garbage')
|
|
|
+ Region.objects.filter(pk=mid.pk).update(sort_path='also-garbage')
|
|
|
+
|
|
|
+ Region.rebuild_sort_paths()
|
|
|
+ parent.refresh_from_db()
|
|
|
+ mid.refresh_from_db()
|
|
|
+ self.assertEqual(parent.sort_path, 'Bravo')
|
|
|
+ self.assertEqual(mid.sort_path, f'Bravo{chr(9)}Mid')
|
|
|
+
|
|
|
+ def test_raises_without_sort_path(self):
|
|
|
+ # InventoryItem uses LtreeModel but doesn't have a sort_path column.
|
|
|
+ from dcim.models import InventoryItem
|
|
|
+ with self.assertRaises(NotImplementedError):
|
|
|
+ InventoryItem.rebuild_sort_paths()
|
|
|
+
|
|
|
+
|
|
|
+class SortPathRefreshTests(TestCase):
|
|
|
+ """
|
|
|
+ save() must refresh the in-memory `sort_path` (not just `path`) after insert
|
|
|
+ and reparent, so callers (notably change logging) never snapshot a stale value.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_create_refreshes_sort_path(self):
|
|
|
+ root = Region.objects.create(name='Root', slug='root-spr')
|
|
|
+ child = Region.objects.create(name='Kid', slug='kid-spr', parent=root)
|
|
|
+ db_sort_path = Region.objects.values_list('sort_path', flat=True).get(pk=child.pk)
|
|
|
+ self.assertEqual(child.sort_path, db_sort_path)
|
|
|
+ self.assertEqual(child.sort_path, f'Root{chr(9)}Kid')
|
|
|
+
|
|
|
+ def test_reparent_refreshes_sort_path(self):
|
|
|
+ a = Region.objects.create(name='Alpha', slug='alpha-spr')
|
|
|
+ b = Region.objects.create(name='Bravo', slug='bravo-spr')
|
|
|
+ child = Region.objects.create(name='Kid', slug='kid2-spr', parent=a)
|
|
|
+ child.parent = b
|
|
|
+ child.save()
|
|
|
+ db_sort_path = Region.objects.values_list('sort_path', flat=True).get(pk=child.pk)
|
|
|
+ self.assertEqual(child.sort_path, db_sort_path)
|
|
|
+ self.assertEqual(child.sort_path, f'Bravo{chr(9)}Kid')
|
|
|
+
|
|
|
+ def test_rename_refreshes_sort_path(self):
|
|
|
+ # The trigger rewrites sort_path on a name change; the in-memory instance
|
|
|
+ # must reflect it without a manual refresh_from_db().
|
|
|
+ root = Region.objects.create(name='Root', slug='root-rn')
|
|
|
+ root.name = 'Renamed'
|
|
|
+ root.save()
|
|
|
+ db_sort_path = Region.objects.values_list('sort_path', flat=True).get(pk=root.pk)
|
|
|
+ self.assertEqual(root.sort_path, db_sort_path)
|
|
|
+ self.assertEqual(root.sort_path, 'Renamed')
|
|
|
+
|
|
|
+
|
|
|
+class ReapplyModelOrderingTests(TestCase):
|
|
|
+ """Ltree-backed models must be exempt from reapply_model_ordering()."""
|
|
|
+
|
|
|
+ def test_ltree_model_is_exempt(self):
|
|
|
+ from utilities.query import reapply_model_ordering
|
|
|
+ # Clear ordering so a non-exempt model would be re-ordered by Meta.ordering.
|
|
|
+ qs = Region.objects.all().order_by()
|
|
|
+ result = reapply_model_ordering(qs)
|
|
|
+ # The LtreeManager is inherited from the abstract base (not in
|
|
|
+ # local_managers), so the exemption must still apply and return qs as-is.
|
|
|
+ self.assertIs(result, qs)
|
|
|
+
|
|
|
+ def test_mptt_model_detection(self):
|
|
|
+ # MPTT support is retained for plugins (NestedGroupModel stays MPTT-backed);
|
|
|
+ # reapply_model_ordering() must keep exempting such models. See _is_mptt_model.
|
|
|
+ # TODO: Remove in NetBox v5.0 alongside MPTT support.
|
|
|
+ from netbox.models import NestedGroupModel
|
|
|
+ from utilities.query import _is_mptt_model
|
|
|
+
|
|
|
+ self.assertTrue(_is_mptt_model(NestedGroupModel))
|
|
|
+ # An ltree-backed model must NOT be misdetected as MPTT.
|
|
|
+ self.assertFalse(_is_mptt_model(Region))
|
|
|
+
|
|
|
+
|
|
|
+class AddRelatedCountErrorTests(TestCase):
|
|
|
+ """
|
|
|
+ add_related_count() must not raise at queryset-build time for unresolvable
|
|
|
+ rel_field — many call sites bind the annotation as a class attribute at
|
|
|
+ module import. The annotation is still attached using the Django default
|
|
|
+ column naming, and any error surfaces at evaluation time.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_unknown_field_does_not_raise_at_build(self):
|
|
|
+ qs = Region.objects.add_related_count(
|
|
|
+ Region.objects.all(), Region, 'not_a_field', 'bogus_count', cumulative=True
|
|
|
+ )
|
|
|
+ self.assertIn('bogus_count', qs.query.annotations)
|
|
|
+
|
|
|
+
|
|
|
+class ChangeLogExclusionTests(TestCase):
|
|
|
+ """
|
|
|
+ Trigger-maintained columns (`path`, `sort_path`) must be excluded from change
|
|
|
+ log diffs, and the postchange snapshot must capture the refreshed values.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_sort_path_excluded_from_diff(self):
|
|
|
+ oc = ObjectChange()
|
|
|
+ oc.changed_object_type = ContentType.objects.get_for_model(Region)
|
|
|
+ self.assertIn('path', oc.diff_exclude_fields)
|
|
|
+ self.assertIn('sort_path', oc.diff_exclude_fields)
|
|
|
+
|
|
|
+ def test_reparent_postchange_snapshot_matches_db(self):
|
|
|
+ a = Region.objects.create(name='Alpha', slug='alpha-cl')
|
|
|
+ b = Region.objects.create(name='Bravo', slug='bravo-cl')
|
|
|
+ child = Region.objects.create(name='Kid', slug='kid-cl', parent=a)
|
|
|
+ # Reload so the prechange snapshot reflects the persisted state.
|
|
|
+ child = Region.objects.get(pk=child.pk)
|
|
|
+ child.snapshot()
|
|
|
+ child.parent = b
|
|
|
+ child.save()
|
|
|
+ oc = child.to_objectchange('update')
|
|
|
+ db = Region.objects.values('path', 'sort_path').get(pk=child.pk)
|
|
|
+ self.assertEqual(oc.postchange_data['path'], db['path'])
|
|
|
+ self.assertEqual(oc.postchange_data['sort_path'], db['sort_path'])
|
|
|
+ # path/sort_path are excluded, so they must not surface in the cleaned diff data.
|
|
|
+ self.assertNotIn('sort_path', oc.postchange_data_clean)
|
|
|
+ self.assertNotIn('path', oc.postchange_data_clean)
|
|
|
+
|
|
|
+
|
|
|
+class MPTTChangeLogExclusionTests(TestCase):
|
|
|
+ """
|
|
|
+ ObjectChange.diff_exclude_fields must hide MPTT bookkeeping columns
|
|
|
+ (lft/rght/tree_id/level) for plugin models still using the deprecated
|
|
|
+ MPTT-backed NestedGroupModel, in addition to ltree's path/sort_path.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_diff_exclude_fields_for_mptt_subclass(self):
|
|
|
+ from unittest.mock import MagicMock
|
|
|
+
|
|
|
+ from mptt.models import MPTTModel
|
|
|
+
|
|
|
+ class _FakeMPTT(MPTTModel):
|
|
|
+ class Meta:
|
|
|
+ abstract = True
|
|
|
+ app_label = 'tests'
|
|
|
+
|
|
|
+ oc = ObjectChange()
|
|
|
+ fake_ct = MagicMock()
|
|
|
+ fake_ct.model_class.return_value = _FakeMPTT
|
|
|
+ # Prime the FK descriptor's cache so accessing changed_object_type
|
|
|
+ # returns our mock without hitting the DB or invoking type checks.
|
|
|
+ ObjectChange._meta.get_field('changed_object_type').set_cached_value(oc, fake_ct)
|
|
|
+
|
|
|
+ excluded = oc.diff_exclude_fields
|
|
|
+ self.assertIn('lft', excluded)
|
|
|
+ self.assertIn('rght', excluded)
|
|
|
+ self.assertIn('tree_id', excluded)
|
|
|
+ self.assertIn('level', excluded)
|
|
|
+
|
|
|
+
|
|
|
+class AddRelatedCountResilienceTests(TestCase):
|
|
|
+ """
|
|
|
+ add_related_count() must not raise FieldDoesNotExist at queryset-build
|
|
|
+ time so that view modules (which bind it as a class attribute) can be
|
|
|
+ imported even if a referenced field has been renamed.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_unknown_field_does_not_raise(self):
|
|
|
+ # Bare manager call equivalent to a view class body using a stale name.
|
|
|
+ qs = Region.objects.add_related_count(
|
|
|
+ Region.objects.all(),
|
|
|
+ Region, # any model; the field name is what matters
|
|
|
+ 'this_field_does_not_exist',
|
|
|
+ 'noop_count',
|
|
|
+ cumulative=True,
|
|
|
+ )
|
|
|
+ # The annotation was attached; evaluating it would fail at the DB
|
|
|
+ # (column doesn't exist) but importing the view module must succeed.
|
|
|
+ self.assertIn('noop_count', qs.query.annotations)
|
|
|
+
|
|
|
+
|
|
|
+class CascadeTriggerScopeTests(TestCase):
|
|
|
+ """
|
|
|
+ The AFTER cascade trigger fires on parent_id or name changes. A rename
|
|
|
+ leaves `path` untouched but pushes the new sort_path into descendants.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_rename_preserves_descendant_path_but_updates_sort_path(self):
|
|
|
+ parent = Region.objects.create(name='Bravo', slug='bravo-ct')
|
|
|
+ child = Region.objects.create(parent=parent, name='Mid', slug='mid-ct')
|
|
|
+ original_child_path = child.path
|
|
|
+ parent.name = 'Zulu'
|
|
|
+ parent.save()
|
|
|
+ child.refresh_from_db()
|
|
|
+ # Path is unaffected by a rename; sort_path follows the new name.
|
|
|
+ self.assertEqual(child.path, original_child_path)
|
|
|
+ self.assertNotIn('Bravo', child.sort_path)
|
|
|
+ self.assertIn('Zulu', child.sort_path)
|
|
|
+
|
|
|
+
|
|
|
+class CycleGuardWithEmptyPathTests(TestCase):
|
|
|
+ """
|
|
|
+ _parent_creates_cycle must catch self-as-parent even when self.path is
|
|
|
+ empty or deferred — otherwise an instance constructed without a loaded
|
|
|
+ path can pass the Python guard and corrupt the tree.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_self_parent_rejected_when_path_is_empty(self):
|
|
|
+ from django.core.exceptions import ValidationError
|
|
|
+ a = Region.objects.create(name='A', slug='a-empty-cyc')
|
|
|
+ # Simulate a caller (script, plugin) that holds an instance whose `path`
|
|
|
+ # attribute was never loaded — e.g. via .only('id', 'parent_id').
|
|
|
+ # Empty-string path is the LtreeField default, so this mirrors what a
|
|
|
+ # deferred-or-unset path looks like at the Python layer.
|
|
|
+ a.path = ''
|
|
|
+ a.parent_id = a.pk
|
|
|
+ with self.assertRaises(ValidationError):
|
|
|
+ a.save()
|
|
|
+
|
|
|
+
|
|
|
+class TriggerCycleGuardTests(TestCase):
|
|
|
+ """
|
|
|
+ The BEFORE INSERT/UPDATE trigger must reject a parent_id assignment that
|
|
|
+ would form a cycle. The Python save()-time guard already catches this for
|
|
|
+ ordinary save(); the trigger backstops QuerySet.update / bulk_update /
|
|
|
+ raw UPDATEs that bypass save().
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_queryset_update_to_self_parent_blocked_by_trigger(self):
|
|
|
+ from django.db import IntegrityError, transaction
|
|
|
+ a = Region.objects.create(name='A', slug='a-tgcyc')
|
|
|
+ # IntegrityError leaves Django's transaction marked-for-rollback; wrap
|
|
|
+ # the failing UPDATE in a savepoint so the outer test transaction can
|
|
|
+ # continue to issue queries (refresh_from_db).
|
|
|
+ with self.assertRaises(IntegrityError) as ctx:
|
|
|
+ with transaction.atomic():
|
|
|
+ Region.objects.filter(pk=a.pk).update(parent_id=a.pk)
|
|
|
+ self.assertIn('cycle detected', str(ctx.exception))
|
|
|
+ a.refresh_from_db()
|
|
|
+ self.assertIsNone(a.parent_id)
|
|
|
+
|
|
|
+ def test_queryset_update_to_descendant_blocked_by_trigger(self):
|
|
|
+ from django.db import IntegrityError, transaction
|
|
|
+ a = Region.objects.create(name='A', slug='a-tgcyc2')
|
|
|
+ b = Region.objects.create(parent=a, name='B', slug='b-tgcyc2')
|
|
|
+ # Try to reparent A under B (B is A's descendant) via raw UPDATE.
|
|
|
+ with self.assertRaises(IntegrityError) as ctx:
|
|
|
+ with transaction.atomic():
|
|
|
+ Region.objects.filter(pk=a.pk).update(parent_id=b.pk)
|
|
|
+ self.assertIn('cycle detected', str(ctx.exception))
|
|
|
+ a.refresh_from_db()
|
|
|
+ self.assertIsNone(a.parent_id)
|
|
|
+
|
|
|
+ def test_legitimate_reparent_via_update_still_works(self):
|
|
|
+ a = Region.objects.create(name='A', slug='a-tgok')
|
|
|
+ b = Region.objects.create(name='B', slug='b-tgok')
|
|
|
+ child = Region.objects.create(parent=a, name='C', slug='c-tgok')
|
|
|
+ Region.objects.filter(pk=child.pk).update(parent_id=b.pk)
|
|
|
+ child.refresh_from_db()
|
|
|
+ self.assertEqual(child.parent_id, b.pk)
|
|
|
+ self.assertEqual(child.path, _path(b.pk, child.pk))
|
|
|
+
|
|
|
+ def test_queryset_update_mid_tree_to_own_descendant_blocked(self):
|
|
|
+ """
|
|
|
+ Reparenting a non-root node under one of its own (non-immediate)
|
|
|
+ descendants must raise. The earlier `lpad(NEW.id) @> parent_path`
|
|
|
+ check only fired when NEW was the root of parent_path, missing
|
|
|
+ mid-tree cycles entirely.
|
|
|
+ """
|
|
|
+ from django.db import IntegrityError, transaction
|
|
|
+ a = Region.objects.create(name='A', slug='a-midcyc')
|
|
|
+ b = Region.objects.create(parent=a, name='B', slug='b-midcyc')
|
|
|
+ c = Region.objects.create(parent=b, name='C', slug='c-midcyc')
|
|
|
+ # Try to make B a child of C — B is mid-tree (path A.B), parent_path is A.B.C.
|
|
|
+ with self.assertRaises(IntegrityError) as ctx:
|
|
|
+ with transaction.atomic():
|
|
|
+ Region.objects.filter(pk=b.pk).update(parent_id=c.pk)
|
|
|
+ self.assertIn('cycle detected', str(ctx.exception))
|
|
|
+ b.refresh_from_db()
|
|
|
+ self.assertEqual(b.parent_id, a.pk)
|
|
|
+
|
|
|
+ def test_queryset_update_mid_tree_self_loop_blocked(self):
|
|
|
+ """A non-root node assigning itself as parent must also raise."""
|
|
|
+ from django.db import IntegrityError, transaction
|
|
|
+ a = Region.objects.create(name='A', slug='a-midself')
|
|
|
+ b = Region.objects.create(parent=a, name='B', slug='b-midself')
|
|
|
+ with self.assertRaises(IntegrityError) as ctx:
|
|
|
+ with transaction.atomic():
|
|
|
+ Region.objects.filter(pk=b.pk).update(parent_id=b.pk)
|
|
|
+ self.assertIn('cycle detected', str(ctx.exception))
|
|
|
+ b.refresh_from_db()
|
|
|
+ self.assertEqual(b.parent_id, a.pk)
|
|
|
+
|
|
|
+
|
|
|
+class NaturalSortSortPathTests(TestCase):
|
|
|
+ """
|
|
|
+ Sort-path separator must produce correct tree-flatten ordering under the
|
|
|
+ `natural_sort` (ICU `und-u-kn-true`) collation used by ModuleBay,
|
|
|
+ TenantGroup, and WirelessLANGroup. chr(1) was variable-ignorable under
|
|
|
+ that collation and would interleave children with unrelated roots;
|
|
|
+ chr(9) (TAB) is treated non-variably and orders deterministically.
|
|
|
+ """
|
|
|
+
|
|
|
+ def test_chr9_separator_collates_below_letters_under_natural_sort(self):
|
|
|
+ # Direct collation probe — independent of any model schema.
|
|
|
+ with connection.cursor() as cur:
|
|
|
+ cur.execute("""
|
|
|
+ SELECT
|
|
|
+ ('A' || chr(9) || 'Z') COLLATE "natural_sort"
|
|
|
+ < 'AA' COLLATE "natural_sort"
|
|
|
+ """)
|
|
|
+ self.assertTrue(cur.fetchone()[0])
|
|
|
+
|
|
|
+ def test_tree_flatten_ordering_under_natural_sort(self):
|
|
|
+ # TenantGroup.sort_path uses natural_sort collation; build a small tree
|
|
|
+ # where chr(1) would have produced wrong sibling ordering. TenantGroup
|
|
|
+ # names are globally unique, so each row gets a distinct name. Under
|
|
|
+ # the old chr(1) separator the child's sort_path was variable-ignorable
|
|
|
+ # and collated AFTER the unrelated root 'nsP1'; chr(9) sorts strictly
|
|
|
+ # below digits/letters and keeps children clustered under their parent.
|
|
|
+ from tenancy.models import TenantGroup
|
|
|
+ parent = TenantGroup.objects.create(name='nsP', slug='nsp-ns')
|
|
|
+ TenantGroup.objects.create(parent=parent, name='nsPchild', slug='nspc-ns')
|
|
|
+ TenantGroup.objects.create(name='nsP1', slug='nsp1-ns') # unrelated root
|
|
|
+
|
|
|
+ names = list(
|
|
|
+ TenantGroup.objects.filter(slug__endswith='-ns')
|
|
|
+ .order_by('sort_path')
|
|
|
+ .values_list('name', flat=True)
|
|
|
+ )
|
|
|
+ # Expected tree-flatten: parent, its child, then the unrelated root.
|
|
|
+ self.assertEqual(names, ['nsP', 'nsPchild', 'nsP1'])
|
Arthur Hanson