Custom script to move all connections from one device to another

[Supermathie]

We did a switch replacement recently and needed to move all connections from the "old switch" to the "new switch".

One way of effecting this would be changing the device itself (e.g. serial number) so we're not really "moving connections" we're "changing the device".

But if we wanted to keep history/maintenance log/etc. with the device itself, we'd want to move all the connections to the new device upon swap.

Does anyone know of an existing custom script to do this work, or shall I get to writing it?

To be precise about what exactly would happen:

pre-state connections:

• switch1/1 → switch2/1
• switch1/2 → server1/eth0
• server2/eth0 → switch1/3

operation: move_connections(switch1, switch1new)

post-state connections:

• switch1new/1 → switch2/1
• switch1new/2 → server1/eth0
• server2/eth0 → switch1new/3

[Supermathie]

Here's a first crack at making one; feedback welcome:

"""
Move connections from one device to another
"""

from dcim.models import Device, Interface
from extras.scripts import Script, ObjectVar

class MoveConnections(Script):
    class Meta:
        name = "Move Connections"
        description = "Move connections from one device to another"
        commit_default = False
        scheduling_enabled = False

    old_device = ObjectVar(model=Device, label="Old Device")
    new_device = ObjectVar(model=Device, label="New Device")

    def run(self, data, commit):
        old_device = data["old_device"]
        new_device = data["new_device"]

        for cable in old_device.get_cables():
            if cable.pk and hasattr(cable, 'snapshot'):
                cable.snapshot()

            if (n := len(cable.a_terminations)) != 1:
                self.log_failure(f"cable {cable.id} has {n} terminations on the A side, I cannot deal")
                continue
            if (n := len(cable.b_terminations)) != 1:
                self.log_failure(f"cable {cable.id} has {n} terminations on the A side, I cannot deal")
                continue
            
            # A side
            if cable.a_terminations[0].device == old_device:
                old_termination = cable.a_terminations[0]
                try:
                    new_termination = Interface.objects.get(name=old_termination.name, device=new_device)
                except (Interface.MultipleObjectsReturned, Interface.DoesNotExist) as e:
                    self.log_failure(f"cable {cable.id}: want to update A side but cannot find interface named `{old_termination.name}` on `{new_device.name}`: {e}")
                else:
                    self.log_info(f"cable {cable.id}: updating A side from `{old_termination.device}/{old_termination}` to `{new_termination.device}/{new_termination}`")
                    cable.a_terminations = [new_termination]

            # B side
            if cable.b_terminations[0].device == old_device:
                old_termination = cable.b_terminations[0]
                try:
                    new_termination = Interface.objects.get(name=old_termination.name, device=new_device)
                except (Interface.MultipleObjectsReturned, Interface.DoesNotExist) as e:
                    self.log_failure(f"cable {cable.id}: want to update B side but cannot find interface named {old_termination.name} on {new_device.name}: {e}")
                else:
                    self.log_info(f"cable {cable.id}: updating B side from `{old_termination.device}/{old_termination}` to `{new_termination.device}/{new_termination}`")
                    cable.b_terminations = [new_termination]
            
            try:
                cable.full_clean()
            except Exception as e:
                self.log_failure(f"[cable {cable.id}]({cable.get_absolute_url()}): validation error: {e}")
            else:
                cable.save()
                self.log_success(f"[cable {cable.id}]({cable.get_absolute_url()}): updated")