Refactor and fix types to get pyright happy

python/understack-workflows/tests/test_bmc_chassis_info.py

@@ -5,16 +5,17 @@
 from ipaddress import IPv4Interface
 
 from understack_workflows import bmc_chassis_info
+from understack_workflows.bmc import Bmc
 
-FIXTURE_PATH = "json_samples/bmc_chassis_info"
+FIXTURE_PATH = pathlib.Path("json_samples/bmc_chassis_info")
 
 
-class FakeBmc:
+class FakeBmc(Bmc):
     def __init__(self, fixtures):
         self.fixtures = fixtures
         self.ip_address = "1.2.3.4"
 
-    def redfish_request(self, path: str) -> dict:
+    def redfish_request(self, path: str, *_args, **_kw) -> dict:
         path = path.replace("/", "_") + ".json"
         return self.fixtures[path]
 

python/understack-workflows/tests/test_nautobot_device.py

@@ -162,7 +162,7 @@ def test_find_or_create():
                 description="Integrated NIC 1 Port 1",
                 mac_address="D4:04:E6:4F:8D:B4",
                 status="Active",
-                ip_address=[],
+                ip_address=None,
                 neighbor_device_id="275ef491-2b27-4d1b-bd45-330bd6b7e0cf",
                 neighbor_device_name="f20-2-1.iad3.rackspace.net",
                 neighbor_interface_id="f9a5cc87-d10a-4827-99e8-48961fd1d773",
@@ -178,7 +178,7 @@ def test_find_or_create():
                 description="Integrated NIC 1 Port 2",
                 mac_address="D4:04:E6:4F:8D:B5",
                 status="Active",
-                ip_address=[],
+                ip_address=None,
                 neighbor_device_id="05f6715a-4dbe-4fd6-af20-1e73adb285c2",
                 neighbor_device_name="f20-2-2.iad3.rackspace.net",
                 neighbor_interface_id="2148cf50-f70e-42c9-9f68-8ce98d61498c",
@@ -194,7 +194,7 @@ def test_find_or_create():
                 description="NIC in Slot 1 Port 1",
                 mac_address="14:23:F3:F5:25:F0",
                 status="Active",
-                ip_address=[],
+                ip_address=None,
                 neighbor_device_id="05f6715a-4dbe-4fd6-af20-1e73adb285c2",
                 neighbor_device_name="f20-2-2.iad3.rackspace.net",
                 neighbor_interface_id="f72bb830-3f3c-4aba-b7d5-9680ea4d358e",
@@ -210,7 +210,7 @@ def test_find_or_create():
                 description="NIC in Slot 1 Port 2",
                 mac_address="14:23:F3:F5:25:F1",
                 status="Active",
-                ip_address=[],
+                ip_address=None,
                 neighbor_device_id="275ef491-2b27-4d1b-bd45-330bd6b7e0cf",
                 neighbor_device_name="f20-2-1.iad3.rackspace.net",
                 neighbor_interface_id="c210be75-1038-4ba3-9923-60050e1c5362",

python/understack-workflows/tests/test_topology.py

@@ -17,8 +17,8 @@
             type="A_25GBASE_X_SFP28",
             description="Integrated NIC 1 Port 1",
             mac_address="D4:04:E6:4F:8D:B4",
-            status={"id": "d4bcbafa-3033-433b-b21b-a20acf9d1324"},
-            ip_address=[],
+            status="Active",
+            ip_address=None,
             neighbor_device_id="275ef491-2b27-4d1b-bd45-330bd6b7e0cf",
             neighbor_device_name="f20-2-1.iad3.rackspace.net",
             neighbor_interface_id="f9a5cc87-d10a-4827-99e8-48961fd1d773",
@@ -33,8 +33,8 @@
             type="A_25GBASE_X_SFP28",
             description="Integrated NIC 1 Port 2",
             mac_address="D4:04:E6:4F:8D:B5",
-            status={"id": "d4bcbafa-3033-433b-b21b-a20acf9d1324"},
-            ip_address=[],
+            status="Active",
+            ip_address=None,
             neighbor_device_id="05f6715a-4dbe-4fd6-af20-1e73adb285c2",
             neighbor_device_name="f20-2-2.iad3.rackspace.net",
             neighbor_interface_id="2148cf50-f70e-42c9-9f68-8ce98d61498c",

python/understack-workflows/understack_workflows/bmc.py

@@ -9,7 +9,7 @@
 from understack_workflows.bmc_password_standard import standard_password
 from understack_workflows.helpers import credential
 
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # type: ignore
 logging.getLogger("urllib3").setLevel(logging.WARNING)
 
 HEADERS = {
@@ -58,6 +58,8 @@ def redfish_request(
             )
         if r.text:
             return r.json()
+        else:
+            return {}
 
     def sushy(self):
         return Sushy(

python/understack-workflows/understack_workflows/bmc_credentials.py

@@ -3,7 +3,7 @@
 
 from understack_workflows.helpers import setup_logger
 
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # type: ignore
 
 FACTORY_PASSWORD = "calvin"
 
@@ -45,7 +45,7 @@ def set_bmc_password(ip_address, required_password, username="root"):
         logger.info("Production BMC credentials are working on this BMC.")
 
 
-def _verify_auth(host: str, username: str = "root", password: str = "") -> str:
+def _verify_auth(host: str, username: str = "root", password: str = "") -> str | None:
     """Test whether provided credentials work against a secured API endpoint.
 
     Returns Session authentication token on success

python/understack-workflows/understack_workflows/bmc_network_config.py

@@ -16,6 +16,9 @@ def bmc_set_permanent_ip_addr(bmc: Bmc, interface_info: InterfaceInfo):
         logger.info("BMC interface was not set to DHCP")
         return
 
+    if not (interface_info.ipv4_address and interface_info.ipv4_gateway):
+        raise ValueError("BMC InterfaceInfo has missing IP information")
+
     payload = {
         "Attributes": {
             "IPv4.1.DHCPEnable": "Disabled",

python/understack-workflows/understack_workflows/ironic/client.py

@@ -58,7 +58,7 @@ def delete_port(self, port_id: str):
             port_id,
         )
 
-    def list_ports(self, node_id: dict):
+    def list_ports(self, node_id: str):
         self._ensure_logged_in()
 
         return self.client.port.list(node=node_id, detail=True)

python/understack-workflows/understack_workflows/ironic_node.py

@@ -29,7 +29,7 @@ def create_or_update(
         ironic_node = create_ironic_node(
             client, node_uuid, device_hostname, driver, bmc
         )
-        return ironic_node.provision_state
+        return ironic_node.provision_state  # type: ignore
 
     if ironic_node.provision_state in STATES_ALLOWING_UPDATES:
         update_ironic_node(client, node_uuid, device_hostname, driver, bmc)

python/understack-workflows/understack_workflows/main/print_bmc_password.py

@@ -18,7 +18,7 @@ def main(program_name, bmc_ip_address=None):
         print("Please set the BMC_MASTER environment variable", file=sys.stderr)
         exit(1)
 
-    password = standard_password(bmc_ip_address, os.getenv("BMC_MASTER"))
+    password = standard_password(str(bmc_ip_address), str(os.getenv("BMC_MASTER")))
     print(password)
 
 

python/understack-workflows/understack_workflows/main/sync_keystone.py

@@ -71,16 +71,16 @@ def handle_project_create(conn: Connection, nautobot: Nautobot, project_id: uuid
     ten = ten_api.create(
         id=str(project_id), name=project.name, description=project.description
     )
-    logger.info(f"tenant '{project_id!s}' created {ten.created}")
+    logger.info(f"tenant '{project_id!s}' created {ten.created}")  # type: ignore
 
 
 def handle_project_update(conn: Connection, nautobot: Nautobot, project_id: uuid.UUID):
     logger.info(f"got request to update tenant {project_id!s}")
     project = conn.identity.get_project(project_id.hex)
     ten = nautobot.session.tenancy.tenants.get(project_id)
-    ten.description = project.description
-    ten.save()
-    logger.info(f"tenant '{project_id!s}' last updated {ten.last_updated}")
+    ten.description = project.description  # type: ignore
+    ten.save()  # type: ignore
+    logger.info(f"tenant '{project_id!s}' last updated {ten.last_updated}")  # type: ignore
 
 
 def handle_project_delete(conn: Connection, nautobot: Nautobot, project_id: uuid.UUID):
@@ -89,7 +89,7 @@ def handle_project_delete(conn: Connection, nautobot: Nautobot, project_id: uuid
     if not ten:
         logger.warn(f"tenant '{project_id!s}' does not exist already")
         return
-    ten.delete()
+    ten.delete()  # type: ignore
     logger.info(f"deleted tenant {project_id!s}")
 
 

python/understack-workflows/understack_workflows/main/undersync_device.py

@@ -50,6 +50,8 @@ def update_nautobot_for_provisioning(
     interface = nautobot.update_switch_interface_status(
         device_id, interface_mac, new_status
     )
+    if not interface.device:
+        raise Exception("Interface has no associated device")
     vlan_group_id = vlan_group_id_for(interface.device.id, nautobot)
     logger.debug(
         f"Switch interface {interface.device} {interface} found in {vlan_group_id=}"

python/understack-workflows/understack_workflows/models.py

@@ -23,25 +23,26 @@ class NIC:
     @classmethod
     def from_redfish(cls, data: NetworkAdapter) -> NIC:
         location = cls.nic_location(data)
-        nic = cls(data.identity, location, [], data.model)
+        nic = cls(data.identity, location, [], data.model)  # type: ignore
         nic.interfaces = [Interface.from_redfish(i, nic) for i in cls.nic_ports(data)]
         return nic
 
     @classmethod
     def from_hp_json(cls, data: dict) -> NIC:
-        nic = cls(data.get("name"), data.get("location"), [], data.get("name"))
-        ports = data.get("network_ports") or data.get("unknown_ports")
+        nic = cls(data["name"], data["location"], [], data["name"])
+        ports = data.get("network_ports") or data.get("unknown_ports", [])
         nic.interfaces = [Interface.from_hp_json(i, nic, ports) for i in ports]
         return nic
 
     @classmethod
     def nic_location(cls, nic: NetworkAdapter) -> str:
-        try:
-            return nic.json["Controllers"][0]["Location"]["PartLocation"][
-                "ServiceLabel"
-            ]
-        except KeyError:
-            return nic.identity
+        json = nic.json or {}
+        controller = json.get("Controllers", [])[0] or {}
+        return (
+            controller.get("Location", {})
+            .get("PartLocation", {})
+            .get("ServiceLabel", nic.identity)
+        )
 
     @classmethod
     def nic_ports(cls, nic: NetworkAdapter) -> list[NetworkPort]:
@@ -58,17 +59,17 @@ class Interface:
 
     @classmethod
     def from_redfish(cls, data: NetworkPort, nic: NIC) -> Interface:
-        if data.root.json["Vendor"] == "HPE":
+        if data.root and data.root.json["Vendor"] == "HPE":
             name = f"{nic.name}_{data.physical_port_number}"
-            macaddr = data.associated_network_addresses[0]
+            macaddr = data.associated_network_addresses[0]  # type: ignore
         else:
             name = data.identity
             macaddr = cls.fetch_macaddr_from_sys_resource(data)
         return cls(
-            name,
+            name,  # type: ignore
             macaddr,
             nic.location,
-            data.current_link_speed_mbps,
+            data.current_link_speed_mbps,  # type: ignore
             nic.model,
         )
 
@@ -78,18 +79,18 @@ def from_hp_json(cls, data: dict, nic: NIC, ports: list) -> Interface:
         interface_name = f"NIC.{nic.location.replace(' ', '.')}_{p_num}"
         return cls(
             interface_name,
-            data.get("mac_addr"),
+            data["mac_addr"],
             nic.location,
-            data.get("speed", 0),
+            data["speed", 0],
             nic.model,
         )
 
     @classmethod
     def fetch_macaddr_from_sys_resource(cls, data: NetworkPort) -> str:
         try:
-            path = f"{data.root.get_system().ethernet_interfaces.path}/{data.identity}"
+            path = f"{data.root.get_system().ethernet_interfaces.path}/{data.identity}"  # type: ignore
             macaddr = (
-                data.root.get_system().ethernet_interfaces.get_member(path).mac_address
+                data.root.get_system().ethernet_interfaces.get_member(path).mac_address  # type: ignore
             )
         except ResourceNotFoundError:
             macaddr = ""
@@ -116,7 +117,7 @@ class Chassis:
     name: str
     nics: list[NIC]
     network_interfaces: list[Interface]
-    system_info: Systeminfo
+    system_info: Systeminfo | None
 
     @classmethod
     def check_manufacturer(cls, manufacturer: str) -> None:
@@ -137,28 +138,31 @@ def bmc_is_ilo4(cls, chassis_data: SushyChassis) -> bool:
     @classmethod
     def from_redfish(cls, oob_obj: Sushy) -> Chassis:
         chassis_data = oob_obj.get_chassis(
-            oob_obj.get_chassis_collection().members_identities[0]
+            oob_obj.get_chassis_collection().members_identities[0]  # type: ignore
         )
 
-        cls.check_manufacturer(chassis_data.manufacturer)
+        cls.check_manufacturer(chassis_data.manufacturer)  # type: ignore
 
         if cls.bmc_is_ilo4(chassis_data):
-            return cls.from_hp_json(oob_obj, chassis_data.name)
+            return cls.from_hp_json(oob_obj, chassis_data.name)  # type: ignore
 
-        chassis = cls(chassis_data.name, [], [], [])
-        chassis.nics = [
+        nics = [
             NIC.from_redfish(i) for i in chassis_data.network_adapters.get_members()
         ]
-        chassis.network_interfaces = cls.interfaces_from_nics(chassis.nics)
-        chassis.system_info = Systeminfo.from_redfish(chassis_data)
-        return chassis
+
+        return cls(
+            name=chassis_data.name,  # type: ignore
+            nics=nics,
+            network_interfaces=cls.interfaces_from_nics(nics),
+            system_info=Systeminfo.from_redfish(chassis_data),
+        )
 
     @classmethod
     def from_hp_json(cls, oob_obj: Sushy, chassis_name: str) -> Chassis:
         data = cls.chassis_hp_json_data(oob_obj)
         nics = [NIC.from_hp_json(i) for i in data]
         network_interfaces = cls.interfaces_from_nics(nics)
-        return cls(chassis_name, nics, network_interfaces)
+        return cls(chassis_name, nics, network_interfaces, None)
 
     @classmethod
     def interfaces_from_nics(cls, nics: list[NIC]) -> list[Interface]:
@@ -167,7 +171,8 @@ def interfaces_from_nics(cls, nics: list[NIC]) -> list[Interface]:
     @classmethod
     def chassis_hp_json_data(cls, oob_obj: Sushy) -> dict:
         oob_obj._conn.set_http_basic_auth(
-            username=oob_obj._auth._username, password=oob_obj._auth._password
+            username=oob_obj._auth._username,  # type: ignore
+            password=oob_obj._auth._password,  # type: ignore
         )
         resp = oob_obj._conn.get(path="/json/comm_controller_info")
         resp.raise_for_status()