netbox/netbox/dcim/tests/test_module_moves.py

import re
import signal
import uuid
from contextlib import contextmanager
from unittest.mock import patch

from django.apps import apps
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db import IntegrityError, OperationalError, connection, router, transaction
from django.db.models import QuerySet
from django.test import RequestFactory, TestCase, override_settings
from django.test.utils import CaptureQueriesContext

from circuits.models import Provider, ProviderNetwork, VirtualCircuit, VirtualCircuitTermination, VirtualCircuitType
from core.models import ObjectChange
from dcim.choices import InterfaceModeChoices, InterfaceTypeChoices, ModuleStatusChoices, PortTypeChoices
from dcim.models import (
    Cable,
    ConsolePortTemplate,
    ConsoleServerPortTemplate,
    CoolingIntake,
    CoolingIntakeTemplate,
    CoolingOutflow,
    CoolingOutflowTemplate,
    Device,
    DeviceBay,
    DeviceRole,
    DeviceType,
    FrontPort,
    FrontPortTemplate,
    Interface,
    InterfaceTemplate,
    InventoryItem,
    MACAddress,
    Manufacturer,
    Module,
    ModuleBay,
    ModuleBayTemplate,
    ModuleType,
    PortMapping,
    PortTemplateMapping,
    PowerOutlet,
    PowerOutletTemplate,
    PowerPort,
    PowerPortTemplate,
    RearPort,
    RearPortTemplate,
    Site,
    VirtualDeviceContext,
)
from dcim.models.device_components import ModularComponentModel
from dcim.models.module_moves import COMPONENT_TEMPLATE_ATTRS, ModuleMovePlan
from dcim.utils import get_module_bay_positions, resolve_module_placeholder
from ipam.choices import FHRPGroupProtocolChoices
from ipam.models import VLAN, VRF, FHRPGroup, FHRPGroupAssignment, IPAddress, VLANTranslationPolicy
from netbox.context_managers import event_tracking
from users.models import User
from utilities.exceptions import AbortRequest
from utilities.ordering import naturalize_interface
from utilities.testing import create_test_device
from vpn.choices import L2VPNTypeChoices, TunnelEncapsulationChoices
from vpn.models import L2VPN, L2VPNTermination, Tunnel, TunnelTermination
from wireless.models import WirelessLAN, WirelessLink


@contextmanager
def fail_after(seconds):
    """
    Fail the enclosed block if it runs longer than the given number of seconds. Backstop
    for the cycle-guard tests: a regressed hang fails one test fast instead of stalling the run.
    """
    def on_alarm(signum, frame):
        raise AssertionError(f'Operation did not complete within {seconds} seconds.')

    previous_handler = signal.signal(signal.SIGALRM, on_alarm)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, previous_handler)


class ModuleMoveValidationTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        cls.device_a = create_test_device('Device A')
        cls.device_b = create_test_device('Device B')
        cls.bays_a = [
            ModuleBay.objects.create(device=cls.device_a, name=f'Bay A{i}') for i in range(1, 4)
        ]
        cls.bays_b = [
            ModuleBay.objects.create(device=cls.device_b, name=f'Bay B{i}') for i in range(1, 3)
        ]
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type M')
        cls.other_module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type N')
        cls.module = Module.objects.create(
            device=cls.device_a, module_bay=cls.bays_a[0], module_type=cls.module_type
        )

    def test_create_in_occupied_bay_fails(self):
        module = Module(device=self.device_a, module_bay=self.bays_a[0], module_type=self.module_type)
        with self.assertRaises(ValidationError) as cm:
            module.full_clean()
        self.assertIn('module_bay', cm.exception.message_dict)

    def test_move_to_occupied_bay_fails(self):
        occupant = Module.objects.create(
            device=self.device_a, module_bay=self.bays_a[1], module_type=self.module_type
        )
        self.module.module_bay = self.bays_a[1]
        with self.assertRaises(ValidationError) as cm:
            self.module.full_clean()
        self.assertIn('module_bay', cm.exception.message_dict)
        self.assertIn(str(occupant), str(cm.exception.message_dict['module_bay']))

    def test_move_to_own_bay_passes(self):
        self.module.serial = 'ABC123'
        self.module.full_clean()

    def test_move_to_other_device_bay_without_device_change_fails(self):
        # Existing device/bay consistency rule must keep rejecting a half-specified move
        self.module.module_bay = self.bays_b[0]
        with self.assertRaises(ValidationError):
            self.module.full_clean()

    def test_move_with_module_type_change_fails(self):
        self.module.module_bay = self.bays_a[1]
        self.module.module_type = self.other_module_type
        with self.assertRaises(ValidationError) as cm:
            self.module.full_clean()
        self.assertIn('module_type', cm.exception.message_dict)

    def test_module_type_change_without_move_passes(self):
        self.module.module_type = self.other_module_type
        self.module.full_clean()
        self.module.save()
        self.module.refresh_from_db()
        self.assertEqual(self.module.module_type, self.other_module_type)


class ModuleSameDeviceMoveTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        cls.device = create_test_device('Device A')
        cls.bay_a = ModuleBay.objects.create(device=cls.device, name='Bay A', position='A')
        cls.bay_b = ModuleBay.objects.create(device=cls.device, name='Bay B', position='B')
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type M')
        cls.module = Module.objects.create(
            device=cls.device, module_bay=cls.bay_a, module_type=cls.module_type
        )
        cls.interface = Interface.objects.create(
            device=cls.device, module=cls.module, name='eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        cls.child_bay = ModuleBay.objects.create(
            device=cls.device, module=cls.module, name='Child Bay 1'
        )

    def test_move_to_empty_bay_updates_module_bay(self):
        self.module.module_bay = self.bay_b
        self.module.full_clean()
        self.module.save()
        self.module.refresh_from_db()
        self.assertEqual(self.module.module_bay, self.bay_b)

    def test_move_reparents_direct_child_bays(self):
        self.module.module_bay = self.bay_b
        self.module.save()
        self.child_bay.refresh_from_db()
        self.assertEqual(self.child_bay.parent_id, self.bay_b.pk)
        self.assertTrue(str(self.child_bay.path).startswith(f'{self.bay_b.path}.'))

    def test_move_keeps_components_untouched_without_templates(self):
        self.module.module_bay = self.bay_b
        self.module.save()
        self.interface.refresh_from_db()
        self.assertEqual(self.interface.name, 'eth0')
        self.assertEqual(self.interface.device, self.device)

    def test_save_without_move_skips_planner(self):
        with patch('dcim.models.modules.ModuleMovePlan.from_module') as mock_plan:
            self.module.serial = 'XYZ789'
            self.module.save()
        mock_plan.assert_not_called()

    def test_save_level_move_with_type_change_raises_abort_request(self):
        other_type = ModuleType.objects.create(
            manufacturer=self.module_type.manufacturer, model='Module Type N'
        )
        self.module.module_bay = self.bay_b
        self.module.module_type = other_type
        with self.assertRaises(AbortRequest):
            self.module.save()


class ModuleSaveRoutingTestCase(TestCase):
    """Pins save() routing against concurrent, stale, deferred, and preset-pk instances."""

    @classmethod
    def setUpTestData(cls):
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        cls.device = create_test_device('Device A')
        cls.device_b = create_test_device('Device B')
        cls.bay_a = ModuleBay.objects.create(device=cls.device, name='Bay A')
        cls.bay_b = ModuleBay.objects.create(device=cls.device, name='Bay B')
        cls.bay_c = ModuleBay.objects.create(device=cls.device_b, name='Bay C')
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type M')
        InterfaceTemplate.objects.create(
            module_type=cls.module_type, name='eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        cls.module = Module.objects.create(
            device=cls.device, module_bay=cls.bay_a, module_type=cls.module_type
        )
        cls.child_bay = ModuleBay.objects.create(device=cls.device, module=cls.module, name='Child Bay 1')

    def _create_nested_subtree(self):
        child_module = Module(
            device=self.device,
            module_bay=self.child_bay,
            module_type=self.module_type,
        )
        child_module._disable_replication = True
        child_module.save()
        root_interface = Interface.objects.get(module=self.module, name='eth0')
        child_interface = Interface.objects.create(
            device=self.device,
            module=child_module,
            name='child0',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        return child_module, root_interface, child_interface

    @contextmanager
    def _move_to_other_device_before_locked_save(self):
        """Complete a competing move after save() routes the write but before the root lock."""
        original_save_existing = Module._save_existing
        moved = False

        def save_existing(instance, *args, **kwargs):
            nonlocal moved
            if not moved:
                moved = True
                concurrent = Module.objects.get(pk=self.module.pk)
                concurrent.device = self.device_b
                concurrent.module_bay = self.bay_c
                concurrent.save()
            return original_save_existing(instance, *args, **kwargs)

        with patch.object(Module, '_save_existing', autospec=True, side_effect=save_existing):
            yield

        self.assertTrue(moved)

    def test_plain_save_uses_locked_placement_for_routing(self):
        """A competing move landing before the root lock is resolved by the planner, not clobbered."""
        child_module, root_interface, child_interface = self._create_nested_subtree()
        stale = Module.objects.get(pk=self.module.pk)
        stale.comments = 'Stale session edit'

        with self._move_to_other_device_before_locked_save():
            stale.save()

        self.module.refresh_from_db()
        child_module.refresh_from_db()
        self.child_bay.refresh_from_db()
        root_interface.refresh_from_db()
        child_interface.refresh_from_db()
        self.assertEqual((self.module.device_id, self.module.module_bay_id), (self.device.pk, self.bay_a.pk))
        self.assertEqual(child_module.device_id, self.device.pk)
        self.assertEqual((self.child_bay.device_id, self.child_bay.parent_id), (self.device.pk, self.bay_a.pk))
        self.assertEqual(root_interface.device_id, self.device.pk)
        self.assertEqual(child_interface.device_id, self.device.pk)
        self.assertEqual(self.module.comments, 'Stale session edit')

    def test_update_fields_are_checked_against_locked_placement(self):
        """update_fields completeness is re-judged against the placement a competing move just committed."""
        child_module, root_interface, child_interface = self._create_nested_subtree()
        module = Module.objects.get(pk=self.module.pk)
        module.module_bay = self.bay_b

        with self._move_to_other_device_before_locked_save():
            with self.assertRaises(AbortRequest):
                module.save(update_fields=['module_bay'])

        self.module.refresh_from_db()
        child_module.refresh_from_db()
        self.child_bay.refresh_from_db()
        root_interface.refresh_from_db()
        child_interface.refresh_from_db()
        self.assertEqual((self.module.device_id, self.module.module_bay_id), (self.device_b.pk, self.bay_c.pk))
        self.assertEqual(child_module.device_id, self.device_b.pk)
        self.assertEqual((self.child_bay.device_id, self.child_bay.parent_id), (self.device_b.pk, self.bay_c.pk))
        self.assertEqual(root_interface.device_id, self.device_b.pk)
        self.assertEqual(child_interface.device_id, self.device_b.pk)

    def test_stale_instance_plain_save_after_move_keeps_children_consistent(self):
        instance_a = Module.objects.get(pk=self.module.pk)
        instance_b = Module.objects.get(pk=self.module.pk)
        instance_b.module_bay = self.bay_b
        instance_b.save()

        instance_a.comments = 'Stale session edit'
        instance_a.save()

        self.module.refresh_from_db()
        self.child_bay.refresh_from_db()
        self.assertEqual(self.module.comments, 'Stale session edit')
        self.assertEqual(self.child_bay.parent_id, self.module.module_bay_id)
        self.assertTrue(str(self.child_bay.path).startswith(f'{self.module.module_bay.path}.'))

    def test_deferred_device_field_save_with_update_fields_succeeds(self):
        module = Module.objects.defer('device').get(pk=self.module.pk)
        module.status = ModuleStatusChoices.STATUS_OFFLINE
        module.save(update_fields=['status'])
        module.refresh_from_db()
        self.assertEqual(module.status, ModuleStatusChoices.STATUS_OFFLINE)
        self.assertEqual(module.module_bay, self.bay_a)

    def test_out_of_band_placement_change_then_refresh_then_save_skips_planner(self):
        # Routing always enters the locked read; a refreshed instance shows no delta there, so no plan is built.
        Module.objects.filter(pk=self.module.pk).update(module_bay=self.bay_b)
        self.module.refresh_from_db()
        with patch.object(ModuleMovePlan, 'from_module') as mock_plan:
            self.module.serial = 'REFRESHED1'
            self.module.save()
        mock_plan.assert_not_called()
        self.module.refresh_from_db()
        self.assertEqual(self.module.serial, 'REFRESHED1')
        self.assertEqual(self.module.module_bay, self.bay_b)

    def test_preset_pk_save_with_unchanged_placement_does_not_replicate_components(self):
        interface_count_before = Interface.objects.filter(module=self.module).count()
        Module(
            pk=self.module.pk, device=self.module.device, module_bay=self.module.module_bay,
            module_type=self.module.module_type,
        ).save()
        self.assertEqual(Interface.objects.filter(module=self.module).count(), interface_count_before)

    def test_preset_unused_pk_save_creates_module(self):
        """A never-saved instance with an explicit unused pk is created, not treated as a move."""
        unused_pk = Module.objects.order_by('-pk').first().pk + 1000
        device = create_test_device('Device P')
        bay = ModuleBay.objects.create(device=device, name='Preset Bay')
        Module(pk=unused_pk, device=device, module_bay=bay, module_type=self.module_type).save()
        self.assertTrue(Module.objects.filter(pk=unused_pk).exists())
        self.assertTrue(Interface.objects.filter(module_id=unused_pk, name='eth0').exists())

    def test_pk_reassignment_clone_creates_new_module(self):
        """A fetched instance saved under a fresh unused pk creates a new row and keeps the original."""
        clone = Module.objects.get(pk=self.module.pk)
        device = create_test_device('Device C')
        bay = ModuleBay.objects.create(device=device, name='Clone Bay')
        clone.pk = Module.objects.order_by('-pk').first().pk + 1000
        clone.device = device
        clone.module_bay = bay
        clone.save()
        self.assertTrue(Module.objects.filter(pk=clone.pk).exists())
        self.assertTrue(Module.objects.filter(pk=self.module.pk).exists())

    def test_plain_save_after_concurrent_delete_recreates_module(self):
        """A stale full save whose row was deleted underneath falls through to create, like a plain Django save."""
        stale = Module.objects.get(pk=self.module.pk)
        Module.objects.filter(pk=stale.pk).delete()
        stale.comments = 'Recreated'
        stale.save()
        self.assertTrue(Module.objects.filter(pk=stale.pk, comments='Recreated').exists())

    def test_cross_device_stale_plain_save_moves_subtree_back_consistently(self):
        """
        A stale full save that reverts a committed cross-device move runs the planner,
        so the root module, child bays, and components land on one device together.
        """
        stale = Module.objects.get(pk=self.module.pk)

        mover = Module.objects.get(pk=self.module.pk)
        mover.device = self.device_b
        mover.module_bay = self.bay_c
        mover.save()

        stale.comments = 'Stale full save'
        stale.save()

        self.module.refresh_from_db()
        self.child_bay.refresh_from_db()
        interface = Interface.objects.get(module=self.module, name='eth0')
        self.assertEqual(self.module.comments, 'Stale full save')
        self.assertEqual(self.module.device, self.device)
        self.assertEqual(self.module.module_bay, self.bay_a)
        self.assertEqual(self.child_bay.device, self.device)
        self.assertEqual(self.child_bay.parent_id, self.bay_a.pk)
        self.assertEqual(interface.device, self.device)


class ModuleMoveRaceTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        cls.device = create_test_device('Device A')
        cls.bay_a = ModuleBay.objects.create(device=cls.device, name='Bay A')
        cls.bay_b = ModuleBay.objects.create(device=cls.device, name='Bay B')
        cls.bay_c = ModuleBay.objects.create(device=cls.device, name='Bay C')
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type M')
        cls.module_1 = Module.objects.create(
            device=cls.device, module_bay=cls.bay_a, module_type=cls.module_type
        )
        cls.module_2 = Module.objects.create(
            device=cls.device, module_bay=cls.bay_b, module_type=cls.module_type
        )

    def test_bulk_update_into_occupied_bay_hits_db_constraint(self):
        with self.assertRaises(IntegrityError):
            with transaction.atomic():
                Module.objects.filter(pk=self.module_2.pk).update(module_bay=self.bay_a)

    def test_locked_validation_catches_occupied_bay_bypassing_clean(self):
        # Simulate a TOCTOU window: clean() was never run (direct save), the target
        # bay is genuinely occupied, and the in-save locked validation must reject.
        self.module_2.module_bay = self.bay_a
        with self.assertRaises(AbortRequest):
            self.module_2.save()

    def test_move_into_own_subtree_bypassing_clean_is_rejected(self):
        child_bay = ModuleBay.objects.create(device=self.device, module=self.module_1, name='Child Bay 1')
        self.module_1.module_bay = child_bay
        with self.assertRaises(AbortRequest):
            self.module_1.save()

    def test_full_clean_on_cyclic_hierarchy_raises_validation_error_promptly(self):
        # A two-module cycle created via .update() (bypassing clean()) must raise promptly, not hang.
        child_bay_1 = ModuleBay.objects.create(device=self.device, module=self.module_1, name='Child Bay 1')
        child_bay_2 = ModuleBay.objects.create(device=self.device, module=self.module_2, name='Child Bay 2')
        Module.objects.filter(pk=self.module_1.pk).update(module_bay=child_bay_2)
        Module.objects.filter(pk=self.module_2.pk).update(module_bay=child_bay_1)
        module = Module.objects.get(pk=self.module_1.pk)
        module.module_bay = self.bay_c
        with fail_after(15):
            with self.assertRaises(ValidationError) as cm:
                module.full_clean()
        self.assertIn('module_bay', cm.exception.message_dict)
        self.assertIn('contains a cycle', str(cm.exception.message_dict['module_bay']))

    def test_move_into_cyclic_hierarchy_bypassing_clean_raises_abort_request(self):
        # Same cycle, reached via direct save() (clean() never runs); must abort, not hang.
        child_bay_1 = ModuleBay.objects.create(device=self.device, module=self.module_1, name='Child Bay 1')
        child_bay_2 = ModuleBay.objects.create(device=self.device, module=self.module_2, name='Child Bay 2')
        Module.objects.filter(pk=self.module_1.pk).update(module_bay=child_bay_2)
        Module.objects.filter(pk=self.module_2.pk).update(module_bay=child_bay_1)
        self.module_1.module_bay = self.bay_c
        with fail_after(15):
            with self.assertRaises(AbortRequest):
                self.module_1.save()

    def test_create_in_cyclic_hierarchy_with_token_template_raises_abort_request(self):
        # _save_new()'s template resolution walks the target bay's ancestry; a cycle must abort, not hang.
        child_bay_1 = ModuleBay.objects.create(device=self.device, module=self.module_1, name='Child Bay 1')
        child_bay_2 = ModuleBay.objects.create(device=self.device, module=self.module_2, name='Child Bay 2')
        Module.objects.filter(pk=self.module_1.pk).update(module_bay=child_bay_2)
        Module.objects.filter(pk=self.module_2.pk).update(module_bay=child_bay_1)
        token_type = ModuleType.objects.create(
            manufacturer=self.module_type.manufacturer, model='Module Type Token'
        )
        InterfaceTemplate.objects.create(
            module_type=token_type, name='eth{module}/0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        target_bay = ModuleBay.objects.create(
            device=self.device, module=self.module_1, name='Child Bay 3', position='3'
        )
        # Fresh fetch so the ancestry walk reads database state, not cached relations
        target_bay = ModuleBay.objects.get(pk=target_bay.pk)
        module = Module(device=self.device, module_bay=target_bay, module_type=token_type)
        with fail_after(15):
            with self.assertRaises(AbortRequest) as cm:
                module.save()
        self.assertIn('contains a cycle', cm.exception.message)

    def test_lock_rediscovers_membership_added_after_planning(self):
        old = Module.objects.get(pk=self.module_1.pk)
        new = Module.objects.get(pk=self.module_1.pk)
        new.module_bay = ModuleBay.objects.create(device=self.device, name='Bay C')
        plan = ModuleMovePlan.from_module(old_module=old, new_module=new)
        late_interface = Interface.objects.create(
            device=self.device, module=self.module_1, name='late0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        with transaction.atomic():
            plan.lock()
        self.assertIn(late_interface.pk, [obj.pk for obj in plan.components[Interface]])

    def test_locked_device_get_does_not_exist_raises_abort_request(self):
        # Same TOCTOU, at the device row lock acquired by ModuleMovePlan.lock(): the
        # single-statement filter(pk__in=...) form returns fewer rows than expected pks.
        self.module_1.module_bay = self.bay_c
        with patch.object(Device.objects, 'select_for_update') as mock_sfu:
            mock_sfu.return_value.filter.return_value.order_by.return_value = []
            with self.assertRaises(AbortRequest):
                self.module_1.save()


class ModuleMoveSaveContractTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        cls.device = create_test_device('Device A')
        cls.bay_a = ModuleBay.objects.create(device=cls.device, name='Bay A')
        cls.bay_b = ModuleBay.objects.create(device=cls.device, name='Bay B')
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type M')
        cls.module = Module.objects.create(
            device=cls.device, module_bay=cls.bay_a, module_type=cls.module_type
        )

    def test_update_fields_excluding_module_bay_saves_without_moving(self):
        # A changed placement field left out of update_fields is never persisted, so no move happens.
        self.module.module_bay = self.bay_b
        self.module.serial = 'ABC123'
        self.module.save(update_fields=['serial'])
        self.module.refresh_from_db()
        self.assertEqual(self.module.serial, 'ABC123')
        self.assertEqual(self.module.module_bay, self.bay_a)

    def test_update_fields_including_placement_moves(self):
        self.module.module_bay = self.bay_b
        self.module.save(update_fields=['device', 'module_bay'])
        self.module.refresh_from_db()
        self.assertEqual(self.module.module_bay, self.bay_b)

    def test_generator_update_fields_excluding_module_bay_saves_without_moving(self):
        self.module.module_bay = self.bay_b
        self.module.serial = 'ABC123'
        self.module.save(update_fields=(field for field in ('serial',)))
        self.module.refresh_from_db()
        self.assertEqual(self.module.serial, 'ABC123')
        self.assertEqual(self.module.module_bay, self.bay_a)

    def test_generator_update_fields_including_placement_moves(self):
        self.module.module_bay = self.bay_b
        self.module.save(update_fields=(field for field in ('device', 'module_bay')))
        self.module.refresh_from_db()
        self.assertEqual(self.module.module_bay, self.bay_b)

    def test_deadlock_is_translated_to_abort_request(self):
        deadlock = OperationalError('Simulated deadlock error')
        deadlock.__cause__ = type('FakeDeadlock', (Exception,), {'sqlstate': '40P01'})()
        self.module.module_bay = self.bay_b
        with patch.object(ModuleMovePlan, 'lock', side_effect=deadlock):
            with self.assertRaises(AbortRequest):
                self.module.save()

    def test_non_deadlock_operational_error_propagates(self):
        error = OperationalError('Simulated connection error')
        self.module.module_bay = self.bay_b
        with patch.object(ModuleMovePlan, 'lock', side_effect=error):
            with self.assertRaises(OperationalError):
                self.module.save()

    def test_post_save_signals_use_write_alias(self):
        ModuleBay.objects.create(device=self.device, module=self.module, name='Child Bay 1')
        self.module.module_bay = self.bay_b
        with patch('dcim.models.module_moves.post_save') as mock_signal:
            self.module.save()
        self.assertTrue(mock_signal.send.called)
        for call in mock_signal.send.call_args_list:
            self.assertEqual(call.kwargs['using'], router.db_for_write(ModuleBay))

    def test_move_to_deleted_bay_raises_abort_request(self):
        target = ModuleBay.objects.create(device=self.device, name='Bay C')
        self.module.module_bay = target
        ModuleBay.objects.filter(pk=target.pk).delete()
        with self.assertRaises(AbortRequest):
            self.module.save()

    def test_partial_update_fields_after_completed_move_succeeds(self):
        self.module.module_bay = self.bay_b
        self.module.save()
        self.module.serial = 'A1'
        self.module.save(update_fields=['serial'])
        self.module.refresh_from_db()
        self.assertEqual(self.module.serial, 'A1')
        self.assertEqual(self.module.module_bay, self.bay_b)

    def test_update_fields_with_module_bay_name_moves_same_device(self):
        child_bay = ModuleBay.objects.create(device=self.device, module=self.module, name='Child Bay 1')
        self.module.module_bay = self.bay_b
        self.module.save(update_fields=['module_bay'])
        self.module.refresh_from_db()
        child_bay.refresh_from_db()
        self.assertEqual(self.module.module_bay, self.bay_b)
        self.assertEqual(child_bay.parent_id, self.bay_b.pk)

    def test_update_fields_with_attnames_moves_cross_device(self):
        device_b = create_test_device('Device B')
        bay_c = ModuleBay.objects.create(device=device_b, name='Bay C')
        self.module.device = device_b
        self.module.module_bay = bay_c
        self.module.save(update_fields=['device_id', 'module_bay_id'])
        self.module.refresh_from_db()
        self.assertEqual(self.module.device, device_b)
        self.assertEqual(self.module.module_bay, bay_c)

    def test_update_fields_missing_device_rejects_cross_device_delta(self):
        device_b = create_test_device('Device B')
        bay_c = ModuleBay.objects.create(device=device_b, name='Bay C')
        self.module.device = device_b
        self.module.module_bay = bay_c
        with self.assertRaises(AbortRequest):
            self.module.save(update_fields=['module_bay'])

    def test_update_fields_incomplete_against_current_placement_rejects(self):
        """
        update_fields completeness is judged against the locked database placement,
        not the placement the saving instance last observed.
        """
        device_b = create_test_device('Device B')
        bay_c = ModuleBay.objects.create(device=device_b, name='Bay C')
        stale = Module.objects.get(pk=self.module.pk)

        mover = Module.objects.get(pk=self.module.pk)
        mover.device = device_b
        mover.module_bay = bay_c
        mover.save()

        stale.module_bay = self.bay_b
        with self.assertRaises(AbortRequest):
            stale.save(update_fields=['module_bay'])
        self.module.refresh_from_db()
        self.assertEqual(self.module.device, device_b)
        self.assertEqual(self.module.module_bay, bay_c)

    def test_update_fields_with_unchanged_placement_field_saves_without_moving(self):
        """
        A placement field listed in update_fields with an unchanged value is written as-is;
        a diverged placement field left out of update_fields never triggers a move.
        """
        self.module.module_bay = self.bay_b
        self.module.serial = 'NOMOVE1'
        with patch.object(ModuleMovePlan, 'from_module') as mock_plan:
            self.module.save(update_fields=['device', 'serial'])
        mock_plan.assert_not_called()
        self.module.refresh_from_db()
        self.assertEqual(self.module.serial, 'NOMOVE1')
        self.assertEqual(self.module.module_bay, self.bay_a)


class ModuleMoveRenameTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        role = DeviceRole.objects.create(name='Role 1', slug='role-1')
        site = Site.objects.create(name='Site 1', slug='site-1')
        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Chassis', slug='chassis')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 1', position='1')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 2', position='2')

        cls.line_card_type = ModuleType.objects.create(manufacturer=manufacturer, model='Line Card')
        InterfaceTemplate.objects.create(
            module_type=cls.line_card_type,
            name='Ethernet{module}/1',
            label='Port {module}',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        ModuleBayTemplate.objects.create(
            module_type=cls.line_card_type, name='SFP bay {module}/1', position='{module}/1'
        )

        cls.sfp_type = ModuleType.objects.create(manufacturer=manufacturer, model='SFP')
        InterfaceTemplate.objects.create(
            module_type=cls.sfp_type, name='SFP {module}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
        )

        cls.device = Device.objects.create(name='Chassis A', device_type=device_type, role=role, site=site)
        cls.slot_1 = cls.device.modulebays.get(name='Slot 1')
        cls.slot_2 = cls.device.modulebays.get(name='Slot 2')
        cls.line_card = Module.objects.create(
            device=cls.device, module_bay=cls.slot_1, module_type=cls.line_card_type
        )
        cls.sfp_bay = cls.line_card.modulebays.get(name='SFP bay 1/1')
        cls.sfp_module = Module.objects.create(
            device=cls.device, module_bay=cls.sfp_bay, module_type=cls.sfp_type
        )

    def _move_line_card_to_slot_2(self):
        self.line_card.module_bay = self.slot_2
        self.line_card.full_clean()
        self.line_card.save()

    def test_same_device_move_renames_templated_interface(self):
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        self._move_line_card_to_slot_2()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'Ethernet2/1')
        self.assertEqual(interface.label, 'Port 2')
        self.assertEqual(interface._name, naturalize_interface('Ethernet2/1', max_length=100))

    def test_move_renames_nested_bay_and_grandchild_components(self):
        sfp_interface = self.sfp_module.interfaces.get(name='SFP 1/1')
        self._move_line_card_to_slot_2()
        self.sfp_bay.refresh_from_db()
        self.assertEqual(self.sfp_bay.name, 'SFP bay 2/1')
        self.assertEqual(self.sfp_bay.position, '2/1')
        sfp_interface.refresh_from_db()
        self.assertEqual(sfp_interface.name, 'SFP 2/1')

    def test_manually_renamed_component_is_preserved(self):
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        interface.name = 'uplink-core'
        interface.save()
        self._move_line_card_to_slot_2()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'uplink-core')

    def test_manually_changed_label_is_preserved_while_name_renames(self):
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        interface.label = 'Uplink'
        interface.save()
        self._move_line_card_to_slot_2()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'Ethernet2/1')
        self.assertEqual(interface.label, 'Uplink')

    def test_ambiguous_template_match_is_preserved(self):
        # A second template resolving to the same old name makes the match ambiguous
        InterfaceTemplate.objects.create(
            module_type=self.line_card_type, name='Ethernet1/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        self._move_line_card_to_slot_2()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'Ethernet1/1')

    def test_move_after_module_type_change_preserves_names(self):
        other_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Other Card'
        )
        self.line_card.module_type = other_type
        self.line_card.save()
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        self._move_line_card_to_slot_2()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'Ethernet1/1')

    def test_manually_changed_bay_position_stops_rename_cascade(self):
        self.sfp_bay.position = 'X'
        self.sfp_bay.save()
        sfp_interface = self.sfp_module.interfaces.get(name='SFP 1/1')
        self._move_line_card_to_slot_2()
        self.sfp_bay.refresh_from_db()
        self.assertEqual(self.sfp_bay.position, 'X')
        sfp_interface.refresh_from_db()
        self.assertEqual(sfp_interface.name, 'SFP 1/1')

    def test_duplicate_final_names_within_moved_set_fail(self):
        Interface.objects.create(
            device=self.device, module=self.line_card, name='Ethernet2/1',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        self.line_card.module_bay = self.slot_2
        with self.assertRaises(ValidationError):
            self.line_card.full_clean()

    def test_conflict_with_existing_destination_component_fails(self):
        Interface.objects.create(
            device=self.device, name='Ethernet2/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        self.line_card.module_bay = self.slot_2
        with self.assertRaises(ValidationError):
            self.line_card.full_clean()

    def test_rename_chain_collision_is_rejected(self):
        # 'E{module}/1' resolves to E1/1 in slot 1 and E2/1 in slot 2, while 'E2/{module}'
        # resolves to E2/1 in slot 1 and E2/2 in slot 2: E1/1 -> E2/1 while E2/1 -> E2/2.
        # Applying both renames on the same device in one statement is order-dependent.
        InterfaceTemplate.objects.create(
            module_type=self.line_card_type, name='E2/{module}', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        InterfaceTemplate.objects.create(
            module_type=self.line_card_type, name='E{module}/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        Interface.objects.create(
            device=self.device, module=self.line_card, name='E2/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        Interface.objects.create(
            device=self.device, module=self.line_card, name='E1/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        self.line_card.module_bay = self.slot_2
        with self.assertRaises(ValidationError) as cm:
            self.line_card.full_clean()
        self.assertIn('current name of another moved', str(cm.exception))

    def test_bay_rename_chain_collision_is_rejected(self):
        # 'Bay E2/{module}' resolves to Bay E2/1 in slot 1 and Bay E2/2 in slot 2, while
        # 'Bay E{module}/1' resolves to Bay E1/1 in slot 1 and Bay E2/1 in slot 2: Bay E1/1
        # -> Bay E2/1 while Bay E2/1 -> Bay E2/2. Applying both renames in one same-device
        # bulk statement would collide on the destination namespace depending on row order.
        ModuleBayTemplate.objects.create(module_type=self.line_card_type, name='Bay E2/{module}')
        ModuleBayTemplate.objects.create(module_type=self.line_card_type, name='Bay E{module}/1')
        ModuleBay.objects.create(device=self.device, module=self.line_card, name='Bay E2/1')
        ModuleBay.objects.create(device=self.device, module=self.line_card, name='Bay E1/1')
        self.line_card.module_bay = self.slot_2
        with self.assertRaises(ValidationError) as cm:
            self.line_card.full_clean()
        self.assertIn('current name of another moved module bay', str(cm.exception))

    def test_rename_exceeding_name_max_length_is_rejected(self):
        name_limit = Interface._meta.get_field('name').max_length
        template_name_limit = InterfaceTemplate._meta.get_field('name').max_length
        position_limit = ModuleBay._meta.get_field('position').max_length
        prefix = 'A' * (template_name_limit - len('{module}'))
        long_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Oversized Name Card'
        )
        InterfaceTemplate.objects.create(
            module_type=long_type, name=f'{prefix}{{module}}', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        long_bay = ModuleBay.objects.create(device=self.device, name='Long Bay', position='X' * position_limit)
        module = Module.objects.create(device=self.device, module_bay=self.slot_2, module_type=long_type)
        interface = module.interfaces.get()
        self.assertLessEqual(len(interface.name), name_limit)

        module.module_bay = long_bay
        with self.assertRaises(ValidationError) as cm:
            module.full_clean()
        self.assertIn(str(interface), str(cm.exception))

    def test_rename_exceeding_position_max_length_is_rejected(self):
        position_limit = ModuleBay._meta.get_field('position').max_length
        template_position_limit = ModuleBayTemplate._meta.get_field('position').max_length
        prefix = 'B' * (template_position_limit - len('{module}'))
        long_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Oversized Position Card'
        )
        ModuleBayTemplate.objects.create(
            module_type=long_type, name='Nested Bay {module}', position=f'{prefix}{{module}}'
        )
        long_bay = ModuleBay.objects.create(device=self.device, name='Long Bay', position='Y' * position_limit)
        module = Module.objects.create(device=self.device, module_bay=self.slot_2, module_type=long_type)
        bay = module.modulebays.get()

        module.module_bay = long_bay
        with self.assertRaises(ValidationError) as cm:
            module.full_clean()
        self.assertIn(str(bay), str(cm.exception))

    def test_leaf_token_left_raw_matches_fresh_walker_resolution(self):
        # A bay whose stored position literally contains an unresolved {module} token
        # (bypassing normal template-driven creation) must resolve its own children's
        # names identically to a fresh get_module_bay_positions() call, both before and
        # after a move, so a later move can still recognize the template match.
        odd_bay = ModuleBay.objects.create(
            device=self.device, module=self.line_card, name='Odd Bay', position='{module}A'
        )
        child_module = Module.objects.create(device=self.device, module_bay=odd_bay, module_type=self.sfp_type)
        child_interface = child_module.interfaces.get()

        self._move_line_card_to_slot_2()
        child_interface.refresh_from_db()
        odd_bay.refresh_from_db()
        expected_name = resolve_module_placeholder('SFP {module}', get_module_bay_positions(odd_bay))
        self.assertEqual(child_interface.name, expected_name)

    def test_second_move_still_renames_child_of_raw_token_bay(self):
        # Regression pin: a second move of the same module must still template-match and
        # rename a child nested under a raw-token bay, not silently stop renaming.
        two_token_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Two Token SFP'
        )
        InterfaceTemplate.objects.create(
            module_type=two_token_type, name='SFP {module}/{module}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
        )
        odd_bay = ModuleBay.objects.create(
            device=self.device, module=self.line_card, name='Odd Bay', position='{module}A'
        )
        child_module = Module.objects.create(device=self.device, module_bay=odd_bay, module_type=two_token_type)
        child_interface = child_module.interfaces.get()
        self.assertEqual(child_interface.name, 'SFP 1/{module}A')

        self._move_line_card_to_slot_2()
        child_interface.refresh_from_db()
        self.assertEqual(child_interface.name, 'SFP 2/{module}A')

        self.line_card.module_bay = self.slot_1
        self.line_card.full_clean()
        self.line_card.save()
        child_interface.refresh_from_db()
        self.assertEqual(child_interface.name, 'SFP 1/{module}A')

    def test_move_into_raw_token_bay_resolves_ancestor_from_child(self):
        """
        A raw {module} token on the DESTINATION bay resolves from the planned child
        position below it, matching a fresh post-move walker resolution exactly.
        """
        raw_slot = ModuleBay.objects.create(device=self.device, name='Raw Slot', position='{module}R')
        b_bay = ModuleBay.objects.create(device=self.device, module=self.line_card, name='B Bay', position='B')
        two_token_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Two Token'
        )
        InterfaceTemplate.objects.create(
            module_type=two_token_type, name='SFP {module}/{module}',
            type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS,
        )
        child_module = Module.objects.create(
            device=self.device, module_bay=b_bay, module_type=two_token_type
        )
        child_interface = child_module.interfaces.get(name='SFP 1/B')

        self.line_card.module_bay = raw_slot
        self.line_card.full_clean()
        self.line_card.save()

        child_interface.refresh_from_db()
        self.assertEqual(child_interface.name, 'SFP BR/B')
        self.assertNotIn('{module}', child_interface.name)
        fresh_chain = get_module_bay_positions(ModuleBay.objects.get(pk=b_bay.pk))
        self.assertEqual(child_interface.name, resolve_module_placeholder('SFP {module}/{module}', fresh_chain))
        # Leaf-raw parity: the moved module's own component sees the destination
        # bay's token unresolved, exactly as an install into that bay would
        root_interface = self.line_card.interfaces.get(name__startswith='Ethernet')
        self.assertEqual(root_interface.name, 'Ethernet{module}R/1')

    def test_move_under_multiple_raw_token_ancestors_resolves_full_chain(self):
        """
        Every ancestor level carrying a raw {module} token inherits from the resolved
        position below it, across more than one level.
        """
        bare_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Bare Carrier'
        )
        three_token_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Three Token'
        )
        InterfaceTemplate.objects.create(
            module_type=three_token_type, name='T{module}.{module}.{module}',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        # Source: depth-3 chain ['7', '8', 'B'] so the three-token template resolves
        s1 = ModuleBay.objects.create(device=self.device, name='S1', position='7')
        mid_src = Module.objects.create(device=self.device, module_bay=s1, module_type=bare_type)
        s2 = ModuleBay.objects.create(device=self.device, module=mid_src, name='S2', position='8')
        carrier = Module.objects.create(device=self.device, module_bay=s2, module_type=bare_type)
        c_bay = ModuleBay.objects.create(device=self.device, module=carrier, name='C Bay', position='B')
        child_module = Module.objects.create(
            device=self.device, module_bay=c_bay, module_type=three_token_type
        )
        child_interface = child_module.interfaces.get(name='T7.8.B')
        # Destination: two stacked raw-token ancestors
        g_bay = ModuleBay.objects.create(device=self.device, name='G Bay', position='{module}X')
        mid_dst = Module.objects.create(device=self.device, module_bay=g_bay, module_type=bare_type)
        r_bay = ModuleBay.objects.create(device=self.device, module=mid_dst, name='R Bay', position='{module}R')

        carrier.module_bay = r_bay
        carrier.full_clean()
        carrier.save()

        child_interface.refresh_from_db()
        self.assertEqual(child_interface.name, 'TBRX.BR.B')
        self.assertNotIn('{module}', child_interface.name)
        fresh_chain = get_module_bay_positions(ModuleBay.objects.get(pk=c_bay.pk))
        self.assertEqual(fresh_chain, ['BRX', 'BR', 'B'])
        self.assertEqual(
            child_interface.name, resolve_module_placeholder('T{module}.{module}.{module}', fresh_chain)
        )

    def test_move_to_shallower_bay_with_unresolvable_template_is_rejected(self):
        """
        A component whose name matched a source template that cannot resolve at the
        destination depth rejects the move instead of silently keeping the stale name.
        """
        two_token_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Two Token'
        )
        InterfaceTemplate.objects.create(
            module_type=two_token_type, name='SFP {module}/{module}',
            type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS,
        )
        b_bay = ModuleBay.objects.create(device=self.device, module=self.line_card, name='B Bay', position='B')
        child_module = Module.objects.create(
            device=self.device, module_bay=b_bay, module_type=two_token_type
        )
        shallow = ModuleBay.objects.create(device=self.device, name='Shallow', position='X')

        child_module.module_bay = shallow
        with self.assertRaises(ValidationError) as cm:
            child_module.full_clean()
        self.assertIn('cannot be resolved', str(cm.exception))
        with self.assertRaises(AbortRequest):
            child_module.save()
        child_module.refresh_from_db()
        self.assertEqual(child_module.module_bay_id, b_bay.pk)
        self.assertTrue(child_module.interfaces.filter(name='SFP 1/B').exists())

    def test_matched_label_unresolvable_at_destination_rejects_move(self):
        """
        A label matching its source template resolution rejects the move when the
        label template cannot resolve at the destination depth.
        """
        label_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Label Type'
        )
        InterfaceTemplate.objects.create(
            module_type=label_type, name='N{module}', label='L{module}/{module}',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        b_bay = ModuleBay.objects.create(device=self.device, module=self.line_card, name='B Bay', position='B')
        child_module = Module.objects.create(device=self.device, module_bay=b_bay, module_type=label_type)
        interface = child_module.interfaces.get(name='NB')
        self.assertEqual(interface.label, 'L1/B')
        shallow = ModuleBay.objects.create(device=self.device, name='Shallow', position='X')

        child_module.module_bay = shallow
        with self.assertRaises(ValidationError):
            child_module.full_clean()

    def test_unmatched_label_with_unresolvable_template_still_moves(self):
        """
        A manually customized label never source-matches, so an unresolvable label
        template does not block the move and the label is preserved.
        """
        label_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Label Type'
        )
        InterfaceTemplate.objects.create(
            module_type=label_type, name='N{module}', label='L{module}/{module}',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        b_bay = ModuleBay.objects.create(device=self.device, module=self.line_card, name='B Bay', position='B')
        child_module = Module.objects.create(device=self.device, module_bay=b_bay, module_type=label_type)
        interface = child_module.interfaces.get(name='NB')
        interface.label = 'Custom'
        interface.save()
        shallow = ModuleBay.objects.create(device=self.device, name='Shallow', position='X')

        child_module.module_bay = shallow
        child_module.full_clean()
        child_module.save()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'NX')
        self.assertEqual(interface.label, 'Custom')

    def test_matched_bay_position_unresolvable_at_destination_rejects_move(self):
        """
        A module bay position matching its source template resolution rejects the move
        when the position template cannot resolve at the destination depth.
        """
        pos_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Pos Type'
        )
        ModuleBayTemplate.objects.create(module_type=pos_type, name='PB', position='{module}/{module}')
        b_bay = ModuleBay.objects.create(device=self.device, module=self.line_card, name='B Bay', position='B')
        pos_module = Module.objects.create(device=self.device, module_bay=b_bay, module_type=pos_type)
        self.assertEqual(pos_module.modulebays.get(name='PB').position, '1/B')
        shallow = ModuleBay.objects.create(device=self.device, name='Shallow', position='X')

        pos_module.module_bay = shallow
        with self.assertRaises(ValidationError):
            pos_module.full_clean()


class ModuleCrossDeviceBlockerTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        cls.device_a = create_test_device('Device A')
        cls.device_b = create_test_device('Device B')
        cls.bay_a = ModuleBay.objects.create(device=cls.device_a, name='Bay A')
        cls.bay_b = ModuleBay.objects.create(device=cls.device_b, name='Bay B')
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Module Type M')
        cls.module = Module.objects.create(
            device=cls.device_a, module_bay=cls.bay_a, module_type=cls.module_type
        )
        cls.interface = Interface.objects.create(
            device=cls.device_a, module=cls.module, name='eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )

    def _assert_move_blocked(self, token):
        self.module.device = self.device_b
        self.module.module_bay = self.bay_b
        with self.assertRaises(ValidationError) as cm:
            self.module.full_clean()
        self.assertIn('cannot be moved to a different device', str(cm.exception))
        self.assertIn(token, str(cm.exception))

    def _assert_move_allowed(self):
        self.module.device = self.device_b
        self.module.module_bay = self.bay_b
        self.module.full_clean()

    def test_cable_blocks(self):
        peer = Interface.objects.create(
            device=self.device_a, name='peer0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        Cable(a_terminations=[self.interface], b_terminations=[peer]).save()
        self._assert_move_blocked('cabled')

    def test_mark_connected_blocks(self):
        self.interface.mark_connected = True
        self.interface.save()
        self._assert_move_blocked('cabled or connection-marked')

    def test_ip_address_blocks(self):
        IPAddress.objects.create(address='192.0.2.1/32', assigned_object=self.interface)
        self._assert_move_blocked('IP addresses')

    def test_fhrp_group_assignment_blocks(self):
        group = FHRPGroup.objects.create(protocol=FHRPGroupProtocolChoices.PROTOCOL_VRRP2, group_id=1)
        FHRPGroupAssignment.objects.create(group=group, interface=self.interface, priority=100)
        self._assert_move_blocked('FHRP')

    def test_tunnel_termination_blocks(self):
        tunnel = Tunnel.objects.create(name='Tunnel 1', encapsulation=TunnelEncapsulationChoices.ENCAP_IP_IP)
        TunnelTermination.objects.create(tunnel=tunnel, termination=self.interface)
        self._assert_move_blocked('tunnel')

    def test_l2vpn_termination_blocks(self):
        l2vpn = L2VPN.objects.create(name='L2VPN 1', slug='l2vpn-1', type=L2VPNTypeChoices.TYPE_VXLAN)
        L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=self.interface)
        self._assert_move_blocked('L2VPN')

    def test_virtual_circuit_termination_blocks(self):
        provider = Provider.objects.create(name='Provider 1', slug='provider-1')
        provider_network = ProviderNetwork.objects.create(provider=provider, name='Provider Network 1')
        vc_type = VirtualCircuitType.objects.create(name='VC Type 1', slug='vc-type-1')
        vc = VirtualCircuit.objects.create(provider_network=provider_network, cid='VC 1', type=vc_type)
        virtual_interface = Interface.objects.create(
            device=self.device_a, module=self.module, name='vc0', type=InterfaceTypeChoices.TYPE_VIRTUAL
        )
        VirtualCircuitTermination.objects.create(virtual_circuit=vc, interface=virtual_interface)
        self._assert_move_blocked('virtual circuit')

    def test_wireless_link_blocks(self):
        radio_a = Interface.objects.create(
            device=self.device_a, module=self.module, name='radio0', type=InterfaceTypeChoices.TYPE_80211AC
        )
        radio_b = Interface.objects.create(
            device=self.device_a, name='radio1', type=InterfaceTypeChoices.TYPE_80211AC
        )
        WirelessLink(interface_a=radio_a, interface_b=radio_b, ssid='LINK1').save()
        self._assert_move_blocked('wireless links')

    def test_wireless_lan_blocks(self):
        wlan = WirelessLAN.objects.create(ssid='SSID1')
        self.interface.wireless_lans.add(wlan)
        self._assert_move_blocked('wireless LAN')

    def test_untagged_vlan_blocks(self):
        vlan = VLAN.objects.create(vid=100, name='VLAN 100')
        self.interface.mode = InterfaceModeChoices.MODE_ACCESS
        self.interface.untagged_vlan = vlan
        self.interface.save()
        self._assert_move_blocked('untagged VLAN')

    def test_tagged_vlan_blocks(self):
        vlan = VLAN.objects.create(vid=200, name='VLAN 200')
        self.interface.mode = InterfaceModeChoices.MODE_TAGGED
        self.interface.save()
        self.interface.tagged_vlans.add(vlan)
        self._assert_move_blocked('tagged VLANs')

    def test_qinq_svlan_blocks(self):
        svlan = VLAN.objects.create(vid=999, name='SVLAN 999')
        self.interface.mode = InterfaceModeChoices.MODE_Q_IN_Q
        self.interface.qinq_svlan = svlan
        self.interface.save()
        self._assert_move_blocked('Q-in-Q')

    def test_vlan_translation_policy_blocks(self):
        policy = VLANTranslationPolicy.objects.create(name='Policy 1')
        self.interface.vlan_translation_policy = policy
        self.interface.save()
        self._assert_move_blocked('VLAN translation')

    def test_vdc_assignment_blocks(self):
        vdc = VirtualDeviceContext.objects.create(device=self.device_a, name='VDC 1', status='active')
        self.interface.vdcs.add(vdc)
        self._assert_move_blocked('VDC')

    def test_vrf_blocks(self):
        vrf = VRF.objects.create(name='VRF 1')
        self.interface.vrf = vrf
        self.interface.save()
        self._assert_move_blocked('VRF')

    def test_parent_outside_moved_set_blocks(self):
        parent = Interface.objects.create(
            device=self.device_a, name='parent0', type=InterfaceTypeChoices.TYPE_VIRTUAL
        )
        Interface.objects.create(
            device=self.device_a, module=self.module, name='child0',
            type=InterfaceTypeChoices.TYPE_VIRTUAL, parent=parent,
        )
        self._assert_move_blocked('boundary')

    def test_bridge_outside_moved_set_blocks(self):
        bridge = Interface.objects.create(
            device=self.device_a, name='bridge0', type=InterfaceTypeChoices.TYPE_BRIDGE
        )
        self.interface.bridge = bridge
        self.interface.save()
        self._assert_move_blocked('boundary')

    def test_lag_outside_moved_set_blocks(self):
        lag = Interface.objects.create(
            device=self.device_a, name='lag0', type=InterfaceTypeChoices.TYPE_LAG
        )
        self.interface.lag = lag
        self.interface.save()
        self._assert_move_blocked('boundary')

    def test_nonmoved_member_of_moved_lag_blocks(self):
        lag = Interface.objects.create(
            device=self.device_a, module=self.module, name='lag0', type=InterfaceTypeChoices.TYPE_LAG
        )
        Interface.objects.create(
            device=self.device_a, name='member0', type=InterfaceTypeChoices.TYPE_1GE_FIXED, lag=lag
        )
        self._assert_move_blocked('boundary')

    def test_split_port_mapping_blocks(self):
        front_port = FrontPort.objects.create(
            device=self.device_a, module=self.module, name='Front 1', type=PortTypeChoices.TYPE_LC
        )
        rear_port = RearPort.objects.create(
            device=self.device_a, name='Rear 1', type=PortTypeChoices.TYPE_LC, positions=1
        )
        PortMapping.objects.create(
            front_port=front_port, front_port_position=1, rear_port=rear_port, rear_port_position=1
        )
        self._assert_move_blocked('port mappings')

    def test_split_power_outlet_blocks(self):
        power_port = PowerPort.objects.create(device=self.device_a, name='PP 1')
        PowerOutlet.objects.create(
            device=self.device_a, module=self.module, name='Outlet 1', power_port=power_port
        )
        self._assert_move_blocked('power outlet')

    def test_attached_inventory_item_blocks(self):
        InventoryItem.objects.create(device=self.device_a, name='Item 1', component=self.interface)
        self._assert_move_blocked('inventory items')

    def test_intra_module_bridge_pair_is_allowed(self):
        other = Interface.objects.create(
            device=self.device_a, module=self.module, name='eth1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        self.interface.bridge = other
        self.interface.save()
        self._assert_move_allowed()

    def test_intra_module_power_pair_is_allowed(self):
        power_port = PowerPort.objects.create(device=self.device_a, module=self.module, name='PP 1')
        PowerOutlet.objects.create(
            device=self.device_a, module=self.module, name='Outlet 1', power_port=power_port
        )
        self._assert_move_allowed()

    #
    # Blockers name the offending components, not just how many there are
    #

    def _blocked_message(self):
        self.module.device = self.device_b
        self.module.module_bay = self.bay_b
        with self.assertRaises(ValidationError) as cm:
            self.module.full_clean()
        return str(cm.exception)

    @staticmethod
    def _samples(message):
        """Every parenthesized "e.g." list in a blocker message, as a list of name lists."""
        return [
            [name.strip() for name in group.split(',')]
            for group in re.findall(r'\(e\.g\. ([^)]*)\)', message)
        ]

    def test_cable_blocker_names_offending_component(self):
        peer = Interface.objects.create(
            device=self.device_a, name='peer0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        Cable(a_terminations=[self.interface], b_terminations=[peer]).save()
        # A second moved interface with no cable must not be named
        Interface.objects.create(
            device=self.device_a, module=self.module, name='quiet0',
            type=InterfaceTypeChoices.TYPE_1GE_FIXED,
        )
        message = self._blocked_message()
        self.assertIn('eth0', message)
        self.assertNotIn('quiet0', message)

    def test_interface_state_blocker_names_offending_interface(self):
        IPAddress.objects.create(address='192.0.2.1/24', assigned_object=self.interface)
        self.assertEqual(self._samples(self._blocked_message()), [['eth0']])

    def test_boundary_blocker_names_both_directions(self):
        outsider = Interface.objects.create(
            device=self.device_a, name='outsider0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        self.interface.bridge = outsider
        self.interface.save()
        lag = Interface.objects.create(
            device=self.device_a, module=self.module, name='lag0', type=InterfaceTypeChoices.TYPE_LAG
        )
        Interface.objects.create(
            device=self.device_a, name='member0', type=InterfaceTypeChoices.TYPE_1GE_FIXED, lag=lag
        )
        message = self._blocked_message()
        self.assertIn('2 parent, bridge, or LAG interface relations', message)
        self.assertEqual(self._samples(message), [['eth0', 'member0']])

    def test_split_power_outlet_blocker_names_outlet(self):
        power_port = PowerPort.objects.create(device=self.device_a, name='PP 1')
        PowerOutlet.objects.create(
            device=self.device_a, module=self.module, name='Outlet 1', power_port=power_port
        )
        self.assertEqual(self._samples(self._blocked_message()), [['Outlet 1']])

    def test_split_port_mapping_blocker_names_front_port(self):
        front_port = FrontPort.objects.create(
            device=self.device_a, module=self.module, name='Front 1', type=PortTypeChoices.TYPE_LC
        )
        rear_port = RearPort.objects.create(
            device=self.device_a, name='Rear 1', type=PortTypeChoices.TYPE_LC, positions=1
        )
        PortMapping.objects.create(
            front_port=front_port, front_port_position=1, rear_port=rear_port, rear_port_position=1
        )
        self.assertEqual(self._samples(self._blocked_message()), [['Front 1']])

    def test_split_port_mapping_blocker_names_moved_rear_port(self):
        """
        With the rear port moving and the front port staying behind, the sample must name the
        rear port: a user told to look for the front port would not find it on this module.
        """
        rear_port = RearPort.objects.create(
            device=self.device_a, module=self.module, name='Moved Rear 1',
            type=PortTypeChoices.TYPE_LC, positions=1,
        )
        front_port = FrontPort.objects.create(
            device=self.device_a, name='Chassis Front 1', type=PortTypeChoices.TYPE_LC
        )
        PortMapping.objects.create(
            front_port=front_port, front_port_position=1, rear_port=rear_port, rear_port_position=1
        )
        message = self._blocked_message()
        self.assertEqual(self._samples(message), [['Moved Rear 1']])
        self.assertNotIn('Chassis Front 1', message)

    def test_inventory_item_blocker_names_component(self):
        InventoryItem.objects.create(device=self.device_a, name='Item 1', component=self.interface)
        self.assertEqual(self._samples(self._blocked_message()), [['eth0']])

    def test_blocker_sample_is_capped_but_count_is_complete(self):
        peer = Interface.objects.create(
            device=self.device_a, name='peer0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        Cable(a_terminations=[self.interface], b_terminations=[peer]).save()
        for i in range(1, 8):
            marked = Interface.objects.create(
                device=self.device_a, module=self.module, name=f'eth{i}',
                type=InterfaceTypeChoices.TYPE_1GE_FIXED,
            )
            marked.mark_connected = True
            marked.save()
        message = self._blocked_message()
        self.assertIn('8 cabled or connection-marked interfaces', message)
        sample, = self._samples(message)
        self.assertEqual(len(sample), ModuleMovePlan.SAMPLE_LIMIT)
        self.assertEqual(sample, ['eth0', 'eth1', 'eth2', 'eth3', 'eth4'])

    def test_mac_address_is_allowed(self):
        mac = MACAddress.objects.create(mac_address='00:11:22:33:44:55', assigned_object=self.interface)
        self.interface.primary_mac_address = mac
        self.interface.save()
        self._assert_move_allowed()


class ModuleCrossDeviceMoveTestCase(TestCase):

    @classmethod
    def setUpTestData(cls):
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        role = DeviceRole.objects.create(name='Role 1', slug='role-1')
        cls.site_a = Site.objects.create(name='Site A', slug='site-a')
        cls.site_b = Site.objects.create(name='Site B', slug='site-b')
        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Chassis', slug='chassis')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 1', position='1')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 2', position='2')

        cls.line_card_type = ModuleType.objects.create(manufacturer=manufacturer, model='Line Card')
        InterfaceTemplate.objects.create(
            module_type=cls.line_card_type, name='Ethernet{module}/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        power_port_template = PowerPortTemplate.objects.create(module_type=cls.line_card_type, name='PP 1')
        PowerOutletTemplate.objects.create(
            module_type=cls.line_card_type, name='Outlet 1', power_port=power_port_template
        )
        front_port_template = FrontPortTemplate.objects.create(
            module_type=cls.line_card_type, name='Front 1', type=PortTypeChoices.TYPE_LC
        )
        rear_port_template = RearPortTemplate.objects.create(
            module_type=cls.line_card_type, name='Rear 1', type=PortTypeChoices.TYPE_LC, positions=1
        )
        PortTemplateMapping.objects.create(
            module_type=cls.line_card_type,
            front_port=front_port_template, front_port_position=1,
            rear_port=rear_port_template, rear_port_position=1,
        )
        ModuleBayTemplate.objects.create(
            module_type=cls.line_card_type, name='SFP bay {module}/1', position='{module}/1'
        )
        cls.sfp_type = ModuleType.objects.create(manufacturer=manufacturer, model='SFP')
        InterfaceTemplate.objects.create(
            module_type=cls.sfp_type, name='SFP {module}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
        )

        cls.device_a = Device.objects.create(
            name='Chassis A', device_type=device_type, role=role, site=cls.site_a
        )
        cls.device_b = Device.objects.create(
            name='Chassis B', device_type=device_type, role=role, site=cls.site_b
        )
        cls.slot_1_a = cls.device_a.modulebays.get(name='Slot 1')
        cls.slot_2_b = cls.device_b.modulebays.get(name='Slot 2')
        cls.line_card = Module.objects.create(
            device=cls.device_a, module_bay=cls.slot_1_a, module_type=cls.line_card_type
        )
        cls.sfp_bay = cls.line_card.modulebays.get(name='SFP bay 1/1')
        cls.sfp_module = Module.objects.create(
            device=cls.device_a, module_bay=cls.sfp_bay, module_type=cls.sfp_type
        )

    def _move_to_device_b(self):
        self.line_card.device = self.device_b
        self.line_card.module_bay = self.slot_2_b
        self.line_card.full_clean()
        self.line_card.save()

    def test_cross_device_move_updates_subtree(self):
        sfp_interface = self.sfp_module.interfaces.get(name='SFP 1/1')
        previous = sfp_interface.last_updated
        self._move_to_device_b()

        self.line_card.refresh_from_db()
        self.assertEqual(self.line_card.device, self.device_b)
        self.assertEqual(self.line_card.module_bay, self.slot_2_b)

        self.sfp_module.refresh_from_db()
        self.assertEqual(self.sfp_module.device, self.device_b)

        self.sfp_bay.refresh_from_db()
        self.assertEqual(self.sfp_bay.device, self.device_b)
        self.assertEqual(self.sfp_bay._site, self.site_b)
        self.assertEqual(self.sfp_bay.parent_id, self.slot_2_b.pk)
        self.assertTrue(str(self.sfp_bay.path).startswith(f'{self.slot_2_b.path}.'))

        sfp_interface.refresh_from_db()
        self.assertEqual(sfp_interface.device, self.device_b)
        self.assertEqual(sfp_interface._site, self.site_b)
        self.assertGreater(sfp_interface.last_updated, previous)

    def test_cross_device_move_renames_templated_components(self):
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        sfp_interface = self.sfp_module.interfaces.get(name='SFP 1/1')
        self._move_to_device_b()
        interface.refresh_from_db()
        self.assertEqual(interface.name, 'Ethernet2/1')
        sfp_interface.refresh_from_db()
        self.assertEqual(sfp_interface.name, 'SFP 2/1')

    def test_cross_device_move_applies_swap_chain_bay_renames(self):
        """
        A rename chain where one bay's new name equals another moved bay's current name
        applies cleanly cross-device, with both bays renamed on the destination device.
        """
        # Templates resolve Bay E1/1 -> Bay E2/1 and Bay E2/1 -> Bay E2/2 across slots
        ModuleBayTemplate.objects.create(module_type=self.line_card_type, name='Bay E2/{module}')
        ModuleBayTemplate.objects.create(module_type=self.line_card_type, name='Bay E{module}/1')
        bay_1 = ModuleBay.objects.create(device=self.device_a, module=self.line_card, name='Bay E1/1')
        bay_2 = ModuleBay.objects.create(device=self.device_a, module=self.line_card, name='Bay E2/1')
        self._move_to_device_b()
        bay_1.refresh_from_db()
        bay_2.refresh_from_db()
        self.assertEqual(bay_1.name, 'Bay E2/1')
        self.assertEqual(bay_2.name, 'Bay E2/2')
        self.assertEqual(bay_1.device, self.device_b)
        self.assertEqual(bay_2.device, self.device_b)

    def test_cross_device_move_updates_port_mappings(self):
        mapping = PortMapping.objects.get(front_port__module=self.line_card)
        self._move_to_device_b()
        mapping.refresh_from_db()
        self.assertEqual(mapping.device, self.device_b)

    @override_settings(BULK_UPDATE_CHUNK_SIZE=1)
    def test_cross_device_move_changelogs_port_mappings(self):
        front_port = FrontPort.objects.create(
            device=self.device_a, module=self.line_card, name='Front 2', type=PortTypeChoices.TYPE_LC
        )
        rear_port = RearPort.objects.create(
            device=self.device_a, module=self.line_card, name='Rear 2', type=PortTypeChoices.TYPE_LC, positions=1
        )
        PortMapping.objects.create(
            front_port=front_port, front_port_position=1, rear_port=rear_port, rear_port_position=1
        )
        mappings = PortMapping.objects.filter(front_port__module=self.line_card)
        self.assertEqual(mappings.count(), 2)
        request = RequestFactory().get('/')
        request.id = uuid.uuid4()
        request.user = User.objects.create_user(username='mapping-mover')
        with event_tracking(request):
            self.line_card.snapshot()
            self._move_to_device_b()
        for mapping in mappings:
            self.assertEqual(mapping.device, self.device_b)
            change = ObjectChange.objects.get(
                changed_object_type=ContentType.objects.get_for_model(PortMapping), changed_object_id=mapping.pk
            )
            self.assertEqual(change.prechange_data['device'], self.device_a.pk)
            self.assertEqual(change.postchange_data['device'], self.device_b.pk)

    def test_cross_device_move_keeps_intra_module_power_link(self):
        outlet = self.line_card.poweroutlets.get(name='Outlet 1')
        self._move_to_device_b()
        outlet.refresh_from_db()
        self.assertEqual(outlet.device, self.device_b)
        self.assertEqual(outlet.power_port.device_id, self.device_b.pk)

    def test_cross_device_move_moves_mac_address(self):
        interface = self.line_card.interfaces.get(name='Ethernet1/1')
        mac = MACAddress.objects.create(mac_address='00:11:22:33:44:55', assigned_object=interface)
        self._move_to_device_b()
        mac.refresh_from_db()
        interface.refresh_from_db()
        self.assertEqual(mac.assigned_object, interface)
        self.assertEqual(interface.device, self.device_b)

    def test_cross_device_move_recomputes_counters(self):
        self._move_to_device_b()
        self.device_a.refresh_from_db()
        self.device_b.refresh_from_db()
        self.assertEqual(self.device_a.interface_count, 0)
        self.assertEqual(self.device_b.interface_count, 2)
        self.assertEqual(self.device_a.module_bay_count, 2)
        self.assertEqual(self.device_b.module_bay_count, 3)

    def test_move_query_count_independent_of_destination_size(self):
        def move_and_count(destination_device, destination_bay):
            module = Module.objects.create(
                device=self.device_a, module_bay=self.device_a.modulebays.get(name='Slot 2'),
                module_type=self.sfp_type,
            )
            module.device = destination_device
            module.module_bay = destination_bay
            module.full_clean()
            with CaptureQueriesContext(connection) as ctx:
                module.save()
            return len(ctx.captured_queries)

        small_count = move_and_count(self.device_b, self.slot_2_b)
        big_device = Device.objects.create(
            name='Warehouse', device_type=self.device_b.device_type, role=self.device_b.role,
            site=self.site_b,
        )
        ModuleBay.objects.bulk_create([
            ModuleBay(device=big_device, name=f'Storage Bay {i}') for i in range(1, 201)
        ])
        target_bay = ModuleBay.objects.create(device=big_device, name='Target Bay')
        big_count = move_and_count(big_device, target_bay)
        self.assertEqual(small_count, big_count)

    def test_move_query_count_independent_of_same_type_child_count(self):
        """
        Moving a module with several installed children of the same module_type issues
        the same number of template-table queries as moving one with a single child: the
        per-pass template lookup is cached per module_type, not repeated per module.
        """
        multi_type = ModuleType.objects.create(
            manufacturer=self.line_card_type.manufacturer, model='Multi-Slot Card'
        )
        for i in range(1, 4):
            ModuleBayTemplate.objects.create(module_type=multi_type, name=f'Child Bay {i}', position=str(i))

        def build_and_move(child_count):
            device = create_test_device(f'Card Device {child_count}')
            card_bay = ModuleBay.objects.create(device=device, name='Card Bay')
            target_bay = ModuleBay.objects.create(device=device, name='Target Bay')
            card = Module.objects.create(device=device, module_bay=card_bay, module_type=multi_type)
            for child_bay in card.modulebays.order_by('name')[:child_count]:
                Module.objects.create(device=device, module_bay=child_bay, module_type=self.sfp_type)
            card.module_bay = target_bay
            card.full_clean()
            with CaptureQueriesContext(connection) as ctx:
                card.save()
            return sum(1 for query in ctx.captured_queries if 'template' in query['sql'].lower())

        one_child_queries = build_and_move(1)
        three_children_queries = build_and_move(3)
        self.assertEqual(one_child_queries, three_children_queries)

    def test_bulk_updates_use_configured_chunk_size(self):
        """
        The move path must honour BULK_UPDATE_CHUNK_SIZE rather than a private constant, so
        that an operator bounding rows-per-statement bounds this operation too.
        """
        original_bulk_update = QuerySet.bulk_update
        batch_sizes = []

        def recording_bulk_update(self, objs, fields, batch_size=None, **kwargs):
            batch_sizes.append(batch_size)
            return original_bulk_update(self, objs, fields, batch_size=batch_size, **kwargs)

        with override_settings(BULK_UPDATE_CHUNK_SIZE=7):
            with patch.object(QuerySet, 'bulk_update', recording_bulk_update):
                self._move_to_device_b()

        self.assertTrue(batch_sizes, 'the move issued no bulk_update calls')
        self.assertEqual(set(batch_sizes), {7})

    @override_settings(BULK_UPDATE_CHUNK_SIZE=1)
    def test_move_is_correct_when_updates_are_chunked(self):
        """
        A chunk size small enough to split every statement must not disturb the staged bay
        writes, whose correctness depends on the ltree triggers settling per level.
        """
        sfp_interface = self.sfp_module.interfaces.get(name='SFP 1/1')
        self._move_to_device_b()

        self.line_card.refresh_from_db()
        self.sfp_module.refresh_from_db()
        self.assertEqual(self.line_card.device, self.device_b)
        self.assertEqual(self.sfp_module.device, self.device_b)

        moved_bay = ModuleBay.objects.get(pk=self.sfp_bay.pk)
        dest_bay = ModuleBay.objects.get(pk=self.slot_2_b.pk)
        self.assertEqual(moved_bay.name, 'SFP bay 2/1')
        self.assertEqual(moved_bay.device, self.device_b)
        self.assertTrue(str(moved_bay.path).startswith(f'{dest_bay.path}.'))

        sfp_interface.refresh_from_db()
        self.assertEqual(sfp_interface.name, 'SFP 2/1')
        self.assertEqual(sfp_interface.device, self.device_b)
        self.assertEqual(sfp_interface._site, self.site_b)
        self.assertEqual(
            self.line_card.interfaces.get().name, 'Ethernet2/1'
        )

    def test_cross_device_move_refreshes_bay_sort_path(self):
        """
        The trigger-maintained sort_path of a moved nested bay reflects the
        destination hierarchy and the renamed chain after reparent plus rename.
        """
        old_sort_path = self.sfp_bay.sort_path
        self._move_to_device_b()
        moved_bay = ModuleBay.objects.get(pk=self.sfp_bay.pk)
        dest_bay = ModuleBay.objects.get(pk=self.slot_2_b.pk)
        self.assertEqual(moved_bay.name, 'SFP bay 2/1')
        self.assertTrue(str(moved_bay.path).startswith(f'{dest_bay.path}.'))
        self.assertNotEqual(moved_bay.sort_path, old_sort_path)
        self.assertTrue(str(moved_bay.sort_path).startswith(str(dest_bay.sort_path)))
        self.assertIn('SFP bay 2/1', str(moved_bay.sort_path))


class ModuleMoveComponentCoverageTestCase(TestCase):
    """
    Guard against a newly introduced modular component model being left out of the move
    planner, which is how cooling intakes and outflows were initially missed.
    """

    def test_every_modular_component_model_is_planned(self):
        # Scoped to dcim: a plugin may define its own ModularComponentModel subclass, which core
        # cannot add to COMPONENT_TEMPLATE_ATTRS, so it must not fail this assertion.
        core_models = {
            model for model in apps.get_models()
            if issubclass(model, ModularComponentModel) and model._meta.app_label == 'dcim'
        }
        # ModuleBay is planned separately (nested hierarchy, distinct uniqueness constraint).
        planned = set(COMPONENT_TEMPLATE_ATTRS) | {ModuleBay}
        self.assertEqual(
            core_models - planned, set(),
            'Modular component model(s) are not relocated by ModuleMovePlan. '
            'Add them to COMPONENT_TEMPLATE_ATTRS.'
        )

    def test_planned_template_attrs_exist_on_module_type(self):
        for model, template_attr in COMPONENT_TEMPLATE_ATTRS.items():
            with self.subTest(model=model._meta.label):
                self.assertTrue(
                    hasattr(ModuleType, template_attr),
                    f'ModuleType has no relation {template_attr!r}'
                )

    def test_device_counters_are_derived_for_every_planned_model(self):
        """
        Counter recomputation is derived from Device's own CounterCacheField declarations, so
        adding a modular component model cannot silently skip it. Assert the derivation still
        resolves a counter for each planned model, and does not reach beyond device-scoped ones.
        """
        counters = ModuleMovePlan.device_counters_by_model(Device)
        planned = set(COMPONENT_TEMPLATE_ATTRS) | {ModuleBay}
        self.assertEqual(
            planned - set(counters), set(),
            'A planned model has no device-scoped Device counter. If that is intended, this '
            'assertion needs to record the exception explicitly.'
        )
        self.assertEqual(counters[CoolingIntake], 'cooling_intake_count')
        self.assertEqual(counters[ModuleBay], 'module_bay_count')
        # Models counted by Device but never moved must not gain a delta
        self.assertNotIn(DeviceBay, planned)
        self.assertNotIn(InventoryItem, planned)


class ModuleMoveCounterTestCase(TestCase):
    """
    A cross-device move must adjust the Device counter of every modular component model the
    planner relocates. Exercises all of them at once, so a broken counter derivation cannot
    pass by covering only the component types other tests happen to create.
    """

    @classmethod
    def setUpTestData(cls):
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        role = DeviceRole.objects.create(name='Role 1', slug='role-1')
        site = Site.objects.create(name='Site A', slug='site-a')
        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Chassis', slug='chassis')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 1')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 2')

        # One template of every modular component type, so each planned model contributes a row
        cls.module_type = ModuleType.objects.create(manufacturer=manufacturer, model='Full Card')
        ConsolePortTemplate.objects.create(module_type=cls.module_type, name='Console 1')
        ConsoleServerPortTemplate.objects.create(module_type=cls.module_type, name='Console Server 1')
        CoolingIntakeTemplate.objects.create(module_type=cls.module_type, name='Intake 1')
        CoolingOutflowTemplate.objects.create(module_type=cls.module_type, name='Outflow 1')
        InterfaceTemplate.objects.create(
            module_type=cls.module_type, name='Ethernet 1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
        )
        power_port = PowerPortTemplate.objects.create(module_type=cls.module_type, name='PP 1')
        PowerOutletTemplate.objects.create(
            module_type=cls.module_type, name='Outlet 1', power_port=power_port
        )
        front_port = FrontPortTemplate.objects.create(
            module_type=cls.module_type, name='Front 1', type=PortTypeChoices.TYPE_LC
        )
        rear_port = RearPortTemplate.objects.create(
            module_type=cls.module_type, name='Rear 1', type=PortTypeChoices.TYPE_LC, positions=1
        )
        PortTemplateMapping.objects.create(
            module_type=cls.module_type,
            front_port=front_port, front_port_position=1,
            rear_port=rear_port, rear_port_position=1,
        )
        ModuleBayTemplate.objects.create(module_type=cls.module_type, name='Sub bay 1')

        cls.device_a = Device.objects.create(
            name='Chassis A', device_type=device_type, role=role, site=site
        )
        cls.device_b = Device.objects.create(
            name='Chassis B', device_type=device_type, role=role, site=site
        )

    def test_cross_device_move_adjusts_every_planned_counter(self):
        module = Module.objects.create(
            device=self.device_a, module_bay=self.device_a.modulebays.get(name='Slot 1'),
            module_type=self.module_type,
        )

        # Fail loudly rather than vacuously if the fixture stops covering every planned model
        rows = {model: model.objects.filter(module=module).count() for model in COMPONENT_TEMPLATE_ATTRS}
        rows[ModuleBay] = ModuleBay.objects.filter(module=module).count()
        self.assertEqual(
            set(rows.values()), {1},
            f'the fixture must create exactly one row per planned model, got {rows}'
        )

        counters = ModuleMovePlan.device_counters_by_model(Device)
        planned_counters = sorted(counters[model] for model in rows)
        self.device_a.refresh_from_db()
        self.device_b.refresh_from_db()
        before_a = {counter: getattr(self.device_a, counter) for counter in planned_counters}
        before_b = {counter: getattr(self.device_b, counter) for counter in planned_counters}

        module.device = self.device_b
        module.module_bay = self.device_b.modulebays.get(name='Slot 2')
        module.full_clean()
        module.save()

        self.device_a.refresh_from_db()
        self.device_b.refresh_from_db()
        for counter in planned_counters:
            with self.subTest(counter=counter):
                self.assertEqual(
                    getattr(self.device_a, counter), before_a[counter] - 1,
                    f'{counter} was not decremented on the source device'
                )
                self.assertEqual(
                    getattr(self.device_b, counter), before_b[counter] + 1,
                    f'{counter} was not incremented on the destination device'
                )


class ModuleMoveCoolingTestCase(TestCase):
    """
    Cooling intakes and outflows are modular components and must be relocated, renamed, and
    counted like any other. See netbox#15289.
    """

    @classmethod
    def setUpTestData(cls):
        manufacturer = Manufacturer.objects.create(name='Manufacturer M', slug='manufacturer-m')
        role = DeviceRole.objects.create(name='Role 1', slug='role-1')
        cls.site_a = Site.objects.create(name='Site A', slug='site-a')
        cls.site_b = Site.objects.create(name='Site B', slug='site-b')
        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Chassis', slug='chassis')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 1', position='1')
        ModuleBayTemplate.objects.create(device_type=device_type, name='Slot 2', position='2')

        cls.card_type = ModuleType.objects.create(manufacturer=manufacturer, model='Cooled Card')
        CoolingIntakeTemplate.objects.create(module_type=cls.card_type, name='Intake {module}/1')
        CoolingOutflowTemplate.objects.create(module_type=cls.card_type, name='Outflow {module}/1')
        ModuleBayTemplate.objects.create(
            module_type=cls.card_type, name='Sub bay {module}/1', position='{module}/1'
        )
        cls.sub_type = ModuleType.objects.create(manufacturer=manufacturer, model='Cooled Sub')
        CoolingIntakeTemplate.objects.create(module_type=cls.sub_type, name='Sub intake {module}')

        cls.device_a = Device.objects.create(
            name='Chassis A', device_type=device_type, role=role, site=cls.site_a
        )
        cls.device_b = Device.objects.create(
            name='Chassis B', device_type=device_type, role=role, site=cls.site_b
        )
        cls.slot_1_a = cls.device_a.modulebays.get(name='Slot 1')
        cls.slot_2_a = cls.device_a.modulebays.get(name='Slot 2')
        cls.slot_2_b = cls.device_b.modulebays.get(name='Slot 2')

    def setUp(self):
        super().setUp()
        self.card = Module.objects.create(
            device=self.device_a, module_bay=self.slot_1_a, module_type=self.card_type
        )
        self.intake = self.card.coolingintakes.get()
        self.outflow = self.card.coolingoutflows.get()

    def _move_to_device_b(self):
        self.card.device = self.device_b
        self.card.module_bay = self.slot_2_b
        self.card.full_clean()
        self.card.save()

    def test_same_device_move_renames_cooling_components(self):
        self.card.module_bay = self.slot_2_a
        self.card.full_clean()
        self.card.save()
        self.intake.refresh_from_db()
        self.outflow.refresh_from_db()
        self.assertEqual(self.intake.name, 'Intake 2/1')
        self.assertEqual(self.outflow.name, 'Outflow 2/1')
        self.assertEqual(self.intake.device, self.device_a)

    def test_cross_device_move_relocates_cooling_components(self):
        self._move_to_device_b()
        self.intake.refresh_from_db()
        self.outflow.refresh_from_db()
        for component in (self.intake, self.outflow):
            self.assertEqual(component.device, self.device_b)
            self.assertEqual(component._site, self.site_b)
            self.assertEqual(component._location, self.device_b.location)
            self.assertEqual(component._rack, self.device_b.rack)
        self.assertEqual(self.intake.name, 'Intake 2/1')
        self.assertEqual(self.outflow.name, 'Outflow 2/1')

    def test_cross_device_move_relocates_nested_cooling_components(self):
        sub_bay = self.card.modulebays.get()
        sub_module = Module.objects.create(
            device=self.device_a, module_bay=sub_bay, module_type=self.sub_type
        )
        sub_intake = sub_module.coolingintakes.get()
        self.assertEqual(sub_intake.name, 'Sub intake 1/1')
        self._move_to_device_b()
        sub_intake.refresh_from_db()
        self.assertEqual(sub_intake.device, self.device_b)
        self.assertEqual(sub_intake.name, 'Sub intake 2/1')

    def test_cross_device_move_recomputes_cooling_counters(self):
        self.device_a.refresh_from_db()
        self.assertEqual(self.device_a.cooling_intake_count, 1)
        self.assertEqual(self.device_a.cooling_outflow_count, 1)
        self._move_to_device_b()
        self.device_a.refresh_from_db()
        self.device_b.refresh_from_db()
        self.assertEqual(self.device_a.cooling_intake_count, 0)
        self.assertEqual(self.device_a.cooling_outflow_count, 0)
        self.assertEqual(self.device_b.cooling_intake_count, 1)
        self.assertEqual(self.device_b.cooling_outflow_count, 1)

    def test_reinstall_into_vacated_bay_after_move(self):
        """
        The vacated bay must be reusable: a stale cooling name left on the source device
        would collide with the replacement module's replicated components.
        """
        self._move_to_device_b()
        replacement = Module(
            device=self.device_a, module_bay=self.slot_1_a, module_type=self.card_type
        )
        replacement.full_clean()
        replacement.save()
        self.assertEqual(replacement.coolingintakes.get().name, 'Intake 1/1')

    def test_cooling_name_conflict_at_destination_is_rejected(self):
        CoolingIntake.objects.create(device=self.device_b, name='Intake 2/1')
        self.card.device = self.device_b
        self.card.module_bay = self.slot_2_b
        with self.assertRaises(ValidationError) as cm:
            self.card.full_clean()
        self.assertIn('would conflict with', str(cm.exception))

    def test_split_cooling_outflow_relation_blocks(self):
        """An outflow's upstream intake must stay on the same device, so it cannot be split."""
        device_intake = CoolingIntake.objects.create(device=self.device_a, name='Chassis Intake')
        self.outflow.cooling_intake = device_intake
        self.outflow.save()
        self.card.device = self.device_b
        self.card.module_bay = self.slot_2_b
        with self.assertRaises(ValidationError) as cm:
            self.card.full_clean()
        self.assertIn('cooling outflow relations crossing', str(cm.exception))
        self.assertIn('Outflow 1/1', str(cm.exception))

    def test_inward_cooling_outflow_relation_blocks(self):
        device_outflow = CoolingOutflow.objects.create(device=self.device_a, name='Chassis Outflow')
        device_outflow.cooling_intake = self.intake
        device_outflow.save()
        self.card.device = self.device_b
        self.card.module_bay = self.slot_2_b
        with self.assertRaises(ValidationError) as cm:
            self.card.full_clean()
        self.assertIn('cooling outflow relations crossing', str(cm.exception))
        self.assertIn('Chassis Outflow', str(cm.exception))

    def test_intra_module_cooling_pair_is_allowed(self):
        self.outflow.cooling_intake = self.intake
        self.outflow.save()
        self._move_to_device_b()
        self.outflow.refresh_from_db()
        self.assertEqual(self.outflow.device, self.device_b)
        self.assertEqual(self.outflow.cooling_intake, self.intake)

    def test_upstream_outflow_on_another_device_is_allowed(self):
        """
        CoolingIntake.cooling_outflow is not device-scoped: an intake is routinely supplied
        by an outflow on another device, such as a CDU. It must not block a move.
        """
        cdu_outflow = CoolingOutflow.objects.create(device=self.device_a, name='CDU Outflow')
        self.intake.cooling_outflow = cdu_outflow
        self.intake.save()
        self._move_to_device_b()
        self.intake.refresh_from_db()
        self.assertEqual(self.intake.device, self.device_b)
        self.assertEqual(self.intake.cooling_outflow, cdu_outflow)

    def test_cooling_components_are_changelogged(self):
        with event_tracking(self._make_request()):
            self.card.snapshot()
            self._move_to_device_b()
        intake_type = ContentType.objects.get_for_model(CoolingIntake)
        change = ObjectChange.objects.get(
            changed_object_type=intake_type, changed_object_id=self.intake.pk
        )
        self.assertEqual(change.prechange_data['name'], 'Intake 1/1')
        self.assertEqual(change.postchange_data['name'], 'Intake 2/1')
        self.assertEqual(change.postchange_data['device'], self.device_b.pk)

    def _make_request(self):
        request = RequestFactory().get('/')
        request.id = uuid.uuid4()
        request.user = User.objects.create_user(username='cooling-mover')
        return request

    def test_attached_inventory_item_on_cooling_component_blocks(self):
        InventoryItem.objects.create(
            device=self.device_a, name='Coolant Sensor', component=self.intake
        )
        self.card.device = self.device_b
        self.card.module_bay = self.slot_2_b
        with self.assertRaises(ValidationError) as cm:
            self.card.full_clean()
        self.assertIn('attached inventory items', str(cm.exception))