diff --git a/.github/workflows/pynetbox_codecov.yaml b/.github/workflows/pynetbox.yaml similarity index 72% rename from .github/workflows/pynetbox_codecov.yaml rename to .github/workflows/pynetbox.yaml index e15fbb03..3ed45176 100644 --- a/.github/workflows/pynetbox_codecov.yaml +++ b/.github/workflows/pynetbox.yaml @@ -1,4 +1,4 @@ -name: Pynetbox Codecov +name: Pylint-Tests-Codecov on: push: @@ -10,7 +10,7 @@ on: - ".github/workflows/pynetbox.yaml" jobs: - Codecov: + Pylint-Tests-Codecov: runs-on: ubuntu-20.04 strategy: matrix: @@ -28,10 +28,15 @@ jobs: python -m pip install --upgrade pip cd pynetbox_query pip install -r requirements.txt - + + - name: Analyse with pylint + run: cd pynetbox_query && pylint $(git ls-files '*.py') + + - name: Run tests and collect coverage + run: cd pynetbox_query && python3 -m pytest - name: Run tests and collect coverage - run: cd pynetbox_query && python3 -m pytest --cov-report xml:coverage.xml --cov + run: cd pynetbox_query && python3 -m pytest . --cov-report xml:coverage.xml --cov - name: Upload coverage to Codecov uses: codecov/codecov-action@v3 diff --git a/.github/workflows/pynetbox_pylint.yaml b/.github/workflows/pynetbox_pylint.yaml deleted file mode 100644 index d3ef6377..00000000 --- a/.github/workflows/pynetbox_pylint.yaml +++ /dev/null @@ -1,35 +0,0 @@ -name: Pynetbox Pylint - -on: - push: - branches: - - master - pull_request: - paths: - - "pynetbox_data_uploader/**" - - ".github/workflows/pynetbox.yaml" - -jobs: - Pylint: - runs-on: ubuntu-20.04 - strategy: - matrix: - python-version: ["3.x"] - steps: - - uses: actions/checkout@v3 - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v3 - with: - python-version: ${{ matrix.python-version }} - cache: "pip" - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - cd pynetbox_query - pip install -r requirements.txt - - - - - name: Analyse with pylint - run: cd pynetbox_query && pylint $(git ls-files '*.py') \ No newline at end of file diff --git a/.github/workflows/pynetbox_tests.yaml b/.github/workflows/pynetbox_tests.yaml deleted file mode 100644 index bf1e2cb3..00000000 --- a/.github/workflows/pynetbox_tests.yaml +++ /dev/null @@ -1,34 +0,0 @@ -name: Pynetbox Tests - -on: - push: - branches: - - master - pull_request: - paths: - - "pynetbox_data_uploader/**" - - ".github/workflows/pynetbox.yaml" - -jobs: - Test: - runs-on: ubuntu-20.04 - strategy: - matrix: - python-version: ["3.x"] - steps: - - uses: actions/checkout@v3 - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v3 - with: - python-version: ${{ matrix.python-version }} - cache: "pip" - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - cd pynetbox_query - pip install -r requirements.txt - - - - name: Run tests and collect coverage - run: cd pynetbox_query && python3 -m pytest diff --git a/.gitignore b/.gitignore index ac7b7a49..c6bbde6c 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ venv/ .coverage coverage.xml +pynetbox_query/pynetboxQuery.egg-info/ +pynetbox_query/build \ No newline at end of file diff --git a/pynetbox_query/.coveragerc b/pynetbox_query/.coveragerc index 5951051c..f4a58699 100644 --- a/pynetbox_query/.coveragerc +++ b/pynetbox_query/.coveragerc @@ -1,4 +1,4 @@ [report] exclude_lines = - if __name__ == .__main__.: \ No newline at end of file + if __name__ == .__main__.: diff --git a/pynetbox_query/.pylintrc b/pynetbox_query/.pylintrc index cbe36ea1..c056933b 100644 --- a/pynetbox_query/.pylintrc +++ b/pynetbox_query/.pylintrc @@ -5,11 +5,6 @@ max-line-length=120 # Disable various warnings: # C0114: Missing module string - we don't need module strings for the small repo +# W0212: Access to protected members - Don't know how to fix this :) -disable=C0114 - -# Ignore the venv folder. -# Despite venv not being pushed to the Repo. -# It's useful to have here for development purposes. - -ignore=venv \ No newline at end of file +disable=C0114, W0212 diff --git a/pynetbox_query/pynetbox_query/netbox_api/netbox_check.py b/pynetbox_query/pynetbox_query/netbox_api/netbox_check.py deleted file mode 100644 index d998ed68..00000000 --- a/pynetbox_query/pynetbox_query/netbox_api/netbox_check.py +++ /dev/null @@ -1,25 +0,0 @@ -class NetboxCheck: - """ - This class contains methods that check if an object exists in Netbox. - """ - - def __init__(self, api): - self.netbox = api - - def check_device_exists(self, device_name: str) -> bool: - """ - This method will check if a device exists in Netbox. - :param device_name: The name of the device. - :return: Returns bool. - """ - device = self.netbox.dcim.devices.get(name=device_name) - return bool(device) - - def check_device_type_exists(self, device_type: str) -> bool: - """ - This method will check if a device exists in Netbox. - :param device_type: The name of the device. - :return: Returns bool. - """ - device_type = self.netbox.dcim.device_types.get(slug=device_type) - return bool(device_type) diff --git a/pynetbox_query/pynetbox_query/netbox_api/netbox_connect.py b/pynetbox_query/pynetbox_query/netbox_api/netbox_connect.py deleted file mode 100644 index 234a92f9..00000000 --- a/pynetbox_query/pynetbox_query/netbox_api/netbox_connect.py +++ /dev/null @@ -1,25 +0,0 @@ -import pynetbox as nb - -# pylint:disable = too-few-public-methods - - -class NetboxConnect: - """ - This class is used to provide the Pynetbox Api object to other classes. - """ - - def __init__(self, url: str, token: str): - """ - This method initialises NetboxConnect with URL and token - :param url: Netbox website URL. - :param token: Netbox authentication token. - """ - self.url = url - self.token = token - - def api_object(self): - """ - This method returns the Pynetbox Api object. - :return: Returns the Api object - """ - return nb.api(self.url, self.token) diff --git a/pynetbox_query/pynetbox_query/netbox_api/netbox_create.py b/pynetbox_query/pynetbox_query/netbox_api/netbox_create.py deleted file mode 100644 index 78dd7313..00000000 --- a/pynetbox_query/pynetbox_query/netbox_api/netbox_create.py +++ /dev/null @@ -1,39 +0,0 @@ -from typing import Union, Dict, List - - -class NetboxCreate: - """ - This class contains methods that will interact create objects in Netbox. - """ - - def __init__(self, api): - """ - This initialises the method with the api object. - :param api: - """ - self.netbox = api - - def create_device(self, data: Union[Dict, List]) -> bool: - """ - This method uses the pynetbox Api to create a device in Netbox. - :param data: A list of or a single dictionary. - :return: Returns bool. - """ - devices = self.netbox.dcim.devices.create(data) - return bool(devices) - - def create_device_type( - self, model: str, slug: str, manufacturer: str, u_height: int = 0 - ) -> bool: - """ - This method creates a new device type in Netbox. - :param model: The model name of the device. - :param slug: The URL friendly version of the model name. - :param manufacturer: The manufacturer of the device. - :param u_height: This is the height of the device in the rack. Default 0. - :return: Returns bool. - """ - device_type = self.netbox.dcim.device_types.create( - model=model, slug=slug, manufacturer=manufacturer, u_height=u_height - ) - return bool(device_type) diff --git a/pynetbox_query/pynetbox_query/user_methods/csv_to_netbox.py b/pynetbox_query/pynetbox_query/user_methods/csv_to_netbox.py deleted file mode 100644 index 6bc2dd1f..00000000 --- a/pynetbox_query/pynetbox_query/user_methods/csv_to_netbox.py +++ /dev/null @@ -1,147 +0,0 @@ -from typing import List, Dict -from pathlib import Path -from dataclasses import asdict -import argparse -import sys -from pynetbox_query.netbox_api.netbox_create import NetboxCreate -from pynetbox_query.netbox_api.netbox_connect import NetboxConnect -from pynetbox_query.netbox_api.netbox_check import NetboxCheck -from pynetbox_query.utils.csv_to_dataclass import open_file, separate_data -from pynetbox_query.utils.query_device import QueryDevice -from pynetbox_query.utils.device_dataclass import Device -from pynetbox_query.utils.error_classes import DeviceFoundError, DeviceTypeNotFoundError - - -class CsvToNetbox: - """ - This class contains organised methods in the 4-step process of reading csv's to then uploading to Netbox. - """ - - def __init__(self, url: str, token: str): - """ - This initialises the class with the following parameters. - It also allows the rest of the class to access the imported Classes. - :param url: The Netbox url. - :param token: The Netbox auth token. - """ - self.netbox = NetboxConnect(url, token).api_object() - self.exist = NetboxCheck(self.netbox) - self.create = NetboxCreate(self.netbox) - self.query_dataclass = QueryDevice(self.netbox) - - @staticmethod - def check_file_path(file_path: str): - """ - This method checks if the filepath is valid. Raises a FileNotFoundError if it's invalid. - :param file_path: The path to the csv file. - """ - valid = Path(file_path).exists() - if not valid: - raise FileNotFoundError(f"Filepath: {file_path} is not valid.") - - @staticmethod - def read_csv(file_path: str) -> List[Device]: - """ - This method calls the csv_to_python and seperate_data method. - This will take the csv file and return a list of device dictionaries. - :param file_path: The file path to the csv file to be read. - :return: Returns a list of devices - """ - return separate_data(open_file(file_path)) - - def check_netbox_device(self, device_list: List[Device]) -> bool: - """ - This method calls the check_device_exists and check_device_type_exists method on each device in the list. - :param device_list: A list of devices. - :return: Returns True if the devices don't exist. Raises a DeviceFoundError otherwise. - """ - for device in device_list: - device_exist = self.exist.check_device_exists(device.name) - if device_exist: - raise DeviceFoundError(device) - return True - - def check_netbox_device_type(self, device_list: List[Device]) -> bool: - """ - This method calls the check_device_exists and check_device_type_exists method on each device in the list. - :param device_list: A list of devices. - :return: Returns True if the device types do exist. Raises a DeviceTypeNotFound otherwise. - """ - for device in device_list: - type_exist = self.exist.check_device_type_exists(device.device_type) - if not type_exist: - raise DeviceTypeNotFoundError(device) - return True - - def convert_data(self, device_list: List[Device]) -> List[Device]: - """ - This method calls the iterate_dict method. - :param device_list: A list of devices. - :return: Returns the updated list of devices. - """ - queried_list = self.query_dataclass.query_list(device_list) - return queried_list - - @staticmethod - def dataclass_to_dict(device_list: List[Device]) -> List[Dict]: - """ - This method converts the list of Devices into a list of dictionaries for Netbox Create. - :param device_list: A list of Device dataclasses. - :return: Returns the list of Devices as dictionaries. - """ - return [asdict(device) for device in device_list] - - def send_data(self, device_list: List[Device]) -> bool: - """ - This method calls the device create method to create devices in Netbox. - :param device_list: A list of devices. - :return: Returns bool whether the devices where created. - """ - dict_list = self.dataclass_to_dict(device_list) - devices = self.create.create_device(dict_list) - return bool(devices) - - -def arg_parser(): - """ - This function creates a parser object and adds 3 arguments to it. - This allows users to run the python file with arguments. Like a script. - """ - parser = argparse.ArgumentParser( - description="Create devices in Netbox from CSV files.", - usage="python csv_to_netbox.py url token file_path", - ) - parser.add_argument("url", help="The Netbox URL.") - parser.add_argument("token", help="Your Netbox Token.") - parser.add_argument("file_path", help="Your file path to csv files.") - return parser.parse_args() - - -def do_csv_to_netbox(args) -> bool: - """ - This function calls the methods from CsvToNetbox class. - :param args: The arguments from argparse. Supplied when the user runs the file from CLI. - :return: Returns bool if devices where created or not. - """ - class_object = CsvToNetbox(url=args.url, token=args.token) - class_object.check_file_path(args.file_path) - device_list = class_object.read_csv(args.file_path) - class_object.check_netbox_device(device_list) - class_object.check_netbox_device_type(device_list) - converted_device_list = class_object.convert_data(device_list) - result = class_object.send_data(converted_device_list) - return result - - -def main(): - """ - This function calls the necessary functions to call all other methods. - """ - sys.stdout.write("Starting...") - arguments = arg_parser() - do_csv_to_netbox(arguments) - sys.stdout.write("Finished.") - - -if __name__ == "__main__": - main() diff --git a/pynetbox_query/pynetbox_query/utils/csv_to_dataclass.py b/pynetbox_query/pynetbox_query/utils/csv_to_dataclass.py deleted file mode 100644 index 9754ecd0..00000000 --- a/pynetbox_query/pynetbox_query/utils/csv_to_dataclass.py +++ /dev/null @@ -1,27 +0,0 @@ -import csv -from typing import List, Dict -from pynetbox_query.utils.device_dataclass import Device - - -def open_file(file_path: str) -> List[Dict]: - """ - This function opens the specified csv file and returns the DictReader class on it. - :param file_path: The file path to the csv file. - :return: Returns an instance of the DictReader Class. - """ - with open(file_path, encoding="UTF-8") as file: - csv_reader_obj = csv.DictReader(file) - return list(csv_reader_obj) - - -def separate_data(csv_dicts: List[Dict]) -> List[Device]: - """ - This method separates the data from the iterator object into a list of dataclasses for each device. - :param csv_dicts: A list of dictionaries with each row of csv data. - :return: Returns a list of dataclass objects. - """ - devices = [] - for dictionary in csv_dicts: - device = Device(**dictionary) - devices.append(device) - return devices diff --git a/pynetbox_query/pynetbox_query/utils/error_classes.py b/pynetbox_query/pynetbox_query/utils/error_classes.py deleted file mode 100644 index e0722f59..00000000 --- a/pynetbox_query/pynetbox_query/utils/error_classes.py +++ /dev/null @@ -1,24 +0,0 @@ -from pynetbox_query.utils.device_dataclass import Device - - -class DeviceFoundError(Exception): - """ - This class is a custom exception for when a device is found to exist in Netbox. - """ - - def __init__(self, device: Device): - self.message = f"The device, {device.name}, already exists in Netbox." - super().__init__(self.message) - - -class DeviceTypeNotFoundError(Exception): - """ - This class is a custom expectation for when a device type is found to not exist in Netbox. - """ - - def __init__(self, device: Device): - self.message = ( - f"The device type, {device.device_type}, does not exist in Netbox.\n" - f"This device type needs to be created in Netbox before the device can be created." - ) - super().__init__(self.message) diff --git a/pynetbox_query/pynetbox_query/__init__.py b/pynetbox_query/pynetboxquery/__init__.py similarity index 100% rename from pynetbox_query/pynetbox_query/__init__.py rename to pynetbox_query/pynetboxquery/__init__.py diff --git a/pynetbox_query/pynetboxquery/__main__.py b/pynetbox_query/pynetboxquery/__main__.py new file mode 100644 index 00000000..8e660925 --- /dev/null +++ b/pynetbox_query/pynetboxquery/__main__.py @@ -0,0 +1,22 @@ +from importlib import import_module +import sys +from pynetboxquery.utils.error_classes import UserMethodNotFoundError + + +def main(): + """ + This function will run the correct user method for the action specified in the CLI. + """ + user_methods_names = ["upload_devices_to_netbox", "validate_data_fields_in_netbox"] + for user_method in user_methods_names: + user_method_module = import_module(f"pynetboxquery.user_methods.{user_method}") + user_method_class = getattr(user_method_module, "Main")() + aliases = user_method_class.aliases() + if sys.argv[1] in aliases: + user_method_module.Main().main() + return + raise UserMethodNotFoundError(f"The user method {sys.argv[1]} was not found.") + + +if __name__ == "__main__": + main() diff --git a/pynetbox_query/pynetbox_query/netbox_api/__init__.py b/pynetbox_query/pynetboxquery/netbox_api/__init__.py similarity index 100% rename from pynetbox_query/pynetbox_query/netbox_api/__init__.py rename to pynetbox_query/pynetboxquery/netbox_api/__init__.py diff --git a/pynetbox_query/pynetboxquery/netbox_api/netbox_connect.py b/pynetbox_query/pynetboxquery/netbox_api/netbox_connect.py new file mode 100644 index 00000000..3a24533d --- /dev/null +++ b/pynetbox_query/pynetboxquery/netbox_api/netbox_connect.py @@ -0,0 +1,11 @@ +from pynetbox import api + + +def api_object(url: str, token: str) -> api: + """ + This function returns the Pynetbox Api object used to interact with Netbox. + :param url: The Netbox URL. + :param token: User Api token. + :return: The Pynetbox api object. + """ + return api(url, token) diff --git a/pynetbox_query/pynetboxquery/netbox_api/netbox_create.py b/pynetbox_query/pynetboxquery/netbox_api/netbox_create.py new file mode 100644 index 00000000..ae3b2022 --- /dev/null +++ b/pynetbox_query/pynetboxquery/netbox_api/netbox_create.py @@ -0,0 +1,31 @@ +from typing import Union, Dict, List + + +class NetboxCreate: + """ + This class contains methods that will interact create objects in Netbox. + """ + + def __init__(self, api): + """ + This initialises the class with the api object to be used by methods. + """ + self.netbox = api + + def create_device(self, data: Union[Dict, List]) -> bool: + """ + This method uses the pynetbox Api to create a device in Netbox. + :param data: A list or a single dictionary containing data required to create devices in Netbox. + :return: Returns bool if the devices where made or not. + """ + devices = self.netbox.dcim.devices.create(data) + return bool(devices) + + def create_device_type(self, data: Union[Dict, List]) -> bool: + """ + This method creates a new device type in Netbox. + :param data: A list or single dictionary containing data required to create device types in Netbox. + :return: Returns bool if the device types where made or not. + """ + device_type = self.netbox.dcim.device_types.create(data) + return bool(device_type) diff --git a/pynetbox_query/pynetbox_query/netbox_api/netbox_get_id.py b/pynetbox_query/pynetboxquery/netbox_api/netbox_get_id.py similarity index 94% rename from pynetbox_query/pynetbox_query/netbox_api/netbox_get_id.py rename to pynetbox_query/pynetboxquery/netbox_api/netbox_get_id.py index 29889630..72e6d618 100644 --- a/pynetbox_query/pynetbox_query/netbox_api/netbox_get_id.py +++ b/pynetbox_query/pynetboxquery/netbox_api/netbox_get_id.py @@ -1,41 +1,41 @@ -from pynetbox_query.utils.device_dataclass import Device - - -# pylint: disable = R0903,C0115 -class NetboxGetId: - def __init__(self, api): - self.netbox = api - - def get_id(self, device: Device, attr: str) -> Device: - """ - This method queries Netbox for ID's of values. - :param device: The device dataclass. - :param attr: The attribute to query Netbox for. - :return: Returns an updated copy of the device dataclass. - """ - value = getattr(device, attr) - netbox_id = "" - if attr in ["status", "face", "airflow", "position", "name", "serial"]: - return getattr(device, attr) - match attr: - case "tenant": - netbox_id = self.netbox.tenancy.tenants.get(name=value).id - case "device_role": - netbox_id = self.netbox.dcim.device_roles.get(name=value).id - case "manufacturer": - netbox_id = self.netbox.dcim.manufacturers.get(name=value).id - case "device_type": - netbox_id = self.netbox.dcim.device_types.get(slug=value).id - case "site": - netbox_id = self.netbox.dcim.sites.get(name=value).id - case "location": - if isinstance(device.site, int): - site_slug = self.netbox.dcim.sites.get(id=device.site).slug - else: - site_slug = device.site.replace(" ", "-").lower() - netbox_id = self.netbox.dcim.locations.get( - name=value, site=site_slug - ).id - case "rack": - netbox_id = self.netbox.dcim.racks.get(name=value).id - return netbox_id +from pynetboxquery.utils.device_dataclass import Device + + +# pylint: disable = R0903,C0115 +class NetboxGetId: + def __init__(self, api): + self.netbox = api + + def get_id(self, device: Device, attr: str) -> Device: + """ + This method queries Netbox for ID's of values. + :param device: The device dataclass. + :param attr: The attribute to query Netbox for. + :return: Returns an updated copy of the device dataclass. + """ + value = getattr(device, attr) + netbox_id = "" + if attr in ["status", "face", "airflow", "position", "name", "serial"]: + return getattr(device, attr) + match attr: + case "tenant": + netbox_id = self.netbox.tenancy.tenants.get(name=value).id + case "device_role": + netbox_id = self.netbox.dcim.device_roles.get(name=value).id + case "manufacturer": + netbox_id = self.netbox.dcim.manufacturers.get(name=value).id + case "device_type": + netbox_id = self.netbox.dcim.device_types.get(slug=value).id + case "site": + netbox_id = self.netbox.dcim.sites.get(name=value).id + case "location": + if isinstance(device.site, int): + site_slug = self.netbox.dcim.sites.get(id=device.site).slug + else: + site_slug = device.site.replace(" ", "-").lower() + netbox_id = self.netbox.dcim.locations.get( + name=value, site=site_slug + ).id + case "rack": + netbox_id = self.netbox.dcim.racks.get(name=value).id + return netbox_id diff --git a/pynetbox_query/pynetboxquery/netbox_api/validate_data.py b/pynetbox_query/pynetboxquery/netbox_api/validate_data.py new file mode 100644 index 00000000..e6846f73 --- /dev/null +++ b/pynetbox_query/pynetboxquery/netbox_api/validate_data.py @@ -0,0 +1,99 @@ +from typing import List +from pynetbox import api +from pynetboxquery.utils.device_dataclass import Device + + +# Disabling this Pylint warning as it is unnecessary. +# pylint: disable = R0903 +class ValidateData: + """ + This class contains methods that check if things exist in Netbox or not. + """ + + def validate_data( + self, device_list: List[Device], netbox_api: api, fields: List[str] + ): + """ + This method will take a list of dataclasses. + Validate any data specified by the key word arguments. + :param device_list: A list of Device dataclasses containing the data. + :param netbox_api: The Api Object for Netbox. + :param fields: The Device fields to check. + """ + for field in fields: + results = self._call_validation_methods(device_list, netbox_api, field) + for result in results: + print(f"{result}\n") + + def _call_validation_methods( + self, device_list: List[Device], netbox_api: api, field: str + ) -> List[str]: + """ + This method will validate the field data by calling the correct validate method. + :param device_list: List of devices to validate. + :param netbox_api: The Api Object for Netbox. + :param field: Field to validate. + :return: Returns the results of the validation call. + """ + match field: + case "name": + device_names = [device.name for device in device_list] + results = self._check_list_device_name_in_netbox( + device_names, netbox_api + ) + case "device_type": + device_types = [device.device_type for device in device_list] + results = self._check_list_device_type_in_netbox( + device_types, netbox_api + ) + case _: + results = [f"Could not find a field for the argument: {field}."] + return results + + @staticmethod + def _check_device_name_in_netbox(device_name: str, netbox_api: api) -> bool: + """ + This method will check if a device exists in Netbox. + :param device_name: The name of the device. + :return: Returns bool. + """ + device = netbox_api.dcim.devices.get(name=device_name) + return bool(device) + + def _check_list_device_name_in_netbox( + self, device_names: List[str], netbox_api: api + ) -> List[str]: + """ + This method will call the validate method on each device name in the list and return the results. + :param device_names: List of device names to check. + :return: Results of the check. + """ + results = [] + for name in device_names: + in_netbox = self._check_device_name_in_netbox(name, netbox_api) + results += [f"Device {name} exists in Netbox: {in_netbox}."] + return results + + @staticmethod + def _check_device_type_in_netbox(device_type: str, netbox_api: api) -> bool: + """ + This method will check if a device type exists in Netbox. + :param device_type: The device type. + :return: Returns bool. + """ + device_type = netbox_api.dcim.device_types.get(slug=device_type) + return bool(device_type) + + def _check_list_device_type_in_netbox( + self, device_type_list: List[str], netbox_api: api + ) -> List[str]: + """ + This method will call the validate method on each device type in the list and return the results. + :param device_type_list: List of device types to check. + :return: Results of the check. + """ + results = [] + for device_type in device_type_list: + in_netbox = self._check_device_type_in_netbox(device_type, netbox_api) + results += [f"Device type {device_type} exists in Netbox: {in_netbox}."] + return results diff --git a/pynetbox_query/pynetbox_query/user_methods/__init__.py b/pynetbox_query/pynetboxquery/user_methods/__init__.py similarity index 100% rename from pynetbox_query/pynetbox_query/user_methods/__init__.py rename to pynetbox_query/pynetboxquery/user_methods/__init__.py diff --git a/pynetbox_query/pynetboxquery/user_methods/abstract_user_method.py b/pynetbox_query/pynetboxquery/user_methods/abstract_user_method.py new file mode 100644 index 00000000..de042d31 --- /dev/null +++ b/pynetbox_query/pynetboxquery/user_methods/abstract_user_method.py @@ -0,0 +1,45 @@ +from abc import ABC, abstractmethod +from typing import List, Dict +from argparse import ArgumentParser + + +class AbstractUserMethod(ABC): + """ + This Abstract class provides a template for user methods. + """ + + def _collect_kwargs(self) -> Dict: + """ + This method collects the arguments from the subparser into a dictionary of kwargs. + :return: Dictionary of kwargs. + """ + main_parser = self._subparser() + args = main_parser.parse_args() + kwargs = vars(args) + return kwargs + + @abstractmethod + def _subparser(self) -> ArgumentParser: + """ + This method creates a subparser with specific arguments to the user method. + :return: Returns the main parser that should contain the subparser information. + """ + + @staticmethod + @abstractmethod + def aliases() -> List[str]: + """Returns the aliases viable for this user_method.""" + + def main(self): + """ + This method gets the arguments and calls the run method with them. + """ + kwargs = self._collect_kwargs() + self._run(**kwargs) + + @staticmethod + @abstractmethod + def _run(url: str, token: str, file_path: str, **kwargs): + """ + This the main method in the user script. It contains all calls needed to perform the action. + """ diff --git a/pynetbox_query/pynetboxquery/user_methods/upload_devices_to_netbox.py b/pynetbox_query/pynetboxquery/user_methods/upload_devices_to_netbox.py new file mode 100644 index 00000000..980dee92 --- /dev/null +++ b/pynetbox_query/pynetboxquery/user_methods/upload_devices_to_netbox.py @@ -0,0 +1,59 @@ +from dataclasses import asdict +from pynetboxquery.utils.read_utils.read_file import ReadFile +from pynetboxquery.netbox_api.validate_data import ValidateData +from pynetboxquery.utils.query_device import QueryDevice +from pynetboxquery.netbox_api.netbox_create import NetboxCreate +from pynetboxquery.netbox_api.netbox_connect import api_object +from pynetboxquery.utils.parsers import Parsers +from pynetboxquery.user_methods.abstract_user_method import AbstractUserMethod + + +class Main(AbstractUserMethod): + """ + This class contains the run method to run the user script. + """ + + @staticmethod + def _run(url: str, token: str, file_path: str, **kwargs): + """ + This function does the following: + Create a Pynetbox Api Object with the users credentials. + Reads the file data into a list of Device dataclasses. + Validates the Device names and device types against Netbox. + Converts the data in the Devices to their Netbox ID's. + Changes those Device dataclasses into dictionaries. + Creates the Devices in Netbox. + :param url: The Netbox URL. + :param token: The user's Netbox api token. + :param file_path: The file path. + """ + api = api_object(url, token) + device_list = ReadFile().read_file(file_path, **kwargs) + ValidateData().validate_data( + device_list, api, **{"fields": ["name", "device_type"]} + ) + queried_devices = QueryDevice(api).query_list(device_list) + dictionary_devices = [asdict(device) for device in queried_devices] + NetboxCreate(api).create_device(dictionary_devices) + print("Devices added to Netbox.\n") + + def _subparser(self): + """ + This function creates the subparser for this user script inheriting the parent parser arguments. + """ + parent_parser, main_parser, subparsers = Parsers().arg_parser() + subparsers.add_parser( + "create_devices", + description="Create devices in Netbox from a file.", + usage="pynetboxquery create_devices ", + parents=[parent_parser], + aliases=self.aliases(), + ) + return main_parser + + @staticmethod + def aliases(): + """ + This function returns a list of aliases the script should be callable by. + """ + return ["create", "create_devices"] diff --git a/pynetbox_query/pynetboxquery/user_methods/validate_data_fields_in_netbox.py b/pynetbox_query/pynetboxquery/user_methods/validate_data_fields_in_netbox.py new file mode 100644 index 00000000..87fce256 --- /dev/null +++ b/pynetbox_query/pynetboxquery/user_methods/validate_data_fields_in_netbox.py @@ -0,0 +1,51 @@ +from pynetboxquery.utils.parsers import Parsers +from pynetboxquery.netbox_api.validate_data import ValidateData +from pynetboxquery.utils.read_utils.read_file import ReadFile +from pynetboxquery.netbox_api.netbox_connect import api_object +from pynetboxquery.user_methods.abstract_user_method import AbstractUserMethod + + +class Main(AbstractUserMethod): + """ + This class contains the run method to run the user script. + """ + + @staticmethod + def _run(url: str, token: str, file_path: str, **kwargs): + """ + This function does the following: + Reads the file from the file path. + Creates a Pynetbox Api Object with the users credentials. + Validates all the fields specified by the user + :param url: The Netbox URL. + :param token: The users Netbox api token. + :param file_path: The file_path. + """ + device_list = ReadFile().read_file(file_path, **kwargs) + api = api_object(url, token) + ValidateData().validate_data(device_list, api, kwargs["fields"]) + + def _subparser(self): + """ + This function creates the subparser for this user script inheriting the parent parser arguments. + It also adds an argument to collect the fields specified in the command line. + """ + parent_parser, main_parser, subparsers = Parsers().arg_parser() + parser_validate_data_fields_in_netbox = subparsers.add_parser( + "validate_data_fields_in_netbox", + description="Check data fields values in Netbox from a file.", + usage="pynetboxquery validate_data_fields_in_netbox ", + parents=[parent_parser], + aliases=self.aliases(), + ) + parser_validate_data_fields_in_netbox.add_argument( + "fields", help="The fields to check in Netbox.", nargs="*" + ) + return main_parser + + @staticmethod + def aliases(): + """ + This function returns a list of aliases the script should be callable by. + """ + return ["validate", "validate_data_fields_in_netbox"] diff --git a/pynetbox_query/pynetbox_query/utils/__init__.py b/pynetbox_query/pynetboxquery/utils/__init__.py similarity index 100% rename from pynetbox_query/pynetbox_query/utils/__init__.py rename to pynetbox_query/pynetboxquery/utils/__init__.py diff --git a/pynetbox_query/pynetbox_query/utils/device_dataclass.py b/pynetbox_query/pynetboxquery/utils/device_dataclass.py similarity index 100% rename from pynetbox_query/pynetbox_query/utils/device_dataclass.py rename to pynetbox_query/pynetboxquery/utils/device_dataclass.py diff --git a/pynetbox_query/pynetboxquery/utils/error_classes.py b/pynetbox_query/pynetboxquery/utils/error_classes.py new file mode 100644 index 00000000..4dcca85a --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/error_classes.py @@ -0,0 +1,32 @@ +# Disabling this Pylint error as these classes do not need Docstring. +# They are just errors +# pylint: disable = C0115 +"""Custom exceptions for the package.""" + + +class DeviceFoundError(Exception): + pass + + +class DeviceTypeNotFoundError(Exception): + pass + + +class FileTypeNotSupportedError(Exception): + pass + + +class DelimiterNotSpecifiedError(Exception): + pass + + +class SheetNameNotSpecifiedError(Exception): + pass + + +class ApiObjectNotParsedError(Exception): + pass + + +class UserMethodNotFoundError(Exception): + pass diff --git a/pynetbox_query/pynetboxquery/utils/parsers.py b/pynetbox_query/pynetboxquery/utils/parsers.py new file mode 100644 index 00000000..c24a8afa --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/parsers.py @@ -0,0 +1,40 @@ +import argparse + + +# Disabling this Pylint warning as it's irrelevant. +# pylint: disable = R0903 +class Parsers: + """ + This class contains the argparse methods for different commands. + """ + + def arg_parser(self): + """ + This function creates a parser object and adds 3 arguments to it. + This allows users to run the python file with arguments. Like a script. + """ + + parent_parser = self._parent_parser() + main_parser = argparse.ArgumentParser( + description="The main command. This cannot be run standalone and requires a subcommand to be provided.", + usage="pynetboxquery [command] [filepath] [url] [token] [kwargs]", + ) + subparsers = main_parser.add_subparsers( + dest="subparsers", + ) + + return parent_parser, main_parser, subparsers + + @staticmethod + def _parent_parser(): + parent_parser = argparse.ArgumentParser(add_help=False) + parent_parser.add_argument("url", help="The Netbox URL.") + parent_parser.add_argument("token", help="Your Netbox Token.") + parent_parser.add_argument("file_path", help="Your file path to csv files.") + parent_parser.add_argument( + "--delimiter", help="The separator in the text file." + ) + parent_parser.add_argument( + "--sheet-name", help="The sheet in the Excel Workbook to read from." + ) + return parent_parser diff --git a/pynetbox_query/pynetbox_query/utils/query_device.py b/pynetbox_query/pynetboxquery/utils/query_device.py similarity index 82% rename from pynetbox_query/pynetbox_query/utils/query_device.py rename to pynetbox_query/pynetboxquery/utils/query_device.py index a1dabdea..267c0b2c 100644 --- a/pynetbox_query/pynetbox_query/utils/query_device.py +++ b/pynetbox_query/pynetboxquery/utils/query_device.py @@ -1,7 +1,6 @@ from typing import List -from dataclasses import replace -from pynetbox_query.utils.device_dataclass import Device -from pynetbox_query.netbox_api.netbox_get_id import NetboxGetId +from pynetboxquery.utils.device_dataclass import Device +from pynetboxquery.netbox_api.netbox_get_id import NetboxGetId class QueryDevice: @@ -33,5 +32,4 @@ def query_device(self, device: Device) -> Device: changes = {} for attr in device.return_attrs(): changes[attr] = NetboxGetId(self.netbox).get_id(device, attr) - new_device = replace(device, **changes) - return new_device + return Device(**changes) diff --git a/pynetbox_query/pynetboxquery/utils/read_utils/__init__.py b/pynetbox_query/pynetboxquery/utils/read_utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/pynetbox_query/pynetboxquery/utils/read_utils/read_abc.py b/pynetbox_query/pynetboxquery/utils/read_utils/read_abc.py new file mode 100644 index 00000000..f7714bf9 --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/read_utils/read_abc.py @@ -0,0 +1,56 @@ +from abc import ABC, abstractmethod +from pathlib import Path +from typing import List, Dict +from pynetboxquery.utils.device_dataclass import Device + + +# Disabling this pylint warning as it is not necessary. +# pylint: disable = R0903 +class ReadAbstractBase(ABC): + """ + This Abstract Base Class ensures any Read methods made contain at least all of this code. + """ + + def __init__(self, file_path, **kwargs): + """ + Assigns the attribute file path. Checks the fie path is valid. Validates any kwargs needed. + """ + self.file_path = file_path + self._check_file_path(self.file_path) + self._validate(kwargs) + + @abstractmethod + def read(self) -> List[Dict]: + """ + This method reads the contents of a file into a list od Device Dataclasses. + """ + + @staticmethod + @abstractmethod + def _validate(kwargs): + """ + This method checks if a certain argument is given in kwargs and if not raise an error. + """ + + @staticmethod + def _check_file_path(file_path: str): + """ + This method checks if the file path exists in the user's system. + A FileNotFoundError will be raised if the file path is invalid. + If the file path is valid nothing will happen. + We don't need to declare that the file path is valid. + :param file_path: The file path to the file + """ + file_path_valid = Path(file_path).exists() + if not file_path_valid: + raise FileNotFoundError + + @staticmethod + def _dict_to_dataclass(dictionary_list: List[Dict]) -> List[Device]: + """ + This method takes a list of dictionaries and converts them all into Device dataclasses + then returns the list. + :param dictionary_list: A list of dictionaries. + :return: A list of Device dataclasses. + """ + return [Device(**dictionary) for dictionary in dictionary_list] diff --git a/pynetbox_query/pynetboxquery/utils/read_utils/read_csv.py b/pynetbox_query/pynetboxquery/utils/read_utils/read_csv.py new file mode 100644 index 00000000..2ab52105 --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/read_utils/read_csv.py @@ -0,0 +1,28 @@ +from csv import DictReader +from typing import List +from pynetboxquery.utils.device_dataclass import Device +from pynetboxquery.utils.read_utils.read_abc import ReadAbstractBase + + +# Disabling this pylint warning as it is not necessary. +# pylint: disable = R0903 +class ReadCSV(ReadAbstractBase): + """ + This class contains methods to read data from CSV files into a list of Device dataclasses. + """ + + def read(self) -> List[Device]: + """ + This method reads the contents of the csv file then returns a list of dictionaries + where each dictionary is a row of data with the keys being the column headers. + :return: A list of dictionaries. + """ + with open(self.file_path, mode="r", encoding="UTF-8") as file: + csv_reader = DictReader(file) + dictionary_list = list(csv_reader) + device_list = self._dict_to_dataclass(dictionary_list) + return device_list + + @staticmethod + def _validate(kwargs): + pass diff --git a/pynetbox_query/pynetboxquery/utils/read_utils/read_file.py b/pynetbox_query/pynetboxquery/utils/read_utils/read_file.py new file mode 100644 index 00000000..73c338f2 --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/read_utils/read_file.py @@ -0,0 +1,36 @@ +from typing import List +from pynetboxquery.utils.device_dataclass import Device +from pynetboxquery.utils.error_classes import FileTypeNotSupportedError +from pynetboxquery.utils.read_utils.read_csv import ReadCSV +from pynetboxquery.utils.read_utils.read_txt import ReadTXT +from pynetboxquery.utils.read_utils.read_xlsx import ReadXLSX + + +# Disabling this pylint warning as it is not necessary. +# pylint: disable = R0903 +class ReadFile: + """ + This class + """ + + @staticmethod + def read_file(file_path, **kwargs) -> List[Device]: + """ + + :param file_path: + :param kwargs: + :return: + """ + file_type = file_path.split(".")[-1] + match file_type: + case "csv": + device_list = ReadCSV(file_path).read() + case "txt": + device_list = ReadTXT(file_path, **kwargs).read() + case "xlsx": + device_list = ReadXLSX(file_path, **kwargs).read() + case _: + raise FileTypeNotSupportedError( + f"The file type '.{file_type}' is not supported by the method." + ) + return device_list diff --git a/pynetbox_query/pynetboxquery/utils/read_utils/read_txt.py b/pynetbox_query/pynetboxquery/utils/read_utils/read_txt.py new file mode 100644 index 00000000..632ddb8a --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/read_utils/read_txt.py @@ -0,0 +1,40 @@ +from csv import DictReader +from typing import List +from pynetboxquery.utils.device_dataclass import Device +from pynetboxquery.utils.error_classes import DelimiterNotSpecifiedError +from pynetboxquery.utils.read_utils.read_abc import ReadAbstractBase + + +# Disabling this pylint warning as it is not necessary. +# pylint: disable = R0903 +class ReadTXT(ReadAbstractBase): + """ + This class contains methods to read data from TXT files into a list of Device dataclasses. + """ + + def __init__(self, file_path, **kwargs): + super().__init__(file_path, **kwargs) + self.delimiter = kwargs["delimiter"].replace("\\t", "\t") + + @staticmethod + def _validate(kwargs): + """ + This method checks that the delimiter kwarg has been parsed. + If not raise an error. + """ + if "delimiter" not in kwargs: + raise DelimiterNotSpecifiedError( + "You must specify the delimiter in the text file." + ) + + def read(self) -> List[Device]: + """ + This method reads the contents of the text file then returns a list of dictionaries + where each dictionary is a row of data with the keys being the column headers. + :return: A list of dictionaries. + """ + with open(self.file_path, mode="r", encoding="UTF-8") as file: + file_reader = DictReader(file, delimiter=self.delimiter) + dictionary_list = list(file_reader) + device_list = self._dict_to_dataclass(dictionary_list) + return device_list diff --git a/pynetbox_query/pynetboxquery/utils/read_utils/read_xlsx.py b/pynetbox_query/pynetboxquery/utils/read_utils/read_xlsx.py new file mode 100644 index 00000000..027ba21a --- /dev/null +++ b/pynetbox_query/pynetboxquery/utils/read_utils/read_xlsx.py @@ -0,0 +1,39 @@ +from typing import List +from pandas import read_excel +from pynetboxquery.utils.device_dataclass import Device +from pynetboxquery.utils.error_classes import SheetNameNotSpecifiedError +from pynetboxquery.utils.read_utils.read_abc import ReadAbstractBase + + +# Disabling this pylint warning as it is not necessary. +# pylint: disable = R0903 +class ReadXLSX(ReadAbstractBase): + """ + This class contains methods to read data from XLSX files into a list of Device dataclasses. + """ + + def __init__(self, file_path, **kwargs): + super().__init__(file_path, **kwargs) + self.sheet_name = kwargs["sheet_name"] + + @staticmethod + def _validate(kwargs): + """ + This method checks that the delimiter kwarg has been parsed. + If not raise an error. + """ + if "sheet_name" not in kwargs: + raise SheetNameNotSpecifiedError( + "You must specify the sheet name in the excel workbook." + ) + + def read(self) -> List[Device]: + """ + This method reads the contents of a Sheet in an Excel Workbook then returns a list of dictionaries + where each dictionary is a row of data with the keys being the column headers. + :return: A list of dictionaries. + """ + dataframe = read_excel(self.file_path, sheet_name=self.sheet_name) + dictionary_list = dataframe.to_dict(orient="records") + device_list = self._dict_to_dataclass(dictionary_list) + return device_list diff --git a/pynetbox_query/setup.py b/pynetbox_query/setup.py index 56bd90f6..aed72ee5 100644 --- a/pynetbox_query/setup.py +++ b/pynetbox_query/setup.py @@ -6,7 +6,7 @@ LONG_DESCRIPTION = """Python package to query Netbox.""" setup( - name="pynetbox_query", + name="pynetboxQuery", version=VERSION, author="Kalibh Halford", author_email="", diff --git a/pynetbox_query/tests/conftest.py b/pynetbox_query/tests/conftest.py index 23423798..3b6b93f3 100644 --- a/pynetbox_query/tests/conftest.py +++ b/pynetbox_query/tests/conftest.py @@ -1,6 +1,6 @@ from typing import Dict from pytest import fixture -from pynetbox_query.utils.device_dataclass import Device +from pynetboxquery.utils.device_dataclass import Device @fixture(scope="function", name="dict_to_device") @@ -59,3 +59,11 @@ def mock_device_2_fixture(dict_to_device): "serial": "se2", } return dict_to_device(device) + + +@fixture(scope="function", name="mock_device_list") +def mock_device_list_fixture(mock_device, mock_device_2): + """ + This fixture returns a list of mock device types. + """ + return [mock_device, mock_device_2] diff --git a/pynetbox_query/tests/test__main__.py b/pynetbox_query/tests/test__main__.py new file mode 100644 index 00000000..8ae0fef8 --- /dev/null +++ b/pynetbox_query/tests/test__main__.py @@ -0,0 +1,37 @@ +from unittest.mock import patch +from pytest import raises +from pynetboxquery.__main__ import main +from pynetboxquery.utils.error_classes import UserMethodNotFoundError + + +@patch("pynetboxquery.__main__.import_module") +@patch("pynetboxquery.__main__.getattr") +@patch("pynetboxquery.__main__.sys") +def test_main(mock_sys, mock_getattr, mock_import_module): + """ + This test ensures the main method for the correct user script is called for the mocked argument. + """ + mock_sys.argv.__getitem__.return_value = "upload_devices_to_netbox" + mock_getattr.return_value.return_value.aliases.return_value = [ + "upload_devices_to_netbox" + ] + main() + mock_import_module.assert_called_with( + "pynetboxquery.user_methods.upload_devices_to_netbox" + ) + mock_getattr.assert_called_with(mock_import_module.return_value, "Main") + mock_import_module.return_value.Main.return_value.main.assert_called_once_with() + + +def test_main_fail(): + """ + This test ensures the main function errors when the incorrect argument is given. + """ + with patch("pynetboxquery.__main__.sys"): + with patch("pynetboxquery.__main__.import_module"): + with patch("pynetboxquery.__main__.getattr") as mock_getattr: + mock_getattr.return_value.return_value.aliases.return_value = [ + "not_real_method" + ] + with raises(UserMethodNotFoundError): + main() diff --git a/pynetbox_query/tests/test_csv_to_dataclass.py b/pynetbox_query/tests/test_csv_to_dataclass.py deleted file mode 100644 index 2b4fe289..00000000 --- a/pynetbox_query/tests/test_csv_to_dataclass.py +++ /dev/null @@ -1,35 +0,0 @@ -from csv import DictReader -from unittest.mock import patch, mock_open -from pynetbox_query.utils.csv_to_dataclass import separate_data, open_file - - -def test_separate_data(mock_device, mock_device_2): - """ - This test checks that when csv data is inputted the dataclass devices are created properly. - """ - mock_data = [ - "tenant,device_role,manufacturer,device_type,status," - "site,location,rack,face,airflow,position,name,serial", - "t1,dr1,m1,dt1,st1,si1,l1,r1,f1,a1,p1,n1,se1", - "t2,dr2,m2,dt2,st2,si2,l2,r2,f2,a2,p2,n2,se2", - ] - mock_reader_obj = DictReader(mock_data) - res = separate_data(mock_reader_obj) - assert res[0] == mock_device - assert res[1] == mock_device_2 - - -def test_open_file(): - """ - This test ensures the csv file is opened appropriately and the DictReader method is called. - """ - mock_data = ( - "tenant,device_role,manufacturer,device_type,status,site,location,rack,face,airflow,position,name," - "serial\nt1,dr1,m1,dt1,st1,si1,l1,r1,f1,a1,p1,n1,se1\nt2,dr2,m2,dt2,st2,si2,l2,r2,f2,a2,p2,n2,se2" - ) - - with patch("builtins.open", mock_open(read_data=mock_data)) as mock_file: - res = open_file("mock_file_path") - mock_file.assert_called_once_with("mock_file_path", encoding="UTF-8") - expected = list(DictReader(mock_data.splitlines())) - assert res == expected diff --git a/pynetbox_query/tests/test_csv_to_netbox.py b/pynetbox_query/tests/test_csv_to_netbox.py deleted file mode 100644 index 9cb02275..00000000 --- a/pynetbox_query/tests/test_csv_to_netbox.py +++ /dev/null @@ -1,196 +0,0 @@ -from unittest.mock import patch -from dataclasses import asdict -from pytest import fixture, raises -from pynetbox_query.user_methods.csv_to_netbox import ( - CsvToNetbox, - arg_parser, - do_csv_to_netbox, - main, -) -from pynetbox_query.utils.error_classes import DeviceFoundError, DeviceTypeNotFoundError - - -@fixture(name="instance") -def instance_fixture(): - """ - This method calls the class being tested. - """ - mock_url = "mock_url" - mock_token = "mock_token" - return CsvToNetbox(mock_url, mock_token) - - -def test_check_file_path_valid(instance): - """ - This test checks that the method doesn't raise any errors for valid filepaths. - """ - mock_file_path = "." - instance.check_file_path(mock_file_path) - - -def test_check_file_path_invalid(instance): - """ - This test checks that the method raises an error when the filepath is invalid. - """ - mock_file_path = ">" - with raises(FileNotFoundError): - instance.check_file_path(mock_file_path) - - -@patch("pynetbox_query.user_methods.csv_to_netbox.open_file") -@patch("pynetbox_query.user_methods.csv_to_netbox.separate_data") -def test_read_csv(mock_separate_data, mock_open_file, instance): - """ - This test ensures the file open method is called and the separate data method is called. - """ - mock_file_path = "mock_file_path" - res = instance.read_csv(mock_file_path) - mock_open_file.assert_called_once_with(mock_file_path) - mock_separate_data.assert_called_once_with(mock_open_file.return_value) - assert res == mock_separate_data.return_value - - -@patch("pynetbox_query.user_methods.csv_to_netbox.NetboxCheck.check_device_exists") -def test_check_netbox_device_does_exist( - mock_check_device_exists, instance, mock_device -): - """ - This test ensures that an error is raised if a device does exist in Netbox. - """ - with raises(DeviceFoundError): - instance.check_netbox_device([mock_device]) - mock_check_device_exists.assert_called_once_with(mock_device.name) - - -@patch("pynetbox_query.user_methods.csv_to_netbox.NetboxCheck.check_device_exists") -def test_check_netbox_device_not_exist(mock_check_device_exists, instance, mock_device): - """ - This test ensures an error is not raised if a device does not exist in Netbox. - """ - mock_check_device_exists.return_value = None - instance.check_netbox_device([mock_device]) - mock_check_device_exists.assert_called_once_with(mock_device.name) - - -@patch("pynetbox_query.user_methods.csv_to_netbox.NetboxCheck.check_device_type_exists") -def test_check_netbox_device_type_does_exist( - mock_check_device_type_exists, instance, mock_device -): - """ - This test ensures an error is not raised if a device type does exist in Netbox. - """ - instance.check_netbox_device_type([mock_device]) - mock_check_device_type_exists.assert_called_once_with(mock_device.device_type) - - -@patch("pynetbox_query.user_methods.csv_to_netbox.NetboxCheck.check_device_type_exists") -def test_check_netbox_device_type_not_exist( - mock_check_device_type_exists, instance, mock_device -): - """ - This test ensures an error is raised if a device type doesn't exist in Netbox. - """ - mock_check_device_type_exists.return_value = None - with raises(DeviceTypeNotFoundError): - instance.check_netbox_device_type([mock_device]) - mock_check_device_type_exists.assert_called_once_with(mock_device.device_type) - - -@patch("pynetbox_query.user_methods.csv_to_netbox.QueryDevice.query_list") -def test_convert_data(mock_query_dataclass, instance): - """ - This test ensures the convert data method is called with the correct arguments. - """ - device_list = ["", ""] - res = instance.convert_data(device_list) - mock_query_dataclass.assert_any_call(device_list) - assert res == mock_query_dataclass.return_value - - -def test_dataclass_to_dict(instance, mock_device, mock_device_2): - """ - This test ensures that the Device dataclasses are returned as dictionaries when the method is called. - """ - mock_device_list = [mock_device, mock_device_2] - res = instance.dataclass_to_dict(mock_device_list) - expected = [asdict(device) for device in mock_device_list] - assert res == expected - - -@patch("pynetbox_query.user_methods.csv_to_netbox.NetboxCreate.create_device") -@patch("pynetbox_query.user_methods.csv_to_netbox.CsvToNetbox.dataclass_to_dict") -def test_send_data(mock_dataclass_to_dict, mock_create_device, instance): - """ - This test ensures the correct methods are called with the correct arguments. - """ - mock_device_list = ["", ""] - res = instance.send_data(mock_device_list) - mock_dataclass_to_dict.assert_called_once_with(mock_device_list) - mock_create_device.assert_called_once_with(mock_dataclass_to_dict.return_value) - assert res - - -@patch("pynetbox_query.user_methods.csv_to_netbox.argparse.ArgumentParser") -def test_arg_parser(mock_argparse): - """ - This test ensures the argparse method adds the correct arguments and returns them. - """ - res = arg_parser() - mock_argparse.assert_called_once_with( - description="Create devices in Netbox from CSV files.", - usage="python csv_to_netbox.py url token file_path", - ) - mock_argparse.return_value.add_argument.assert_any_call( - "file_path", help="Your file path to csv files." - ) - mock_argparse.return_value.add_argument.assert_any_call( - "token", help="Your Netbox Token." - ) - mock_argparse.return_value.add_argument.assert_any_call( - "url", help="The Netbox URL." - ) - mock_argparse.return_value.parse_args.assert_called() - assert res == mock_argparse.return_value.parse_args() - - -@patch("pynetbox_query.user_methods.csv_to_netbox.CsvToNetbox") -def test_do_csv_to_netbox(mock_csv_to_netbox_class): - """ - This test ensures all the correct methods are called with the correct arguments. - """ - - class Args: - # pylint: disable=R0903 - """ - This class mocks the argument class in argparse. - """ - - def __init__(self, file_path, url, token): - self.file_path = file_path - self.url = url - self.token = token - - args = Args("mock_file_path", "mock_url", "mock_token") - res = do_csv_to_netbox(args) - mock_csv_to_netbox_class.assert_any_call(url=args.url, token=args.token) - mock_class_object = mock_csv_to_netbox_class.return_value - mock_class_object.check_file_path.assert_any_call(args.file_path) - mock_class_object.read_csv.assert_any_call(args.file_path) - mock_device_list = mock_class_object.read_csv.return_value - mock_class_object.check_netbox_device.assert_any_call(mock_device_list) - mock_class_object.check_netbox_device_type.assert_any_call(mock_device_list) - mock_class_object.convert_data.assert_any_call(mock_device_list) - mock_converted_device_list = mock_class_object.convert_data.return_value - mock_class_object.send_data.assert_any_call(mock_converted_device_list) - assert res - - -@patch("pynetbox_query.user_methods.csv_to_netbox.arg_parser") -@patch("pynetbox_query.user_methods.csv_to_netbox.do_csv_to_netbox") -def test_main(mock_do_csv_to_netbox, mock_arg_parser): - """ - This test ensures that when main is called the argparse method and do method are called with arguments. - """ - main() - mock_arg_parser.assert_called_once() - mock_do_csv_to_netbox.assert_called_once_with(mock_arg_parser.return_value) diff --git a/pynetbox_query/tests/test_netbox_api/test_netbox_connect.py b/pynetbox_query/tests/test_netbox_api/test_netbox_connect.py new file mode 100644 index 00000000..c5ba3db6 --- /dev/null +++ b/pynetbox_query/tests/test_netbox_api/test_netbox_connect.py @@ -0,0 +1,14 @@ +from unittest.mock import patch +from pynetboxquery.netbox_api.netbox_connect import api_object + + +@patch("pynetboxquery.netbox_api.netbox_connect.api") +def test_api_object(mock_api): + """ + This test checks that the Api object is returned. + """ + mock_url = "url" + mock_token = "token" + res = api_object(mock_url, mock_token) + mock_api.assert_called_once_with(mock_url, mock_token) + assert res == mock_api.return_value diff --git a/pynetbox_query/tests/test_netbox_api/test_netbox_create.py b/pynetbox_query/tests/test_netbox_api/test_netbox_create.py new file mode 100644 index 00000000..5c79559c --- /dev/null +++ b/pynetbox_query/tests/test_netbox_api/test_netbox_create.py @@ -0,0 +1,60 @@ +from unittest.mock import NonCallableMock +from dataclasses import asdict +from pytest import fixture +from pynetboxquery.netbox_api.netbox_create import NetboxCreate + + +@fixture(name="instance") +def instance_fixture(): + """ + This fixture method calls the class being tested. + :return: The class object. + """ + mock_api = NonCallableMock() + return NetboxCreate(mock_api) + + +def test_create_device_one(instance, mock_device): + """ + This test ensures the .create method is called once with the correct arguments. + """ + mock_device_dict = asdict(mock_device) + res = instance.create_device(mock_device_dict) + instance.netbox.dcim.devices.create.assert_called_once_with(mock_device_dict) + assert res + + +def test_create_device_many(instance, mock_device, mock_device_2): + """ + This test ensures the .create method is called once with the correct arguments. + """ + mock_device_dict = asdict(mock_device) + mock_device_2_dict = asdict(mock_device_2) + res = instance.create_device([mock_device_dict, mock_device_2_dict]) + instance.netbox.dcim.devices.create.assert_called_once_with( + [mock_device_dict, mock_device_2_dict] + ) + assert res + + +def test_create_device_type_one(instance): + """ + This test ensures the .create method is called once with the correct arguments. + """ + mock_device_type = "" + res = instance.create_device_type(mock_device_type) + instance.netbox.dcim.device_types.create.assert_called_once_with(mock_device_type) + assert res + + +def test_create_device_type_many(instance): + """ + This test ensures the .create method is called once with the correct arguments. + """ + mock_device_type = "" + mock_device_type_2 = "" + res = instance.create_device_type([mock_device_type, mock_device_type_2]) + instance.netbox.dcim.device_types.create.assert_called_once_with( + [mock_device_type, mock_device_type_2] + ) + assert res diff --git a/pynetbox_query/tests/test_netbox_get_id.py b/pynetbox_query/tests/test_netbox_api/test_netbox_get_id.py similarity index 97% rename from pynetbox_query/tests/test_netbox_get_id.py rename to pynetbox_query/tests/test_netbox_api/test_netbox_get_id.py index c812328a..a640a9b0 100644 --- a/pynetbox_query/tests/test_netbox_get_id.py +++ b/pynetbox_query/tests/test_netbox_api/test_netbox_get_id.py @@ -1,6 +1,6 @@ from unittest.mock import patch from pytest import fixture -from pynetbox_query.netbox_api.netbox_get_id import NetboxGetId +from pynetboxquery.netbox_api.netbox_get_id import NetboxGetId @fixture(name="instance") diff --git a/pynetbox_query/tests/test_netbox_api/test_validate_data.py b/pynetbox_query/tests/test_netbox_api/test_validate_data.py new file mode 100644 index 00000000..36ada2be --- /dev/null +++ b/pynetbox_query/tests/test_netbox_api/test_validate_data.py @@ -0,0 +1,211 @@ +from unittest.mock import patch, NonCallableMock +from pytest import fixture +from pynetboxquery.netbox_api.validate_data import ValidateData + + +@fixture(name="instance") +def instance_fixture(): + """ + This fixture returns the class instance to be tested. + """ + return ValidateData() + + +@patch("pynetboxquery.netbox_api.validate_data.ValidateData._call_validation_methods") +def test_validate_data_one_field( + mock_call_validation_methods, instance, mock_device_list +): + """ + This test ensures that the correct methods are called when validating one field. + """ + mock_netbox_api = "" + mock_fields = ["field1"] + instance.validate_data(mock_device_list, mock_netbox_api, mock_fields) + mock_call_validation_methods.assert_called_once_with( + mock_device_list, mock_netbox_api, mock_fields[0] + ) + + +@patch("pynetboxquery.netbox_api.validate_data.ValidateData._call_validation_methods") +def test_validate_data_many_fields( + mock_call_validation_methods, instance, mock_device_list +): + """ + This test ensures that the correct methods are called when validating more than one field. + """ + mock_netbox_api = "" + mock_fields = ["field1", "field2"] + instance.validate_data(mock_device_list, mock_netbox_api, mock_fields) + mock_call_validation_methods.assert_any_call( + mock_device_list, mock_netbox_api, mock_fields[0] + ) + mock_call_validation_methods.assert_any_call( + mock_device_list, mock_netbox_api, mock_fields[1] + ) + + +@patch("pynetboxquery.netbox_api.validate_data.ValidateData._call_validation_methods") +@patch("pynetboxquery.netbox_api.validate_data.print") +def test_validate_data_one_result( + mock_print, mock_call_validation_methods, instance, mock_device_list +): + """ + This test ensures that the correct methods are called when printing one set of results. + """ + mock_netbox_api = "" + mock_fields = ["field1"] + mock_call_validation_methods.return_value = ["result1"] + instance.validate_data(mock_device_list, mock_netbox_api, mock_fields) + assert mock_print.call_count == 1 + + +@patch("pynetboxquery.netbox_api.validate_data.ValidateData._call_validation_methods") +@patch("pynetboxquery.netbox_api.validate_data.print") +def test_validate_data_many_results( + mock_print, mock_call_validation_methods, instance, mock_device_list +): + """ + This test ensures that the correct methods are called when printing more than one set of results. + """ + mock_netbox_api = "" + mock_fields = ["field1"] + mock_call_validation_methods.return_value = ["result1", "result2"] + instance.validate_data(mock_device_list, mock_netbox_api, mock_fields) + assert mock_print.call_count == 2 + + +@patch( + "pynetboxquery.netbox_api.validate_data.ValidateData._check_list_device_name_in_netbox" +) +def test_call_validation_methods_name( + mock_check_list_device_name_in_netbox, instance, mock_device_list +): + """ + This test ensures the correct methods are called when validating the name field. + """ + mock_netbox_api = "" + res = instance._call_validation_methods(mock_device_list, mock_netbox_api, "name") + mock_device_values = [device.name for device in mock_device_list] + mock_check_list_device_name_in_netbox.assert_called_once_with( + mock_device_values, mock_netbox_api + ) + assert res == mock_check_list_device_name_in_netbox.return_value + + +@patch( + "pynetboxquery.netbox_api.validate_data.ValidateData._check_list_device_type_in_netbox" +) +def test_call_validation_methods_type( + mock_check_list_device_type_in_netbox, instance, mock_device_list +): + """ + This test ensures the correct methods are called when validating the device_type field. + """ + mock_netbox_api = "" + res = instance._call_validation_methods( + mock_device_list, mock_netbox_api, "device_type" + ) + mock_device_values = [device.device_type for device in mock_device_list] + mock_check_list_device_type_in_netbox.assert_called_once_with( + mock_device_values, mock_netbox_api + ) + assert res == mock_check_list_device_type_in_netbox.return_value + + +def test_call_validation_methods_wildcard(instance, mock_device_list): + """ + This test ensures the correct methods are called when validating a field that doesn't exist. + """ + mock_netbox_api = "" + res = instance._call_validation_methods( + mock_device_list, mock_netbox_api, "wildcard" + ) + assert res == ["Could not find a field for the argument: wildcard."] + + +def test_check_device_name_in_netbox(instance): + """ + This test ensures the .get() method is called with the correct arguments when checking the device name. + """ + netbox_api = NonCallableMock() + res = instance._check_device_name_in_netbox("name", netbox_api) + netbox_api.dcim.devices.get.assert_called_once_with(name="name") + assert res + + +@patch( + "pynetboxquery.netbox_api.validate_data.ValidateData._check_device_name_in_netbox" +) +def test_check_list_device_name_in_netbox_one( + mock_check_device_name_in_netbox, instance +): + """ + This test ensures the correct methods are called for checking a list of device names that holds one value. + """ + res = instance._check_list_device_name_in_netbox(["name"], "api") + mock_check_device_name_in_netbox.assert_called_once_with("name", "api") + assert res == [ + f"Device name exists in Netbox: {mock_check_device_name_in_netbox.return_value}." + ] + + +@patch( + "pynetboxquery.netbox_api.validate_data.ValidateData._check_device_name_in_netbox" +) +def test_check_list_device_name_in_netbox_many( + mock_check_device_name_in_netbox, instance +): + """ + This test ensures the correct methods are called for checking a list of device names that holds many values. + """ + res = instance._check_list_device_name_in_netbox(["name1", "name2"], "api") + mock_check_device_name_in_netbox.assert_any_call("name1", "api") + mock_check_device_name_in_netbox.assert_any_call("name2", "api") + assert res == [ + f"Device name1 exists in Netbox: {mock_check_device_name_in_netbox.return_value}.", + f"Device name2 exists in Netbox: {mock_check_device_name_in_netbox.return_value}.", + ] + + +def test_check_device_type_in_netbox(instance): + """ + This test ensures the .get() method is called with the correct arguments when checking the device type. + """ + netbox_api = NonCallableMock() + res = instance._check_device_type_in_netbox("type", netbox_api) + netbox_api.dcim.device_types.get.assert_called_once_with(slug="type") + assert res + + +@patch( + "pynetboxquery.netbox_api.validate_data.ValidateData._check_device_type_in_netbox" +) +def test_check_list_device_type_in_netbox_one( + mock_check_device_type_in_netbox, instance +): + """ + This test ensures the correct methods are called for checking a list of device types that holds one value. + """ + res = instance._check_list_device_type_in_netbox(["type"], "api") + mock_check_device_type_in_netbox.assert_called_once_with("type", "api") + assert res == [ + f"Device type type exists in Netbox: {mock_check_device_type_in_netbox.return_value}." + ] + + +@patch( + "pynetboxquery.netbox_api.validate_data.ValidateData._check_device_type_in_netbox" +) +def test_check_list_device_type_in_netbox_many( + mock_check_device_type_in_netbox, instance +): + """ + This test ensures the correct methods are called for checking a list of device types that holds many values. + """ + res = instance._check_list_device_type_in_netbox(["type1", "type2"], "api") + mock_check_device_type_in_netbox.assert_any_call("type1", "api") + mock_check_device_type_in_netbox.assert_any_call("type2", "api") + assert res == [ + f"Device type type1 exists in Netbox: {mock_check_device_type_in_netbox.return_value}.", + f"Device type type2 exists in Netbox: {mock_check_device_type_in_netbox.return_value}.", + ] diff --git a/pynetbox_query/tests/test_netbox_check.py b/pynetbox_query/tests/test_netbox_check.py deleted file mode 100644 index 30176ab1..00000000 --- a/pynetbox_query/tests/test_netbox_check.py +++ /dev/null @@ -1,35 +0,0 @@ -from unittest.mock import MagicMock, NonCallableMock -import pytest -from pynetbox_query.netbox_api.netbox_check import NetboxCheck - - -@pytest.fixture(name="instance") -def instance_fixture(): - """ - This fixture method calls the class being tested. - :return: The class object. - """ - netbox = NonCallableMock() - return NetboxCheck(netbox) - - -def test_check_device_exists(instance): - """ - This test ensures the .get method is called once with the correct argument. - """ - mock_device = MagicMock() - device_name = NonCallableMock() - instance.netbox.dcim.devices = mock_device - instance.check_device_exists(device_name) - mock_device.get.assert_called_once_with(name=device_name) - - -def test_check_device_type_exists(instance): - """ - This test ensures the .get method is called once with the correct argument. - """ - mock_device_types = MagicMock() - device_type = NonCallableMock() - instance.netbox.dcim.device_types = mock_device_types - instance.check_device_type_exists(device_type) - mock_device_types.get.assert_called_once_with(slug=device_type) diff --git a/pynetbox_query/tests/test_netbox_connect.py b/pynetbox_query/tests/test_netbox_connect.py deleted file mode 100644 index 95b20ef3..00000000 --- a/pynetbox_query/tests/test_netbox_connect.py +++ /dev/null @@ -1,24 +0,0 @@ -from unittest.mock import NonCallableMock, patch -import pytest -from pynetbox_query.netbox_api.netbox_connect import NetboxConnect - - -@pytest.fixture(name="instance") -def instance_fixture(): - """ - This fixture method calls the class being tested. - :return: The class object. - """ - url = NonCallableMock() - token = NonCallableMock() - return NetboxConnect(url, token) - - -def test_api_object(instance): - """ - This test checks that the Api method is called once. - """ - with patch("pynetbox_query.netbox_api.netbox_connect.nb") as mock_netbox: - res = instance.api_object() - mock_netbox.api.assert_called_once_with(instance.url, instance.token) - assert res == mock_netbox.api.return_value diff --git a/pynetbox_query/tests/test_netbox_create.py b/pynetbox_query/tests/test_netbox_create.py deleted file mode 100644 index 7bce6b69..00000000 --- a/pynetbox_query/tests/test_netbox_create.py +++ /dev/null @@ -1,41 +0,0 @@ -from unittest.mock import MagicMock, NonCallableMock -import pytest -from pynetbox_query.netbox_api.netbox_create import NetboxCreate - - -@pytest.fixture(name="instance") -def instance_fixture(): - """ - This fixture method calls the class being tested. - :return: The class object. - """ - netbox = NonCallableMock() - return NetboxCreate(netbox) - - -def test_create_device(instance): - """ - This test ensures the .create method is called once with the correct argument. - """ - mock_device = MagicMock() - mock_data = NonCallableMock() - instance.netbox.dcim.devices = mock_device - instance.create_device(mock_data) - mock_device.create.assert_called_once_with(mock_data) - - -def test_create_device_type(instance): - """ - This test ensures the .creat method is called once with the correct arguments. - """ - mock_device_types = MagicMock() - mock_model = NonCallableMock() - mock_slug = NonCallableMock() - mock_manufacturer = NonCallableMock() - instance.netbox.dcim.device_types = mock_device_types - instance.create_device_type( - model=mock_model, manufacturer=mock_manufacturer, slug=mock_slug - ) - mock_device_types.create.assert_called_once_with( - model=mock_model, slug=mock_slug, manufacturer=mock_manufacturer, u_height=0 - ) diff --git a/pynetbox_query/tests/test_user_methods/test_abstract_user_method.py b/pynetbox_query/tests/test_user_methods/test_abstract_user_method.py new file mode 100644 index 00000000..4e70e7bc --- /dev/null +++ b/pynetbox_query/tests/test_user_methods/test_abstract_user_method.py @@ -0,0 +1,48 @@ +from argparse import ArgumentParser +from typing import List +from unittest.mock import patch +from pytest import fixture +from pynetboxquery.user_methods.abstract_user_method import AbstractUserMethod + + +@fixture(name="instance") +def instance_fixture(): + """ + This fixture returns the Stub class to be tested. + """ + return StubAbstractUserMethod() + + +class StubAbstractUserMethod(AbstractUserMethod): + """ + This class provides a Stub version of the AbstractUserMethod class. + So we do not have to patch all the abstract methods. + """ + + def _subparser(self) -> ArgumentParser: + """Placeholder Method.""" + + @staticmethod + def aliases() -> List[str]: + """Placeholder Method.""" + + @staticmethod + def _run(url: str, token: str, file_path: str, **kwargs): + """Placeholder Method.""" + + +@patch( + "pynetboxquery.user_methods.abstract_user_method.AbstractUserMethod._collect_kwargs" +) +def test_main(mock_collect_kwargs, instance): + """ + This test ensures the collect kwargs method is called. + We do not need to assert that ._run is called as it is an abstract method. + """ + mock_collect_kwargs.return_value = { + "url": "mock_url", + "token": "mock_token", + "file_path": "mock_file_path", + } + instance.main() + mock_collect_kwargs.assert_called_once() diff --git a/pynetbox_query/tests/test_user_methods/test_upload_devices_to_netbox.py b/pynetbox_query/tests/test_user_methods/test_upload_devices_to_netbox.py new file mode 100644 index 00000000..bffeed37 --- /dev/null +++ b/pynetbox_query/tests/test_user_methods/test_upload_devices_to_netbox.py @@ -0,0 +1,66 @@ +from dataclasses import asdict +from unittest.mock import patch, NonCallableMock +from pynetboxquery.user_methods.upload_devices_to_netbox import Main + + +def test_aliases(): + """ + This test ensures that the aliases function returns a list of aliases. + """ + res = Main().aliases() + assert res == ["create", "create_devices"] + + +# pylint: disable = R0801 +@patch("pynetboxquery.user_methods.upload_devices_to_netbox.Parsers") +def test_subparser(mock_parsers): + """ + This test ensures all the correct methods are called with the correct arguments to create a subparser. + """ + mock_subparsers = NonCallableMock() + mock_parsers.return_value.arg_parser.return_value = ( + "mock_parent_parser", + "mock_main_parser", + mock_subparsers, + ) + res = Main()._subparser() + mock_parsers.return_value.arg_parser.assert_called_once() + mock_subparsers.add_parser.assert_called_once_with( + "create_devices", + description="Create devices in Netbox from a file.", + usage="pynetboxquery create_devices ", + parents=["mock_parent_parser"], + aliases=["create", "create_devices"], + ) + assert res == "mock_main_parser" + + +@patch("pynetboxquery.user_methods.upload_devices_to_netbox.ReadFile") +@patch("pynetboxquery.user_methods.upload_devices_to_netbox.api_object") +@patch("pynetboxquery.user_methods.upload_devices_to_netbox.ValidateData") +@patch("pynetboxquery.user_methods.upload_devices_to_netbox.QueryDevice") +@patch("pynetboxquery.user_methods.upload_devices_to_netbox.NetboxCreate") +def test_upload_devices_to_netbox( + mock_netbox_create, mock_query_device, mock_validate_data, mock_api, mock_read_file +): + """ + This test ensures all the correct methods are called with the correct arguments + """ + Main()._run("mock_url", "mock_token", "mock_file_path") + mock_api_object = mock_api.return_value + mock_api.assert_called_once_with("mock_url", "mock_token") + mock_device_list = mock_read_file.return_value.read_file.return_value + mock_read_file.return_value.read_file.assert_called_once_with("mock_file_path") + mock_validate_data.return_value.validate_data.assert_called_once_with( + mock_device_list, mock_api_object, **{"fields": ["name", "device_type"]} + ) + mock_queried_devices = mock_query_device.return_value.query_list.return_value + mock_query_device.assert_called_once_with(mock_api_object) + mock_query_device.return_value.query_list.assert_called_once_with(mock_device_list) + mock_dictionary_devices = [ + asdict(mock_device) for mock_device in mock_queried_devices + ] + mock_netbox_create.assert_called_once_with(mock_api_object) + mock_netbox_create.return_value.create_device.assert_called_once_with( + mock_dictionary_devices + ) diff --git a/pynetbox_query/tests/test_user_methods/test_validate_data_fields_in_netbox.py b/pynetbox_query/tests/test_user_methods/test_validate_data_fields_in_netbox.py new file mode 100644 index 00000000..fbefb909 --- /dev/null +++ b/pynetbox_query/tests/test_user_methods/test_validate_data_fields_in_netbox.py @@ -0,0 +1,59 @@ +from unittest.mock import patch, NonCallableMock +from pynetboxquery.user_methods.validate_data_fields_in_netbox import Main + + +def test_aliases(): + """ + This test ensures that the aliases function returns a list of aliases. + """ + res = Main().aliases() + assert res == ["validate", "validate_data_fields_in_netbox"] + + +# pylint: disable = R0801 +@patch("pynetboxquery.user_methods.validate_data_fields_in_netbox.Parsers") +def test_subparser(mock_parsers): + """ + This test ensures all the correct methods are called with the correct arguments to create a subparser. + """ + mock_subparsers = NonCallableMock() + mock_parsers.return_value.arg_parser.return_value = ( + "mock_parent_parser", + "mock_main_parser", + mock_subparsers, + ) + res = Main()._subparser() + mock_parsers.return_value.arg_parser.assert_called_once() + mock_subparsers.add_parser.assert_called_once_with( + "validate_data_fields_in_netbox", + description="Check data fields values in Netbox from a file.", + usage="pynetboxquery validate_data_fields_in_netbox ", + parents=["mock_parent_parser"], + aliases=Main().aliases(), + ) + mock_subparsers.add_parser.return_value.add_argument.assert_called_once_with( + "fields", help="The fields to check in Netbox.", nargs="*" + ) + assert res == "mock_main_parser" + + +@patch("pynetboxquery.user_methods.validate_data_fields_in_netbox.ReadFile") +@patch("pynetboxquery.user_methods.validate_data_fields_in_netbox.api_object") +@patch("pynetboxquery.user_methods.validate_data_fields_in_netbox.ValidateData") +def test_validate_data_fields_in_netbox( + mock_validate_data, mock_api_object, mock_read_file +): + """ + This test ensures all the correct methods are called with the correct arguments + """ + mock_kwargs = {"fields": ["mock_val"]} + Main()._run("mock_url", "mock_token", "mock_file_path", **mock_kwargs) + mock_device_list = mock_read_file.return_value.read_file.return_value + mock_read_file.return_value.read_file.assert_called_once_with( + "mock_file_path", **mock_kwargs + ) + mock_api = mock_api_object.return_value + mock_api_object.assert_called_once_with("mock_url", "mock_token") + mock_validate_data.return_value.validate_data.assert_called_once_with( + mock_device_list, mock_api, mock_kwargs["fields"] + ) diff --git a/pynetbox_query/tests/test_device_dataclass.py b/pynetbox_query/tests/test_utils/test_device_dataclass.py similarity index 100% rename from pynetbox_query/tests/test_device_dataclass.py rename to pynetbox_query/tests/test_utils/test_device_dataclass.py diff --git a/pynetbox_query/tests/test_utils/test_parsers.py b/pynetbox_query/tests/test_utils/test_parsers.py new file mode 100644 index 00000000..048033a4 --- /dev/null +++ b/pynetbox_query/tests/test_utils/test_parsers.py @@ -0,0 +1,56 @@ +from unittest.mock import patch +from pytest import fixture +from pynetboxquery.utils.parsers import Parsers + + +@fixture(name="instance") +def instance_fixture(): + """ + This fixture returns the Parsers class to be used in tests. + """ + return Parsers() + + +@patch("pynetboxquery.utils.parsers.Parsers._parent_parser") +@patch("pynetboxquery.utils.parsers.argparse") +def test_arg_parser(mock_argparse, mock_parent_parser, instance): + """ + This test ensures the argparse method adds the correct arguments and returns them. + """ + res = instance.arg_parser() + mock_parent_parser.assert_called_once_with() + mock_argparse.ArgumentParser.assert_called_once_with( + description="The main command. This cannot be run standalone and requires a subcommand to be provided.", + usage="pynetboxquery [command] [filepath] [url] [token] [kwargs]", + ) + mock_argparse.ArgumentParser.return_value.add_subparsers.assert_called_once_with( + dest="subparsers", + ) + expected_parent = mock_parent_parser.return_value + expected_main = mock_argparse.ArgumentParser.return_value + expected_subparser = ( + mock_argparse.ArgumentParser.return_value.add_subparsers.return_value + ) + assert res == (expected_parent, expected_main, expected_subparser) + + +@patch("pynetboxquery.utils.parsers.argparse") +def test_parent_parser(mock_argparse, instance): + """ + This test ensures the parent parser is created with the correct arguments. + """ + res = instance._parent_parser() + mock_argparse.ArgumentParser.assert_called_once_with(add_help=False) + mock_parent_parser = mock_argparse.ArgumentParser.return_value + mock_parent_parser.add_argument.assert_any_call("url", help="The Netbox URL.") + mock_parent_parser.add_argument.assert_any_call("token", help="Your Netbox Token.") + mock_parent_parser.add_argument.assert_any_call( + "file_path", help="Your file path to csv files." + ) + mock_parent_parser.add_argument.assert_any_call( + "--delimiter", help="The separator in the text file." + ) + mock_parent_parser.add_argument.assert_any_call( + "--sheet-name", help="The sheet in the Excel Workbook to read from." + ) + assert res == mock_parent_parser diff --git a/pynetbox_query/tests/test_query_device.py b/pynetbox_query/tests/test_utils/test_query_device.py similarity index 77% rename from pynetbox_query/tests/test_query_device.py rename to pynetbox_query/tests/test_utils/test_query_device.py index 1bb5c4fd..28121b37 100644 --- a/pynetbox_query/tests/test_query_device.py +++ b/pynetbox_query/tests/test_utils/test_query_device.py @@ -1,7 +1,7 @@ from unittest.mock import NonCallableMock, patch from dataclasses import asdict from pytest import fixture -from pynetbox_query.utils.query_device import QueryDevice +from pynetboxquery.utils.query_device import QueryDevice @fixture(name="instance") @@ -18,7 +18,7 @@ def test_query_list_no_device(instance): This test ensures that an empty list is returned if an empty list is given to the query list method. """ mock_device_list = [] - with patch("pynetbox_query.utils.query_device.QueryDevice.query_device"): + with patch("pynetboxquery.utils.query_device.QueryDevice.query_device"): res = instance.query_list(mock_device_list) assert res == [] @@ -28,7 +28,9 @@ def test_query_list_one_device(instance): This test ensures that one device is returned if one device is given to the method. """ mock_device_list = [""] - with patch("pynetbox_query.utils.query_device.QueryDevice.query_device") as mock_query_device: + with patch( + "pynetboxquery.utils.query_device.QueryDevice.query_device" + ) as mock_query_device: res = instance.query_list(mock_device_list) assert res == [mock_query_device.return_value] @@ -38,7 +40,9 @@ def test_query_list_multiple_devices(instance): This test ensures 2 devices are returned if 2 devices are given. """ mock_device_list = ["", ""] - with patch("pynetbox_query.utils.query_device.QueryDevice.query_device") as mock_query_device: + with patch( + "pynetboxquery.utils.query_device.QueryDevice.query_device" + ) as mock_query_device: res = instance.query_list(mock_device_list) assert res == [mock_query_device.return_value, mock_query_device.return_value] @@ -47,7 +51,7 @@ def test_query_device(instance, mock_device, dict_to_device): """ This test ensures the get_id is called on all fields in a dataclass. """ - with patch("pynetbox_query.utils.query_device.NetboxGetId.get_id") as mock_get_id: + with patch("pynetboxquery.utils.query_device.NetboxGetId.get_id") as mock_get_id: res = instance.query_device(mock_device) val = mock_get_id.return_value expected_device_dict = asdict(mock_device) diff --git a/pynetbox_query/tests/test_utils/test_read_utils/test_read_abc.py b/pynetbox_query/tests/test_utils/test_read_utils/test_read_abc.py new file mode 100644 index 00000000..262ac336 --- /dev/null +++ b/pynetbox_query/tests/test_utils/test_read_utils/test_read_abc.py @@ -0,0 +1,51 @@ +from dataclasses import asdict +from unittest.mock import patch +from pytest import raises +from pynetboxquery.utils.read_utils.read_abc import ReadAbstractBase + + +# pylint: disable = R0903 +class StubAbstractBase(ReadAbstractBase): + """ + This class provides a Stub version of the ReadAbstractBase class. + So we do not have to patch all the abstract methods. + """ + + def read(self): + """Placeholder method.""" + + @staticmethod + def _validate(kwargs): + """Placeholder method.""" + + +@patch("pynetboxquery.utils.read_utils.read_abc.Path") +def test_check_file_path(mock_path): + """ + This test ensures the _check_file_path method is called. + """ + StubAbstractBase("mock_file_path")._check_file_path("mock_file_path") + mock_path.assert_called_with("mock_file_path") + mock_path.return_value.exists.assert_called() + + +@patch("pynetboxquery.utils.read_utils.read_abc.Path") +def test_check_file_path_error(mock_path): + """ + This test ensures the _check_file_path method is called and raises an error. + """ + mock_path.return_value.exists.return_value = False + with raises(FileNotFoundError): + StubAbstractBase("mock_file_path")._check_file_path("mock_file_path") + mock_path.assert_called_with("mock_file_path") + mock_path.return_value.exists.assert_called() + + +def test_dict_to_dataclass(mock_device): + """ + This test ensures that the method _dict_to_dataclass returns the right list of devices. + """ + with patch("pynetboxquery.utils.read_utils.read_abc.Path"): + mock_dictionary = [asdict(mock_device)] + res = StubAbstractBase("mock_file_path")._dict_to_dataclass(mock_dictionary) + assert res == [mock_device] diff --git a/pynetbox_query/tests/test_utils/test_read_utils/test_read_csv.py b/pynetbox_query/tests/test_utils/test_read_utils/test_read_csv.py new file mode 100644 index 00000000..0faa056a --- /dev/null +++ b/pynetbox_query/tests/test_utils/test_read_utils/test_read_csv.py @@ -0,0 +1,28 @@ +from unittest.mock import patch +from pynetboxquery.utils.read_utils.read_csv import ReadCSV + + +@patch("pynetboxquery.utils.read_utils.read_csv.ReadCSV._check_file_path") +@patch("pynetboxquery.utils.read_utils.read_csv.open") +@patch("pynetboxquery.utils.read_utils.read_csv.DictReader") +@patch("pynetboxquery.utils.read_utils.read_csv.list") +@patch("pynetboxquery.utils.read_utils.read_csv.ReadCSV._dict_to_dataclass") +def test_read( + mock_dict_to_dataclass, + mock_list, + mock_dict_reader, + mock_open_func, + mock_check_file_path, +): + """ + This test ensures all calls are made correctly in the read method. + """ + res = ReadCSV("mock_file_path").read() + mock_check_file_path.assert_called_once_with("mock_file_path") + mock_open_func.assert_called_once_with("mock_file_path", mode="r", encoding="UTF-8") + mock_dict_reader.assert_called_once_with( + mock_open_func.return_value.__enter__.return_value + ) + mock_list.assert_called_once_with(mock_dict_reader.return_value) + mock_dict_to_dataclass.assert_called_once_with(mock_list.return_value) + assert res == mock_dict_to_dataclass.return_value diff --git a/pynetbox_query/tests/test_utils/test_read_utils/test_read_file.py b/pynetbox_query/tests/test_utils/test_read_utils/test_read_file.py new file mode 100644 index 00000000..d5130b3c --- /dev/null +++ b/pynetbox_query/tests/test_utils/test_read_utils/test_read_file.py @@ -0,0 +1,45 @@ +from unittest.mock import patch +from pytest import raises +from pynetboxquery.utils.read_utils.read_file import ReadFile +from pynetboxquery.utils.error_classes import FileTypeNotSupportedError + + +def test_read_file_wildcard(): + """ + This test ensures an error is raised when a user supplies an unsupported file type. + """ + with raises(FileTypeNotSupportedError): + ReadFile().read_file("mock_file_path.wildcard") + + +@patch("pynetboxquery.utils.read_utils.read_file.ReadCSV") +def test_read_file_csv(mock_read_csv): + """ + This test ensures the correct read method is called when supplied with a csv file path. + """ + res = ReadFile().read_file("mock_file_path.csv") + mock_read_csv.assert_called_once_with("mock_file_path.csv") + mock_read_csv.return_value.read.assert_called_once_with() + assert res == mock_read_csv.return_value.read.return_value + + +@patch("pynetboxquery.utils.read_utils.read_file.ReadTXT") +def test_read_file_txt(mock_read_txt): + """ + This test ensures the correct read method is called when supplied with a txt file path. + """ + res = ReadFile().read_file("mock_file_path.txt") + mock_read_txt.assert_called_once_with("mock_file_path.txt") + mock_read_txt.return_value.read.assert_called_once_with() + assert res == mock_read_txt.return_value.read.return_value + + +@patch("pynetboxquery.utils.read_utils.read_file.ReadXLSX") +def test_read_file_xlsx(mock_read_xlsx): + """ + This test ensures the correct read method is called when supplied with a xlsx file path. + """ + res = ReadFile().read_file("mock_file_path.xlsx") + mock_read_xlsx.assert_called_once_with("mock_file_path.xlsx") + mock_read_xlsx.return_value.read.assert_called_once_with() + assert res == mock_read_xlsx.return_value.read.return_value diff --git a/pynetbox_query/tests/test_utils/test_read_utils/test_read_txt.py b/pynetbox_query/tests/test_utils/test_read_utils/test_read_txt.py new file mode 100644 index 00000000..bad40a86 --- /dev/null +++ b/pynetbox_query/tests/test_utils/test_read_utils/test_read_txt.py @@ -0,0 +1,41 @@ +from unittest.mock import patch +from pytest import raises +from pynetboxquery.utils.read_utils.read_txt import ReadTXT +from pynetboxquery.utils.error_classes import DelimiterNotSpecifiedError + + +@patch("pynetboxquery.utils.read_utils.read_txt.ReadTXT._check_file_path") +@patch("pynetboxquery.utils.read_utils.read_txt.open") +@patch("pynetboxquery.utils.read_utils.read_txt.DictReader") +@patch("pynetboxquery.utils.read_utils.read_txt.list") +@patch("pynetboxquery.utils.read_utils.read_txt.ReadTXT._dict_to_dataclass") +def test_read( + mock_dict_to_dataclass, mock_list, mock_dict_reader, mock_open, mock_check_file_path +): + """ + This test ensures all calls are made correctly in the read method. + """ + res = ReadTXT("mock_file_path", **{"delimiter": ","}).read() + mock_check_file_path.assert_called_once_with("mock_file_path") + mock_open.assert_called_once_with("mock_file_path", mode="r", encoding="UTF-8") + mock_dict_reader.assert_called_once_with( + mock_open.return_value.__enter__.return_value, delimiter="," + ) + mock_list.assert_called_once_with(mock_dict_reader.return_value) + mock_dict_to_dataclass.assert_called_once_with(mock_list.return_value) + assert res == mock_dict_to_dataclass.return_value + + +def test_validate(): + """ + This test ensures the validate method is called and doesn't error for a correct case. + """ + ReadTXT("", **{"delimiter": ","}) + + +def test_validate_fail(): + """ + This test ensures the validate method is called and does error for an incorrect case. + """ + with raises(DelimiterNotSpecifiedError): + ReadTXT("") diff --git a/pynetbox_query/tests/test_utils/test_read_utils/test_read_xlsx.py b/pynetbox_query/tests/test_utils/test_read_utils/test_read_xlsx.py new file mode 100644 index 00000000..c0160519 --- /dev/null +++ b/pynetbox_query/tests/test_utils/test_read_utils/test_read_xlsx.py @@ -0,0 +1,37 @@ +from unittest.mock import patch +from pytest import raises +from pynetboxquery.utils.read_utils.read_xlsx import ReadXLSX +from pynetboxquery.utils.error_classes import SheetNameNotSpecifiedError + + +def test_validate(): + """ + This test ensures the validate method is called and doesn't error for a correct case. + """ + ReadXLSX("", **{"sheet_name": "test"}) + + +def test_validate_fail(): + """ + This test ensures the validate method is called and does error for an incorrect case. + """ + with raises(SheetNameNotSpecifiedError): + ReadXLSX("") + + +@patch("pynetboxquery.utils.read_utils.read_xlsx.ReadXLSX._check_file_path") +@patch("pynetboxquery.utils.read_utils.read_xlsx.read_excel") +@patch("pynetboxquery.utils.read_utils.read_xlsx.ReadXLSX._dict_to_dataclass") +def test_read(mock_dict_to_dataclass, mock_read_excel, mock_check_file_path): + """ + This test ensures all calls are made correctly in the read method. + """ + res = ReadXLSX("mock_file_path", **{"sheet_name": "test"}).read() + mock_check_file_path.assert_called_once_with("mock_file_path") + mock_read_excel.assert_called_once_with("mock_file_path", sheet_name="test") + mock_dataframe = mock_read_excel.return_value + mock_dataframe.to_dict.assert_called_once_with(orient="records") + mock_dictionary_list = mock_dataframe.to_dict.return_value + mock_dict_to_dataclass.assert_called_once_with(mock_dictionary_list) + mock_device_list = mock_dict_to_dataclass.return_value + assert res == mock_device_list