diff --git a/aquilon/netbox2aquilon.py b/aquilon/netbox2aquilon.py index a2d30e0..e01bf26 100755 --- a/aquilon/netbox2aquilon.py +++ b/aquilon/netbox2aquilon.py @@ -15,11 +15,18 @@ class Netbox2Aquilon(SCDNetbox): - """ Extends base SCDNetbox class with aquilon specific functionality """ + """ + Extends base SCDNetbox class with aquilon specific functionality + """ @classmethod def get_current_sandbox(cls): - """ Get owner and name of sandbox if command is being run while inside one """ + """ + Get owner and name of sandbox if command is being run while inside one + + :param cls: + :returns: Owner and name of sandbox + """ sandbox = None git_rev_parse = subprocess.run( ['git', 'rev-parse', '--show-toplevel'], @@ -35,6 +42,13 @@ def get_current_sandbox(cls): return sandbox def _call_aq(self, cmd): + """ + Calls aquilon and runs an process based on command + + :param self: + :param cmd: Aquilon commands + :returns: Process return code + """ logging.info('Calling %s', cmd[0]) logging.debug( 'Calling "%s %s"', @@ -67,6 +81,14 @@ def _call_aq(self, cmd): return process.returncode def _call_aq_cmds(self, cmds, dryrun=False): + """ + Call aquilon commands and returns list of commands committed in aquilon + + :param self: + :param cmds: Aquilon commands + :param dryrun: Boolean for whether to do a dryrun before applying commands + :returns: List of commands that were commmitted + """ cmds_committed = [] for cmd in cmds: if dryrun: @@ -85,6 +107,13 @@ def _call_aq_cmds(self, cmds, dryrun=False): return cmds_committed def _netbox_get_device(self, opts): + """ + Get device from Netbox based on device specification + + :param self: + :param opts: Option for how to get device with MagDB ID, device name, or hostname + :returns: Netbox device + """ if opts.magdb_id: device = self.get_device_by_magdb_id(opts.magdb_id) elif opts.netboxname: @@ -108,6 +137,13 @@ def _netbox_get_device(self, opts): return device def _netbox_copy_device(self, device): + """ + Copy details of the device in netbox and build command to copy machine into aquilon + + :param self: + :param device: Netbox device + :returns: Aquilon `add_machine` args for creating netbox device + """ cmds = [] # check if host is in rack - query netbox for rack @@ -133,6 +169,10 @@ def _netbox_copy_vm_disks(self, virtual_machine): Check if VM has any new-style virtual disks defined, If so, use them and set the first as bootable, If not, fall back to the classic single bootable disk method. + + :param self: + :param virtual_machine: Virtual machine in netbox + :returns: List of `cmd` args for `add_disk` to add disk to machine in aquilon """ cmds = [] @@ -166,6 +206,14 @@ def _netbox_copy_vm_disks(self, virtual_machine): return cmds def _netbox_copy_vm(self, virtual_machine): + """ + Copy virtual machine details from netbox and build cmd args to add the VM + into aquilon + + :param self: + :param virtual_machine: Virtual machine in netbox + :returns: List of `cmd` args to add VM in aquilon + """ cmds = [] if not virtual_machine.disk: @@ -195,6 +243,14 @@ def _netbox_copy_vm(self, virtual_machine): return cmds def _netbox_copy_interfaces(self, device): + """ + Copy interfaces for a netbox device and add the interface to the matching + machine in aquilon + + :param self: + :param device: Netbox device + :returns: List of `cmd` args to add interfaces to the machine in aquilon + """ cmds = [] interfaces = self.get_interfaces_from_device(device) for interface in interfaces: @@ -244,6 +300,14 @@ def _netbox_copy_interfaces(self, device): return cmds def _netbox_copy_addresses(self, device): + """ + Copy IP addresses for a netbox device and add the addresses to the matching + machine in aquilon + + :param self: + :param device: Netbox device + :returns: List of `cmd` args to add addresses to the machine in aquilon + """ cmds = [] interfaces = self.get_interfaces_from_device(device) for interface in interfaces: @@ -267,6 +331,15 @@ def _netbox_copy_addresses(self, device): return cmds def _netbox_get_personality(self, device, archetype, personality=None): + """ + Set the personality to use for a netbox device in aquilon. + + :param self: + :param device: Netbox device + :param archetype: Aquilon archetype + :param personality: Aquilon personality, by default it is `None` + :returns: Personality to set for the machine in aquilon + """ if not personality: personality = 'inventory' @@ -292,7 +365,13 @@ def _netbox_get_personality(self, device, archetype, personality=None): return personality def netbox_copy(self, opts): - """ Copy a device from NetBox to Aquilon """ + """ + Copy a device from NetBox to Aquilon + + :param self: + :param opts: Option for how to get device in netbox + :returns: Process code after executing aquilon command + """ device = self._netbox_get_device(opts) aqdesttype = None @@ -374,6 +453,13 @@ def netbox_copy(self, opts): @classmethod def _undo_cmds(cls, cmds_run): + """ + Class method to reverse changes made with aquilon commands + + :param cls: + :param cmds_run: List of commands that had been run + :returns: List of commands that have been undone in aquilon + """ cmds_undone = [] # Map of required arguments for delete commands @@ -401,6 +487,10 @@ def _undo_cmds(cls, cmds_run): def _main(): + """ + Main method calling `Netbox2Aquilon` class and adding arguments that can be parsed in aquilon + and runs `netbox_copy` to copy device into aquilon from netbox + """ logging.basicConfig(format='%(levelname)s: %(message)s') netbox2aquilon = Netbox2Aquilon(additonal_config_name='netbox2aquilon') diff --git a/aquilon/netbox_dump_subnetdata.py b/aquilon/netbox_dump_subnetdata.py index ac1ed56..5fd63cc 100755 --- a/aquilon/netbox_dump_subnetdata.py +++ b/aquilon/netbox_dump_subnetdata.py @@ -14,7 +14,9 @@ class NetboxDumpSubnetdata(SCDNetbox): - """ Extends base SCDNetbox class with functionality to dump subnets to a file """ + """ + Extends base SCDNetbox class with functionality to dump subnets to a file + """ def __init__(self): super().__init__() if 'dump_subnetdata' not in self.config: @@ -23,11 +25,17 @@ def __init__(self): self.config['dump_subnetdata']['tenants'] = 'tier1,cloud,secops' def _get_subnet_fields(self): + """ + Get all IPv4 prefixes for configured tenants + Aquilon doesn't support syncing IPv6 prefixes via this method + We only want prefixes without child prefixes + Only synchronise the Global VRF which corresponds to the aquilon "internal" network environment + + :param self: + :returns: IPv4 prefixes for configured tenants + """ results = [] - # Get all IPv4 prefixes for configured tenants - # Aquilon doesn't support syncing IPv6 prefixes via this method - # We only want prefixes without child prefixes - # Only synchronise the Global VRF which corresponds to the aquilon "internal" network environment + tenants = [t.strip() for t in self.config['dump_subnetdata']['tenants'].split(',')] for prefix in self.netbox.ipam.prefixes.filter(tenant=tenants, family=4, children=0, vrf_id=None): subnet_name = prefix.description @@ -63,6 +71,10 @@ def write_subnetdata_txt(self, directory): - A field is a key/value pair, separated by a space - The value of the DefaultRouters field is a comma-separated list of IP addresses - The value of the UDF field is a list of "=" pairs, separated by ';' + + :param self: + :param directory: Directory to write subnet data to + :returns: None """ lines = [] subnet_fields = self._get_subnet_fields() @@ -77,13 +89,24 @@ def write_subnetdata_txt(self, directory): dumpfile.writelines(lines) def write_subnetdata_json(self, directory): - """ Dump subnetdata field structure in JSON format """ + """ + Dump subnetdata field structure in JSON format + + :param self: + :param directory: Directory to write data to + :returns: None + """ subnet_fields = self._get_subnet_fields() with open(os.path.join(directory, 'subnetdata.json'), 'w', encoding='utf-8') as dumpfile: json.dump(subnet_fields, dumpfile) def _main(): + """ + Main method to collect subnetdata and store in text and json files + + :returns: None + """ logging.basicConfig(format='%(levelname)s: %(message)s') netbox_dump_subnetdata = NetboxDumpSubnetdata() diff --git a/aquilon/scd_netbox.py b/aquilon/scd_netbox.py index 5c52fee..54b3b5b 100644 --- a/aquilon/scd_netbox.py +++ b/aquilon/scd_netbox.py @@ -53,7 +53,14 @@ def __init__(self, additonal_config_name=None): self.netbox.http_session = netbox_session def get_device_by_magdb_id(self, magdb_id): - """ Get a single device from NetBox based on MagDB system ID """ + """ + Get a single device from NetBox based on MagDB system ID + + :param self: + :param magdb_id: MagDB ID + + :returns: Netbox device that matches MagDB ID + """ device = self.netbox.dcim.devices.get(cf_magdb_system_id=magdb_id) if device is None: @@ -64,7 +71,12 @@ def get_device_by_magdb_id(self, magdb_id): return device def get_device_by_name(self, name): - """ Get a single device from NetBox based on device name """ + """ + Get a single device from netbox based on device name + + :param self: + :param name: Name of device to get from netbox + """ device = self.netbox.dcim.devices.get(name=name) if device is None: @@ -75,7 +87,13 @@ def get_device_by_name(self, name): return device def get_device_by_hostname(self, hostname): - """ Get a single device from NetBox based on fully qualified domain name """ + """ + Get a single device from NetBox based on fully qualified domain name + + :param self: + :param hostname: Fully qualified domain name + :returns: Netbox device matching the FQDN + """ ip_addresses = self.netbox.ipam.ip_addresses.filter(dns_name=hostname, family=4) if ip_addresses is None: logging.error("Hostname not found in NetBox") @@ -111,7 +129,12 @@ def get_device_by_hostname(self, hostname): return device def get_rack_from_device(self, device): - """ check if host is in rack - query netbox for rack """ + """ + Check if host is in rack - query netbox for rack + + :param self: + :param device: Netbox device to find rack location + """ rack = self.netbox.dcim.racks.get(device.rack.id) if rack is None: @@ -130,6 +153,10 @@ def get_interfaces_from_device(self, device): The next step is to add interfaces to the machine in Aquilon We need to use 'filter' to retrieve the interface id for all interfaces This will assume that the device name IS UNIQUE + + :param self: + :param device: Netbox device + :returns: List of interfaces for the device """ if isinstance(device, pynetbox.models.dcim.Devices): filter_interfaces = self.netbox.dcim.interfaces.filter(device=device.name) @@ -159,7 +186,13 @@ def get_interfaces_from_device(self, device): return interfaces def get_addresses_from_interface(self, interface): - """ Get all address objects associated with a physical or virtual interface """ + """ + Get all address objects associated with a physical or virtual interface + + :param self: + :param interface: Interface associated wiht a device from netbox + :returns: List of IP addresses associated with the interface + """ if interface.count_ipaddresses == 0: return [] @@ -189,6 +222,10 @@ def get_addresses_from_interface(self, interface): def get_disks_from_device(self, device): """ Get all virtual disks associated with a virtual machine + + :param self: + :param device: Netbox virtual machine device + :returns: List of disks associated with the VM """ if isinstance(device, pynetbox.models.virtualization.VirtualMachines): filtered_disks = self.netbox.virtualization.virtual_disks.filter(virtual_machine_id=device.id) diff --git a/aquilon/test_netbox2aquilon.py b/aquilon/test_netbox2aquilon.py index 99ef67e..a624b84 100644 --- a/aquilon/test_netbox2aquilon.py +++ b/aquilon/test_netbox2aquilon.py @@ -17,6 +17,9 @@ def test_get_current_sandbox(mocker): + """ + Tests the sandbox can be found if present and returns none otherwise + """ test_obj = Netbox2Aquilon() # git rev-parse succeeds @@ -45,6 +48,9 @@ def test_get_current_sandbox(mocker): def test__netbox_copy_interfaces(mocker): + """ + Tests that interfaces from device are copied into args for aquilon command + """ test_obj = Netbox2Aquilon() # Physical devices with a management interface @@ -122,6 +128,9 @@ def test__netbox_copy_interfaces(mocker): def test__netbox_copy_addresses(mocker): + """ + Tests that addresses from device are copied into args for aquilon command + """ test_obj = Netbox2Aquilon() fake_device = FAKE.DEVICE_PHYSICAL @@ -146,6 +155,9 @@ def test__netbox_copy_addresses(mocker): def test__netbox_get_personality(mocker): + """ + Tests that personality for device is defined based on device role + """ test_obj = Netbox2Aquilon() fake_devices = [ @@ -208,6 +220,9 @@ def test__netbox_get_personality(mocker): def test__undo_cmds(): + """ + Test that commands that were run in netbox2aquilon can be reversed + """ test_obj = Netbox2Aquilon() cmds_forward = [ @@ -304,6 +319,9 @@ def test__undo_cmds(): def test__netbox_copy_vm_disks(mocker): + """ + Tests that VM disks from device are copied into args for aquilon command + """ test_obj = Netbox2Aquilon() test_obj.get_disks_from_device = mocker.MagicMock(return_value=deepcopy(FAKE.DISKS_VIRTUAL)) diff --git a/aquilon/test_netbox_dump_subnetdata.py b/aquilon/test_netbox_dump_subnetdata.py index b3a4c32..8527241 100644 --- a/aquilon/test_netbox_dump_subnetdata.py +++ b/aquilon/test_netbox_dump_subnetdata.py @@ -18,6 +18,9 @@ def test__get_subnet_fields(mocker): + """ + Test IPv4 prefixes for configured tenants can be retrieved from netbox + """ test_obj = NetboxDumpSubnetdata() test_obj.netbox.ipam.prefixes = SimpleNamespace() @@ -36,6 +39,9 @@ def test__get_subnet_fields(mocker): def test_write_subnetdata_txt(mocker): + """ + Test subnet data can be written to a text file + """ test_obj = NetboxDumpSubnetdata() test_obj.netbox.ipam.prefixes = SimpleNamespace() @@ -51,6 +57,9 @@ def test_write_subnetdata_txt(mocker): def test_write_subnetdata_json(mocker): + """ + Test subnet data can be written to a JSON file + """ test_obj = NetboxDumpSubnetdata() test_obj.netbox.ipam.prefixes = SimpleNamespace() diff --git a/aquilon/test_scd_netbox.py b/aquilon/test_scd_netbox.py index 29f2900..5075372 100644 --- a/aquilon/test_scd_netbox.py +++ b/aquilon/test_scd_netbox.py @@ -152,7 +152,9 @@ def test_get_rack_from_device(mocker): def test_get_interfaces_from_device(mocker): - """ Test get_interfaces_from_device """ + """ + Test get_interfaces_from_device + """ test_obj = SCDNetbox() # Test physical interfaces @@ -176,7 +178,9 @@ def test_get_interfaces_from_device(mocker): def test_get_addresses_from_interface(mocker): - """ Test get_addresses_from_interface """ + """ + Test get_addresses_from_interface + """ scd_netbox = SCDNetbox() # Should filter by interface_id if physical