mirror of
https://github.com/drygdryg/netbox-plugin-interface-sync
synced 2024-11-25 18:10:52 +03:00
423 lines
22 KiB
Python
423 lines
22 KiB
Python
from django.db.models.fields.related import ForeignKey
|
|
from django.shortcuts import get_object_or_404, render, redirect
|
|
from django.views.generic import View
|
|
from dcim.choices import PortTypeChoices
|
|
from dcim.models import Device, Interface, InterfaceTemplate, PowerPort, PowerPortTemplate, ConsolePort, ConsolePortTemplate, ConsoleServerPort, ConsoleServerPortTemplate, DeviceBay, DeviceBayTemplate, FrontPort, FrontPortTemplate,PowerOutlet, PowerOutletTemplate, RearPort, RearPortTemplate
|
|
from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin
|
|
from django.conf import settings
|
|
from django.contrib import messages
|
|
|
|
from .utils import natural_keys, get_components, post_components
|
|
from .comparison import PowerPortComparison, PowerOutletComparison, InterfaceComparison, ConsolePortComparison, ConsoleServerPortComparison, DeviceBayComparison
|
|
from .forms import InterfaceComparisonForm
|
|
|
|
config = settings.PLUGINS_CONFIG['netbox_interface_sync']
|
|
|
|
|
|
class InterfaceComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
interfaces = device.vc_interfaces()
|
|
if config["exclude_virtual_interfaces"]:
|
|
interfaces = list(filter(lambda i: not i.is_virtual, interfaces))
|
|
interface_templates = InterfaceTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
unified_interfaces = [InterfaceComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), i.mgmt_only) for i in interfaces]
|
|
unified_interface_templates = [
|
|
InterfaceComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), i.mgmt_only, is_template=True) for i in interface_templates]
|
|
|
|
return get_components(request, device, interfaces, unified_interfaces, unified_interface_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
interfaces = device.vc_interfaces()
|
|
if config["exclude_virtual_interfaces"]:
|
|
interfaces = interfaces.exclude(type__in=["virtual", "lag"])
|
|
interface_templates = InterfaceTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
# Getting and validating a list of interfaces to rename
|
|
fix_name_components = filter(
|
|
lambda i: str(i.id) in request.POST.getlist("fix_name"), interfaces
|
|
)
|
|
|
|
unified_interface_templates = [
|
|
InterfaceComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), i.mgmt_only, is_template=True) for i in interface_templates]
|
|
|
|
unified_interfaces = []
|
|
|
|
for component in fix_name_components:
|
|
unified_interfaces.append((component, InterfaceComparison(
|
|
component.id,
|
|
component.name,
|
|
component.label,
|
|
component.description,
|
|
component.type,
|
|
component.get_type_display(),
|
|
component.mgmt_only)))
|
|
|
|
return post_components(request, device, interfaces, interface_templates, Interface, InterfaceTemplate, unified_interfaces, unified_interface_templates)
|
|
|
|
class PowerPortComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
powerports = device.powerports.all()
|
|
powerports_templates = PowerPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
unified_powerports = [PowerPortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), i.maximum_draw, i.allocated_draw) for i in powerports]
|
|
unified_powerport_templates = [
|
|
PowerPortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), i.maximum_draw, i.allocated_draw, is_template=True) for i in powerports_templates]
|
|
|
|
return get_components(request, device, powerports, unified_powerports, unified_powerport_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
powerports = device.powerports.all()
|
|
powerports_templates = PowerPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
# Getting and validating a list of interfaces to rename
|
|
fix_name_components = filter(
|
|
lambda i: str(i.id) in request.POST.getlist("fix_name"), powerports
|
|
)
|
|
|
|
unified_powerport_templates = [
|
|
PowerPortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), i.maximum_draw, i.allocated_draw, is_template=True) for i in powerports_templates]
|
|
|
|
unified_powerports = []
|
|
|
|
for component in fix_name_components:
|
|
unified_powerports.append((component, PowerPortComparison(
|
|
component.id,
|
|
component.name,
|
|
component.label,
|
|
component.description,
|
|
component.type,
|
|
component.get_type_display(),
|
|
component.maximum_draw,
|
|
component.allocated_draw)))
|
|
|
|
return post_components(request, device, powerports, powerports_templates, PowerPort, PowerPortTemplate, unified_powerports, unified_powerport_templates)
|
|
|
|
class ConsolePortComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
consoleports = device.consoleports.all()
|
|
consoleports_templates = ConsolePortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
unified_consoleports = [ConsolePortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display()) for i in consoleports]
|
|
unified_consoleport_templates = [
|
|
ConsolePortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), is_template=True) for i in consoleports_templates]
|
|
|
|
return get_components(request, device, consoleports, unified_consoleports, unified_consoleport_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
consoleports = device.consoleports.all()
|
|
consoleports_templates = ConsolePortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
# Getting and validating a list of interfaces to rename
|
|
fix_name_components = filter(
|
|
lambda i: str(i.id) in request.POST.getlist("fix_name"), consoleports
|
|
)
|
|
|
|
unified_consoleport_templates = [
|
|
ConsolePortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), is_template=True) for i in consoleports_templates]
|
|
|
|
unified_consoleports = []
|
|
|
|
for component in fix_name_components:
|
|
unified_consoleports.append((component, ConsolePortComparison(
|
|
component.id,
|
|
component.name,
|
|
component.label,
|
|
component.description,
|
|
component.type,
|
|
component.get_type_display())))
|
|
|
|
return post_components(request, device, consoleports, consoleports_templates, ConsolePort, ConsolePortTemplate, unified_consoleports, unified_consoleport_templates)
|
|
|
|
class ConsoleServerPortComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
consoleserverports = device.consoleserverports.all()
|
|
consoleserverports_templates = ConsoleServerPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
unified_consoleserverports = [ConsoleServerPortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display()) for i in consoleserverports]
|
|
unified_consoleserverport_templates = [
|
|
ConsoleServerPortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), is_template=True) for i in consoleserverports_templates]
|
|
|
|
return get_components(request, device, consoleserverports, unified_consoleserverports, unified_consoleserverport_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
consoleserverports = device.consoleserverports.all()
|
|
consoleserverports_templates = ConsoleServerPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
# Getting and validating a list of interfaces to rename
|
|
fix_name_components = filter(
|
|
lambda i: str(i.id) in request.POST.getlist("fix_name"), consoleserverports
|
|
)
|
|
|
|
unified_consoleserverport_templates = [
|
|
ConsoleServerPortComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), is_template=True) for i in consoleserverports_templates]
|
|
|
|
unified_consoleserverports = []
|
|
|
|
for component in fix_name_components:
|
|
unified_consoleserverports.append((component, ConsoleServerPortComparison(
|
|
component.id,
|
|
component.name,
|
|
component.label,
|
|
component.description,
|
|
component.type,
|
|
component.get_type_display())))
|
|
|
|
return post_components(request, device, consoleserverports, consoleserverports_templates, ConsoleServerPort, ConsoleServerPortTemplate, unified_consoleserverports, unified_consoleserverport_templates)
|
|
|
|
class PowerOutletComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
poweroutlets = device.poweroutlets.all()
|
|
poweroutlets_templates = PowerOutletTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
unified_poweroutlets = [PowerOutletComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), power_port_name=PowerPort.objects.get(id=i.power_port_id).name if i.power_port_id is not None else "", feed_leg=i.feed_leg) for i in poweroutlets]
|
|
unified_poweroutlet_templates = [
|
|
PowerOutletComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), power_port_name=PowerPortTemplate.objects.get(id=i.power_port_id).name if i.power_port_id is not None else "", feed_leg=i.feed_leg, is_template=True) for i in poweroutlets_templates]
|
|
|
|
return get_components(request, device, poweroutlets, unified_poweroutlets, unified_poweroutlet_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
poweroutlets = device.poweroutlets.all()
|
|
poweroutlets_templates = PowerOutletTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
# Generating result message
|
|
message = []
|
|
created = 0
|
|
updated = 0
|
|
fixed = 0
|
|
|
|
remove_from_device = filter(
|
|
lambda i: i in poweroutlets.values_list("id", flat=True),
|
|
map(int, filter(lambda x: x.isdigit(), request.POST.getlist("remove_from_device")))
|
|
)
|
|
|
|
# Remove selected interfaces from the device and count them
|
|
deleted = PowerOutlet.objects.filter(id__in=remove_from_device).delete()[0]
|
|
|
|
# Get device power ports to check dependency between power outlets
|
|
device_pp = PowerPort.objects.filter(device_id=device.id)
|
|
|
|
matching = {}
|
|
mismatch = False
|
|
for i in poweroutlets_templates:
|
|
found = False
|
|
if i.power_port_id is not None:
|
|
ppt = PowerPortTemplate.objects.get(id=i.power_port_id)
|
|
for pp in device_pp:
|
|
if pp.name == ppt.name:
|
|
# Save matching to add the correct power port later
|
|
matching[i.id] = pp.id
|
|
found = True
|
|
|
|
# If at least on power port is found there is a dependency
|
|
# Better not to sync at all
|
|
if not found:
|
|
mismatch = True
|
|
break
|
|
|
|
if not mismatch:
|
|
# Manually validating interfaces and interface templates lists
|
|
add_to_device = filter(
|
|
lambda i: i in poweroutlets_templates.values_list("id", flat=True),
|
|
map(int, filter(lambda x: x.isdigit(), request.POST.getlist("add_to_device")))
|
|
)
|
|
|
|
# Add selected interfaces to the device and count them
|
|
add_to_device_component = PowerOutletTemplate.objects.filter(id__in=add_to_device)
|
|
|
|
bulk_create = []
|
|
updated = 0
|
|
keys_to_avoid = ["id"]
|
|
|
|
for i in add_to_device_component.values():
|
|
to_create = False
|
|
|
|
try:
|
|
# If power outlets already exists, update and do not recreate
|
|
po = device.poweroutlets.get(name=i["name"])
|
|
except PowerOutlet.DoesNotExist:
|
|
po = PowerOutlet()
|
|
po.device = device
|
|
to_create = True
|
|
|
|
# Copy all fields from template
|
|
for k in i.keys():
|
|
if k not in keys_to_avoid:
|
|
setattr(po, k, i[k])
|
|
po.power_port_id = matching.get(i["id"], None)
|
|
|
|
if to_create:
|
|
bulk_create.append(po)
|
|
else:
|
|
po.save()
|
|
updated += 1
|
|
|
|
created = len(PowerOutlet.objects.bulk_create(bulk_create))
|
|
|
|
# Getting and validating a list of interfaces to rename
|
|
fix_name_components = filter(lambda i: str(i.id) in request.POST.getlist("fix_name"), poweroutlets)
|
|
|
|
# Casting interface templates into UnifiedInterface objects for proper comparison with interfaces for renaming
|
|
unified_component_templates = [
|
|
PowerOutletComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), power_port_name=PowerPortTemplate.objects.get(id=i.power_port_id).name if i.power_port_id is not None else "", feed_leg=i.feed_leg, is_template=True) for i in poweroutlets_templates]
|
|
|
|
# Rename selected interfaces
|
|
fixed = 0
|
|
for component in fix_name_components:
|
|
unified_component = [PowerOutletComparison(i.id, i.name, i.label, i.description, i.type, i.get_type_display(), power_port_name=PowerPort.objects.get(id=i.power_port_id).name if i.power_port_id is not None else "", feed_leg=i.feed_leg) for i in poweroutlets]
|
|
|
|
try:
|
|
# Try to extract an interface template with the corresponding name
|
|
corresponding_template = unified_component_templates[unified_component_templates.index(unified_component)]
|
|
component.name = corresponding_template.name
|
|
component.save()
|
|
fixed += 1
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
message.append("Dependecy detected, sync power ports first!")
|
|
|
|
if created > 0:
|
|
message.append(f"created {created} interfaces")
|
|
if updated > 0:
|
|
message.append(f"updated {updated} interfaces")
|
|
if deleted > 0:
|
|
message.append(f"deleted {deleted} interfaces")
|
|
if fixed > 0:
|
|
message.append(f"fixed {fixed} interfaces")
|
|
|
|
messages.info(request, "; ".join(message).capitalize())
|
|
|
|
return redirect(request.path)
|
|
|
|
class FrontPortComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
frontports = device.frontports.all()
|
|
frontports_templates = FrontPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
return get_components(request, device, frontports, frontports_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
frontports = device.frontports.all()
|
|
frontports_templates = FrontPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
return post_components(request, device, frontports, frontports_templates, FrontPort, FrontPortTemplate)
|
|
|
|
class RearPortComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
rearports = device.rearports.all()
|
|
rearports_templates = RearPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
return get_components(request, device, rearports, rearports_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
rearports = device.rearports.all()
|
|
rearports_templates = RearPortTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
return post_components(request, device, rearports, rearports_templates, RearPort, RearPortTemplate)
|
|
|
|
class DeviceBayComparisonView(LoginRequiredMixin, PermissionRequiredMixin, View):
|
|
"""Comparison of interfaces between a device and a device type and beautiful visualization"""
|
|
permission_required = ("dcim.view_interface", "dcim.add_interface", "dcim.change_interface", "dcim.delete_interface")
|
|
|
|
def get(self, request, device_id):
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
devicebays = device.devicebays.all()
|
|
devicebays_templates = DeviceBayTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
unified_devicebays = [DeviceBayComparison(i.id, i.name, i.label, i.description) for i in devicebays]
|
|
unified_devicebay_templates = [
|
|
DeviceBayComparison(i.id, i.name, i.label, i.description, is_template=True) for i in devicebays_templates]
|
|
|
|
return get_components(request, device, devicebays, unified_devicebays, unified_devicebay_templates)
|
|
|
|
def post(self, request, device_id):
|
|
form = InterfaceComparisonForm(request.POST)
|
|
if form.is_valid():
|
|
device = get_object_or_404(Device.objects.filter(id=device_id))
|
|
|
|
devicebays = device.devicebays.all()
|
|
devicebays_templates = DeviceBayTemplate.objects.filter(device_type=device.device_type)
|
|
|
|
# Getting and validating a list of devicebays to rename
|
|
fix_name_components = filter(
|
|
lambda i: str(i.id) in request.POST.getlist("fix_name"), devicebays
|
|
)
|
|
|
|
unified_devicebay_templates = [
|
|
DeviceBayComparison(i.id, i.name, i.label, i.description, is_template=True) for i in devicebays_templates]
|
|
|
|
unified_devicebays = []
|
|
|
|
for component in fix_name_components:
|
|
unified_devicebays.append((component, DeviceBayComparison(
|
|
component.id,
|
|
component.name,
|
|
component.label,
|
|
component.description
|
|
)))
|
|
|
|
return post_components(request, device, devicebays, devicebays_templates, DeviceBay, DeviceBayTemplate, unified_devicebays, unified_devicebay_templates)
|