sources/azure: address mypy/pyright typing complaints (#1245)

Raise runtime errors for unhandled cases which would cause other
exceptions.  Ignore types for a few cases where a non-trivial
refactor would be required to prevent the warning.

Signed-off-by: Chris Patterson <cpatterson@microsoft.com>

5 files changed, 57 insertions, 43 deletions

diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py
index a261c16e..76acd6a3 100755
--- a/cloudinit/distros/__init__.py
+++ b/cloudinit/distros/__init__.py
@@ -16,7 +16,7 @@ import stat
 import string
 import urllib.parse
 from io import StringIO
-from typing import Any, Mapping  # noqa: F401
+from typing import Any, Mapping, Type
 
 from cloudinit import importer
 from cloudinit import log as logging
@@ -26,7 +26,7 @@ from cloudinit.features import ALLOW_EC2_MIRRORS_ON_NON_AWS_INSTANCE_TYPES
 from cloudinit.net import activators, eni, network_state, renderers
 from cloudinit.net.network_state import parse_net_config_data
 
-from .networking import LinuxNetworking
+from .networking import LinuxNetworking, Networking
 
 # Used when a cloud-config module can be run on all cloud-init distibutions.
 # The value 'all' is surfaced in module documentation for distro support.
@@ -76,9 +76,9 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta):
     hostname_conf_fn = "/etc/hostname"
     tz_zone_dir = "/usr/share/zoneinfo"
     init_cmd = ["service"]  # systemctl, service etc
-    renderer_configs = {}  # type: Mapping[str, Mapping[str, Any]]
+    renderer_configs: Mapping[str, Mapping[str, Any]] = {}
     _preferred_ntp_clients = None
-    networking_cls = LinuxNetworking
+    networking_cls: Type[Networking] = LinuxNetworking
     # This is used by self.shutdown_command(), and can be overridden in
     # subclasses
     shutdown_options_map = {"halt": "-H", "poweroff": "-P", "reboot": "-r"}
@@ -91,7 +91,7 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta):
         self._paths = paths
         self._cfg = cfg
         self.name = name
-        self.networking = self.networking_cls()
+        self.networking: Networking = self.networking_cls()
 
     def _unpickle(self, ci_pkl_version: int) -> None:
         """Perform deserialization fixes for Distro."""
diff --git a/cloudinit/distros/networking.py b/cloudinit/distros/networking.py
index e18a48ca..b24b6233 100644
--- a/cloudinit/distros/networking.py
+++ b/cloudinit/distros/networking.py
@@ -1,6 +1,7 @@
 import abc
 import logging
 import os
+from typing import List, Optional
 
 from cloudinit import net, subp, util
 
@@ -22,7 +23,7 @@ class Networking(metaclass=abc.ABCMeta):
     """
 
     def __init__(self):
-        self.blacklist_drivers = None
+        self.blacklist_drivers: Optional[List[str]] = None
 
     def _get_current_rename_info(self) -> dict:
         return net._get_current_rename_info()
diff --git a/cloudinit/sources/DataSourceAzure.py b/cloudinit/sources/DataSourceAzure.py
index f4be4cda..2566bd8e 100755
--- a/cloudinit/sources/DataSourceAzure.py
+++ b/cloudinit/sources/DataSourceAzure.py
@@ -13,7 +13,7 @@ import re
 import xml.etree.ElementTree as ET
 from enum import Enum
 from time import sleep, time
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 from xml.dom import minidom
 
 import requests
@@ -83,7 +83,7 @@ class PPSType(Enum):
     UNKNOWN = "Unknown"
 
 
-PLATFORM_ENTROPY_SOURCE = "/sys/firmware/acpi/tables/OEM0"
+PLATFORM_ENTROPY_SOURCE: Optional[str] = "/sys/firmware/acpi/tables/OEM0"
 
 # List of static scripts and network config artifacts created by
 # stock ubuntu suported images.
@@ -155,7 +155,7 @@ def find_busdev_from_disk(camcontrol_out, disk_drv):
     return None
 
 
-def find_dev_from_busdev(camcontrol_out, busdev):
+def find_dev_from_busdev(camcontrol_out: str, busdev: str) -> Optional[str]:
     # find the daX from 'camcontrol devlist' output
     # if busdev matches the specified value, i.e. 'scbus2'
     """
@@ -172,9 +172,9 @@ def find_dev_from_busdev(camcontrol_out, busdev):
     return None
 
 
-def execute_or_debug(cmd, fail_ret=None):
+def execute_or_debug(cmd, fail_ret=None) -> str:
     try:
-        return subp.subp(cmd)[0]
+        return subp.subp(cmd)[0]  # type: ignore
     except subp.ProcessExecutionError:
         LOG.debug("Failed to execute: %s", " ".join(cmd))
         return fail_ret
@@ -192,7 +192,7 @@ def get_camcontrol_dev():
     return execute_or_debug(["camcontrol", "devlist"])
 
 
-def get_resource_disk_on_freebsd(port_id):
+def get_resource_disk_on_freebsd(port_id) -> Optional[str]:
     g0 = "00000000"
     if port_id > 1:
         g0 = "00000001"
@@ -316,7 +316,9 @@ class DataSourceAzure(sources.DataSource):
 
     def _get_subplatform(self):
         """Return the subplatform metadata source details."""
-        if self.seed.startswith("/dev"):
+        if self.seed is None:
+            subplatform_type = "unknown"
+        elif self.seed.startswith("/dev"):
             subplatform_type = "config-disk"
         elif self.seed.lower() == "imds":
             subplatform_type = "imds"
@@ -541,9 +543,9 @@ class DataSourceAzure(sources.DataSource):
         if metadata_source == "IMDS" and not crawled_data["files"]:
             try:
                 contents = build_minimal_ovf(
-                    username=imds_username,
-                    hostname=imds_hostname,
-                    disableSshPwd=imds_disable_password,
+                    username=imds_username,  # type: ignore
+                    hostname=imds_hostname,  # type: ignore
+                    disableSshPwd=imds_disable_password,  # type: ignore
                 )
                 crawled_data["files"] = {"ovf-env.xml": contents}
             except Exception as e:
@@ -803,7 +805,11 @@ class DataSourceAzure(sources.DataSource):
         # We don't want Azure to react to an UPPER/lower difference as a new
         # instance id as it rewrites SSH host keys.
         # LP: #1835584
-        iid = dmi.read_dmi_data("system-uuid").lower()
+        system_uuid = dmi.read_dmi_data("system-uuid")
+        if system_uuid is None:
+            raise RuntimeError("failed to read system-uuid")
+
+        iid = system_uuid.lower()
         if os.path.exists(prev_iid_path):
             previous = util.load_file(prev_iid_path).strip()
             if previous.lower() == iid:
@@ -866,7 +872,7 @@ class DataSourceAzure(sources.DataSource):
                 path, "{pid}: {time}\n".format(pid=os.getpid(), time=time())
             )
         except AssertionError as error:
-            report_diagnostic_event(error, logger_func=LOG.error)
+            report_diagnostic_event(str(error), logger_func=LOG.error)
             raise
 
     @azure_ds_telemetry_reporter
@@ -888,10 +894,10 @@ class DataSourceAzure(sources.DataSource):
         attempts = 0
         LOG.info("Unbinding and binding the interface %s", ifname)
         while True:
-
-            devicename = net.read_sys_net(ifname, "device/device_id").strip(
-                "{}"
-            )
+            device_id = net.read_sys_net(ifname, "device/device_id")
+            if device_id is False or not isinstance(device_id, str):
+                raise RuntimeError("Unable to read device ID: %s" % device_id)
+            devicename = device_id.strip("{}")
             util.write_file(
                 "/sys/bus/vmbus/drivers/hv_netvsc/unbind", devicename
             )
@@ -1118,7 +1124,7 @@ class DataSourceAzure(sources.DataSource):
                     break
 
         except AssertionError as error:
-            report_diagnostic_event(error, logger_func=LOG.error)
+            report_diagnostic_event(str(error), logger_func=LOG.error)
 
     @azure_ds_telemetry_reporter
     def _wait_for_all_nics_ready(self):
@@ -1168,7 +1174,7 @@ class DataSourceAzure(sources.DataSource):
                     logger_func=LOG.info,
                 )
         except netlink.NetlinkCreateSocketError as e:
-            report_diagnostic_event(e, logger_func=LOG.warning)
+            report_diagnostic_event(str(e), logger_func=LOG.warning)
             raise
         finally:
             if nl_sock:
@@ -1234,6 +1240,11 @@ class DataSourceAzure(sources.DataSource):
                 self._setup_ephemeral_networking(timeout_minutes=20)
 
             try:
+                if (
+                    self._ephemeral_dhcp_ctx is None
+                    or self._ephemeral_dhcp_ctx.iface is None
+                ):
+                    raise RuntimeError("Missing ephemeral context")
                 iface = self._ephemeral_dhcp_ctx.iface
 
                 nl_sock = netlink.create_bound_netlink_socket()
@@ -1873,7 +1884,7 @@ def read_azure_ovf(contents):
         raise BrokenAzureDataSource("no child nodes of configuration set")
 
     md_props = "seedfrom"
-    md = {"azure_data": {}}
+    md: Dict[str, Any] = {"azure_data": {}}
     cfg = {}
     ud = ""
     password = None
@@ -2084,9 +2095,7 @@ def _get_random_seed(source=PLATFORM_ENTROPY_SOURCE):
     # string. Same number of bits of entropy, just with 25% more zeroes.
     # There's no need to undo this base64-encoding when the random seed is
     # actually used in cc_seed_random.py.
-    seed = base64.b64encode(seed).decode()
-
-    return seed
+    return base64.b64encode(seed).decode()  # type: ignore
 
 
 @azure_ds_telemetry_reporter
@@ -2151,7 +2160,7 @@ def _generate_network_config_from_imds_metadata(imds_metadata) -> dict:
     @param: imds_metadata: Dict of content read from IMDS network service.
     @return: Dictionary containing network version 2 standard configuration.
     """
-    netconfig = {"version": 2, "ethernets": {}}
+    netconfig: Dict[str, Any] = {"version": 2, "ethernets": {}}
     network_metadata = imds_metadata["network"]
     for idx, intf in enumerate(network_metadata["interface"]):
         has_ip_address = False
@@ -2160,7 +2169,7 @@ def _generate_network_config_from_imds_metadata(imds_metadata) -> dict:
         # addresses.
         nicname = "eth{idx}".format(idx=idx)
         dhcp_override = {"route-metric": (idx + 1) * 100}
-        dev_config = {
+        dev_config: Dict[str, Any] = {
             "dhcp4": True,
             "dhcp4-overrides": dhcp_override,
             "dhcp6": False,
@@ -2214,9 +2223,12 @@ def _generate_network_config_from_fallback_config() -> dict:
 
     @return: Dictionary containing network version 2 standard configuration.
     """
-    return net.generate_fallback_config(
+    cfg = net.generate_fallback_config(
         blacklist_drivers=BLACKLIST_DRIVERS, config_driver=True
     )
+    if cfg is None:
+        return {}
+    return cfg
 
 
 @azure_ds_telemetry_reporter
diff --git a/cloudinit/sources/helpers/azure.py b/cloudinit/sources/helpers/azure.py
index ec6ab80c..d07dc3c0 100755
--- a/cloudinit/sources/helpers/azure.py
+++ b/cloudinit/sources/helpers/azure.py
@@ -334,12 +334,10 @@ def _get_dhcp_endpoint_option_name():
 
 
 @azure_ds_telemetry_reporter
-def http_with_retries(url, **kwargs) -> str:
+def http_with_retries(url, **kwargs) -> url_helper.UrlResponse:
     """Wrapper around url_helper.readurl() with custom telemetry logging
     that url_helper.readurl() does not provide.
     """
-    exc = None
-
     max_readurl_attempts = 240
     default_readurl_timeout = 5
     sleep_duration_between_retries = 5
@@ -374,16 +372,18 @@ def http_with_retries(url, **kwargs) -> str:
             return ret
 
         except Exception as e:
-            exc = e
             if attempt % periodic_logging_attempts == 0:
                 report_diagnostic_event(
                     "Failed HTTP request with Azure endpoint %s during "
                     "attempt %d with exception: %s" % (url, attempt, e),
                     logger_func=LOG.debug,
                 )
-            time.sleep(sleep_duration_between_retries)
+            if attempt == max_readurl_attempts:
+                raise
+
+        time.sleep(sleep_duration_between_retries)
 
-    raise exc
+    raise RuntimeError("Failed to return in http_with_retries")
 
 
 def build_minimal_ovf(
@@ -433,14 +433,16 @@ class AzureEndpointHttpClient:
             "x-ms-guest-agent-public-x509-cert": certificate,
         }
 
-    def get(self, url, secure=False):
+    def get(self, url, secure=False) -> url_helper.UrlResponse:
         headers = self.headers
         if secure:
             headers = self.headers.copy()
             headers.update(self.extra_secure_headers)
         return http_with_retries(url, headers=headers)
 
-    def post(self, url, data=None, extra_headers=None):
+    def post(
+        self, url, data=None, extra_headers=None
+    ) -> url_helper.UrlResponse:
         headers = self.headers
         if extra_headers is not None:
             headers = self.headers.copy()
diff --git a/cloudinit/url_helper.py b/cloudinit/url_helper.py
index 790e2fbf..5b2f2ef9 100644
--- a/cloudinit/url_helper.py
+++ b/cloudinit/url_helper.py
@@ -188,7 +188,7 @@ def readurl(
     infinite=False,
     log_req_resp=True,
     request_method=None,
-):
+) -> UrlResponse:
     """Wrapper around requests.Session to read the url and retry if necessary
 
     :param url: Mandatory url to request.
@@ -339,9 +339,8 @@ def readurl(
                         sec_between,
                     )
                 time.sleep(sec_between)
-    if excps:
-        raise excps[-1]
-    return None  # Should throw before this...
+
+    raise excps[-1]
 
 
 def wait_for_url(