Commit b8fcb17c authored by NexAdn's avatar NexAdn
Browse files

patch: Add one-to-n bulk patch


Signed-off-by: default avatarNexAdn <git@nexadn.de>
parent 00568c9a
# Changelog
## Upcoming
### Additions
* `patch.py` One-to-N bulk patch (one device / multiple interfaces to multiple devices / one interface)
### Bugfixes
* `hausaufgang.py` Use new netbox choices instead of global constants which aren't available anymore in newer Netbox versions
......
......@@ -198,6 +198,175 @@ class BulkPatch(Script):
self.log_info("Not creating patches due to previous errors")
return
self.log_info("Creating %d patches" %len(port_pairs))
for port_pair in port_pairs:
a_port = port_pair[0]
b_port = port_pair[1]
self.log_info(f"{a_port} -> {b_port}")
if cable_label:
conn = Cable(termination_a = a_port,
termination_b = b_port,
label = cable_label
)
conn.save()
else:
conn = Cable(termination_a = a_port,
termination_b = b_port
)
conn.save()
self.log_success("Successfully patched %d connections"
%len(port_pairs))
class BulkPatchOneToN(Script):
class Meta:
name = "One-to-N bulk patch"
description = "Patch multiple ports from a single device to single ports from multiple devices"
PORT_TYPES = (
('i', 'Interface'),
('f', 'Front Port'),
('r', 'Rear Port')
)
dev_a = ObjectVar(
label = "Device A",
queryset = Device.objects
)
port_type_a = ChoiceVar(
label = "Device A port type",
choices = PORT_TYPES
)
port_type_b = ChoiceVar(
label = "Device B port type",
choices = PORT_TYPES
)
patch_pairs = TextVar(
label = "Patch pairs",
description = "Syntax: [A-Port] | [B-Device] | [B-Port]. Each patch pair on its own"
" line."
)
cable_label = StringVar(
label = "Cable label",
required = False,
description = "An optional label to set for each created cable"
)
def log_patch_pair_fail(self, line_no, line_txt, error_txt):
self.log_failure("Invalid patch pair on line %d: %s (%s)"
%(line_no, line_txt, error_txt))
def run(self, data, commit):
dev_a = data["dev_a"]
port_a_type = data["port_type_a"]
port_b_type = data["port_type_b"]
patch_pairs = data["patch_pairs"]
cable_label = data["cable_label"]
create_patches = True
patch_pair_no = 0
port_pairs = []
for patch_pair in patch_pairs.splitlines():
if patch_pair.find("|") == -1:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Missing separator")
create_patches = False
continue
patch_port = patch_pair.split(" | ")
if len(patch_port) != 3:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Invalid amount of separators")
create_patches = False
continue
dev_a_port = None
dev_b_port = None
if port_a_type == "i":
dev_a_port = Interface.objects.filter(
device = dev_a,
name__contains = patch_port[0]
)
elif port_a_type == "f":
dev_a_port = FrontPort.objects.filter(
device = dev_a,
name__contains = patch_port[0]
)
elif port_a_type == "r":
dev_a_port = RearPort.objects.filter(
device = dev_a,
name__contains = patch_port[0]
)
dev_b = Device.objects.get(name = patch_port[1])
if not dev_b:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Could not find Device B")
create_patches = False
patch_pair_no += 1
continue
if port_b_type == "i":
dev_b_port = Interface.objects.filter(
device = dev_b,
name__contains = patch_port[2]
)
elif port_b_type == "f":
dev_b_port = FrontPort.objects.filter(
device = dev_b,
name__contains = patch_port[2]
)
elif port_b_type == "r":
dev_b_port = RearPort.objects.filter(
device = dev_b,
name__contains = patch_port[2]
)
if not dev_a_port:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Could not find Port or Interface A")
if not dev_b_port:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Could not find Port or Interface B")
if len(dev_a_port) != 1:
found = False
for port in dev_a_port:
if port.name == patch_port[0]:
dev_a_port = [port]
found = True
if not found:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Interface A name is ambiguous")
dev_a_port = None
if len(dev_b_port) != 1:
found = False
for port in dev_b_port:
if port.name == patch_port[2]:
dev_b_port = [port]
found = True
if not found:
self.log_patch_pair_fail(patch_pair_no, patch_pair,
"Interface B name is ambiguous")
dev_b_port = None
if not dev_a_port or not dev_b_port:
create_patches = False
continue
port_pairs.append((dev_a_port[0], dev_b_port[0]))
patch_pair_no += 1
if not create_patches:
self.log_info("Not creating patches due to previous errors")
return
self.log_info("Creating %d patches" %len(port_pairs))
for port_pair in port_pairs:
a_port = port_pair[0]
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment