Merge pull request #116 from stfc/dictionaries-to-dataclass

Dictionaries to dataclass

pynetbox_data_uploader/.pylintrc

@@ -5,6 +5,5 @@ max-line-length=120
 
 # Disable various warnings:
 # C0114: Missing module string - we don't need module strings for the small repo
-# W0511: TODOs they're well....to do later
 
-disable=C0114,W0511
+disable=C0114

pynetbox_data_uploader/lib/user_methods/csv_to_netbox.py

@@ -1,4 +1,5 @@
 from typing import List
+import pathlib
 import argparse
 from lib.netbox_api.netbox_create import NetboxCreate
 from lib.netbox_api.netbox_connect import NetboxConnect
@@ -14,6 +15,7 @@ class CsvToNetbox:
     """
     This class contains organised methods in the 4 step proccess of reading csv's to then uploading to Netbox.
     """
+
     def __init__(self, url: str, token: str):
         """
         This initialises the class with the following parameters.
@@ -34,6 +36,10 @@ def read_csv(self, file_path) -> List:
         :return: Returns a list of devices
         """
         print("Reading CSV...")
+        try:
+            pathlib.Path(file_path).exists()
+        except FileNotFoundError:
+            raise Exception("The given path is not valid.", FileNotFoundError)
         device_data = self.format_dict.csv_to_python(file_path)
         device_list = self.format_dict.separate_data(device_data)
         print("Read CSV.")

pynetbox_data_uploader/lib/utils/csv_to_dataclass.py

@@ -0,0 +1,41 @@
+import csv
+from typing import List
+from lib.utils.dataclass_data import Device
+
+
+def open_file(file_path: str) -> csv.DictReader:
+    """
+    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 csv_reader_obj
+
+
+def separate_data(csv_reader_obj: csv.DictReader) -> List[Device]:
+    """
+    This method separates the data from the iterator object into a list of dataclasses for each device.
+    :param csv_reader_obj: The DictReader class with the data.
+    :return: Returns a list of dataclass objects.
+    """
+    devices = []
+    for row in csv_reader_obj:
+        device = Device(
+            tenant=row["tenant"],
+            device_role=row["device_role"],
+            manufacturer=row["manufacturer"],
+            device_type=row["device_type"],
+            status=row["status"],
+            site=row["site"],
+            location=row["location"],
+            rack=row["rack"],
+            face=row["face"],
+            airflow=row["airflow"],
+            position=row["position"],
+            name=row["name"],
+            serial=row["serial"],
+        )
+        devices.append(device)
+    return devices

pynetbox_data_uploader/lib/utils/dataclass_data.py

@@ -0,0 +1,24 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+# pylint: disable = R0902
+@dataclass
+class Device:
+    """
+    This class instantiates device objects with the device data.
+    """
+
+    tenant: str
+    device_role: str
+    manufacturer: str
+    device_type: str
+    status: str
+    site: str
+    location: str
+    rack: str
+    position: str
+    name: str
+    serial: str
+    face: Optional[str] = None
+    airflow: Optional[str] = None

pynetbox_data_uploader/lib/utils/format_dict.py

@@ -34,11 +34,12 @@ def format_dict(self, dictionary) -> Dict:
         :return: Returns the formatted dictionary
         """
         for key in dictionary:
-            netbox_id = NetboxGetID(self.netbox).get_id_from_key(key=key, dictionary=dictionary)
+            netbox_id = NetboxGetID(self.netbox).get_id_from_key(
+                key=key, dictionary=dictionary
+            )
             dictionary[key] = netbox_id
         return dictionary
 
-
     @staticmethod
     def csv_to_python(file_path: str) -> Dict:
         """

pynetbox_data_uploader/tests/test_csv_to_dataclass.py

@@ -0,0 +1,60 @@
+from csv import DictReader
+from unittest.mock import patch, mock_open
+from lib.utils.csv_to_dataclass import separate_data, open_file
+from lib.utils.dataclass_data import Device
+
+
+def test_separate_data():
+    """
+    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] == Device(
+        tenant="t1",
+        device_role="dr1",
+        manufacturer="m1",
+        device_type="dt1",
+        status="st1",
+        site="si1",
+        location="l1",
+        rack="r1",
+        face="f1",
+        airflow="a1",
+        position="p1",
+        name="n1",
+        serial="se1",
+    )
+    assert res[1] == Device(
+        tenant="t2",
+        device_role="dr2",
+        manufacturer="m2",
+        device_type="dt2",
+        status="st2",
+        site="si2",
+        location="l2",
+        rack="r2",
+        face="f2",
+        airflow="a2",
+        position="p2",
+        name="n2",
+        serial="se2",
+    )
+
+
+def test_open_file():
+    """
+    This test ensures the csv file is opened appropriately and the DictReader method is called.
+    """
+    with patch("builtins.open", mock_open(read_data="mock_file_path")) as mock_file:
+        with patch("csv.DictReader") as mock_dict_reader:
+            res = open_file("mock_file_path")
+    mock_file.assert_called_once_with("mock_file_path", encoding="UTF-8")
+    mock_dict_reader.assert_called_once_with(mock_file.return_value)
+    assert res == mock_dict_reader.return_value

pynetbox_data_uploader/tests/test_format_dict.py

@@ -87,7 +87,9 @@ def test_format_dict_no_keys(instance):
     This test ensures the get_id_from_key method is not called.
     """
     mock_dict = {}
-    with patch("lib.netbox_api.netbox_get_id.NetboxGetID.get_id_from_key") as mock_get_id:
+    with patch(
+        "lib.netbox_api.netbox_get_id.NetboxGetID.get_id_from_key"
+    ) as mock_get_id:
         res = instance.format_dict(mock_dict)
     mock_get_id.assert_not_called()
     assert res == mock_dict
@@ -98,7 +100,9 @@ def test_format_dict_one_key(instance):
     This test ensures the get_id_from_key method is called once.
     """
     mock_dict = {"test": "data"}
-    with patch("lib.netbox_api.netbox_get_id.NetboxGetID.get_id_from_key") as mock_get_id:
+    with patch(
+        "lib.netbox_api.netbox_get_id.NetboxGetID.get_id_from_key"
+    ) as mock_get_id:
         res = instance.format_dict(mock_dict)
     mock_get_id.assert_called_once()
     assert res == mock_dict
@@ -108,9 +112,10 @@ def test_format_dict_many_keys(instance):
     """
     This test ensures the get_id_from_key method is called the correct number of times.
     """
-    mock_dict = {"test": "data",
-                 "test1": "data1"}
-    with patch("lib.netbox_api.netbox_get_id.NetboxGetID.get_id_from_key") as mock_get_id:
+    mock_dict = {"test": "data", "test1": "data1"}
+    with patch(
+        "lib.netbox_api.netbox_get_id.NetboxGetID.get_id_from_key"
+    ) as mock_get_id:
         res = instance.format_dict(mock_dict)
     mock_get_id.assert_called()
     assert mock_get_id.call_count == len(mock_dict.keys())

pynetbox_data_uploader/tests/test_netbox_get_id.py

@@ -21,8 +21,7 @@ def test_get_id_from_key_with_id_enums(instance):
     """
     with patch("lib.netbox_api.netbox_get_id.NetboxGetID.get_id") as mock_get_id:
         for member in [prop.name for prop in DeviceInfoID]:
-            mock_dictionary = {member: "abc",
-                               "site": "def"}
+            mock_dictionary = {member: "abc", "site": "def"}
             res = instance.get_id_from_key(key=member, dictionary=mock_dictionary)
             mock_get_id.assert_called()
             assert res == mock_get_id.return_value
@@ -34,8 +33,7 @@ def test_get_id_from_key_with_no_id_enums(instance):
     """
     with patch("lib.netbox_api.netbox_get_id.NetboxGetID.get_id") as mock_get_id:
         for member in [prop.name for prop in DeviceInfoNoID]:
-            mock_dictionary = {member: "abc",
-                               "site": "def"}
+            mock_dictionary = {member: "abc", "site": "def"}
             res = instance.get_id_from_key(key=member, dictionary=mock_dictionary)
             mock_get_id.assert_not_called()
             assert res == mock_dictionary[member]