fix: type annotations for several modules (#5733)

Remove those modules from the override list.

cloudinit/config/schema.py

@@ -508,22 +508,13 @@ def get_jsonschema_validator():
     meta_schema["properties"]["label"] = {"type": "string"}
 
     validator_kwargs = {}
-    if hasattr(Draft4Validator, "TYPE_CHECKER"):  # jsonschema 3.0+
-        meta_schema["additionalProperties"] = False  # Unsupported in 2.6.0
-        type_checker = Draft4Validator.TYPE_CHECKER.redefine(
-            "string", is_schema_byte_string
-        )
-        validator_kwargs = {
-            "type_checker": type_checker,
-        }
-    else:  # jsonschema 2.6 workaround
-        # pylint:disable-next=no-member
-        types = Draft4Validator.DEFAULT_TYPES  # pylint: disable=E1101
-        # Allow bytes as well as string (and disable a spurious unsupported
-        # assignment-operation pylint warning which appears because this
-        # code path isn't written against the latest jsonschema).
-        types["string"] = (str, bytes)  # pylint: disable=E1137
-        validator_kwargs = {"default_types": types}
+    meta_schema["additionalProperties"] = False
+    type_checker = Draft4Validator.TYPE_CHECKER.redefine(
+        "string", is_schema_byte_string
+    )
+    validator_kwargs = {
+        "type_checker": type_checker,
+    }
 
     # Add deprecation handling
     validators = dict(Draft4Validator.VALIDATORS)
@@ -553,7 +544,9 @@ def is_valid(self, instance, _schema=None, **__):
         )
         return next(errors, None) is None
 
-    cloudinitValidator.is_valid = is_valid
+    # this _could_ be an error, but it's not in this case
+    # https://github.com/python/mypy/issues/2427#issuecomment-1419206807
+    cloudinitValidator.is_valid = is_valid  # type: ignore [method-assign]
 
     return (cloudinitValidator, FormatChecker)
 
@@ -1918,7 +1911,7 @@ def handle_schema_args(name, args):
                 if args.config_file:
                     cfg = cfg_part.config_path
                 else:
-                    cfg = cfg_part.config_type
+                    cfg = str(cfg_part.config_type)
                 print(f"{nested_output_prefix}Valid schema {cfg!s}")
     if error_types:
         error(

cloudinit/mergers/__init__.py

@@ -75,7 +75,7 @@ def _handle_unknown(self, meth_wanted, value, merge_with):
 
 
 def dict_extract_mergers(config):
-    parsed_mergers = []
+    parsed_mergers: list = []
     raw_mergers = config.pop("merge_how", None)
     if raw_mergers is None:
         raw_mergers = config.pop("merge_type", None)
@@ -145,7 +145,7 @@ def construct(parsed_mergers):
             mod_attr = getattr(mod, MERGER_ATTR)
             mergers_to_be.append((mod_attr, m_ops))
     # Now form them...
-    mergers = []
+    mergers: list = []
     root = LookupMerger(mergers)
     for attr, opts in mergers_to_be:
         mergers.append(attr(root, opts))

cloudinit/net/__init__.py

@@ -764,7 +764,7 @@ def find_entry(mac, driver, device_id):
     for mac, new_name, driver, device_id in renames:
         if mac:
             mac = mac.lower()
-        cur_ops = []
+        cur_ops: list = []
         cur = find_entry(mac, driver, device_id)
         if not cur:
             if strict_present:
@@ -808,7 +808,7 @@ def find_entry(mac, driver, device_id):
                 else:
                     cur_ops.append(("down", mac, new_name, (new_name,)))
 
-            tmp_name = None
+            tmp_name: Optional[str] = None
             while tmp_name is None or tmp_name in cur_byname:
                 tmpi += 1
                 tmp_name = tmpname_fmt % tmpi
@@ -824,7 +824,7 @@ def find_entry(mac, driver, device_id):
         cur_byname = update_byname(cur_info)
         ops += cur_ops
 
-    opmap = {
+    opmap: Dict[str, Callable] = {
         "rename": Iproute2.link_rename,
         "down": Iproute2.link_down,
         "up": Iproute2.link_up,
@@ -844,7 +844,7 @@ def find_entry(mac, driver, device_id):
 
         for op, mac, new_name, params in ops + ups:
             try:
-                opmap.get(op)(*params)
+                opmap[op](*params)
             except Exception as e:
                 errors.append(
                     "[unknown] Error performing %s%s for %s, %s: %s"
@@ -1133,7 +1133,7 @@ def filter_hyperv_vf_with_synthetic_interface(
 def get_ib_hwaddrs_by_interface():
     """Build a dictionary mapping Infiniband interface names to their hardware
     address."""
-    ret = {}
+    ret: Dict[str, str] = {}
     for name, _, _, _ in get_interfaces():
         ib_mac = get_ib_interface_hwaddr(name, False)
         if ib_mac:

cloudinit/net/dhcp.py

@@ -125,7 +125,7 @@ def networkd_load_leases(leases_d=None):
     if leases_d is None:
         leases_d = NETWORKD_LEASES_DIR
 
-    ret = {}
+    ret: Dict[str, dict] = {}
     if not os.path.isdir(leases_d):
         return ret
     for lfile in os.listdir(leases_d):
@@ -654,7 +654,7 @@ def dhcp_discovery(
             if is_ib_interface(interface):
                 infiniband_argument = ["--clientid"]
             command = [
-                self.dhcp_client_path,  # pyright: ignore
+                self.client_name,
                 "--ipv4only",  # only attempt configuring ipv4
                 "--waitip",  # wait for ipv4 to be configured
                 "--persistent",  # don't deconfigure when dhcpcd exits
@@ -860,7 +860,7 @@ def get_newest_lease(self, interface: str) -> Dict[str, Any]:
             return self.parse_dhcpcd_lease(
                 subp.subp(
                     [
-                        self.dhcp_client_path,
+                        self.client_name,
                         "--dumplease",
                         "--ipv4only",
                         interface,
@@ -914,7 +914,7 @@ class Udhcpc(DhcpClient):
 
     def __init__(self):
         super().__init__()
-        self.lease_file = None
+        self.lease_file = ""
 
     def dhcp_discovery(
         self,
@@ -944,7 +944,7 @@ def dhcp_discovery(
         util.write_file(udhcpc_script, UDHCPC_SCRIPT, 0o755)
 
         cmd = [
-            self.dhcp_client_path,
+            self.client_name,
             "-O",
             "staticroutes",
             "-i",

cloudinit/net/netplan.py

@@ -387,7 +387,6 @@ def _net_setup_link(self, run=False):
         # net_setup_link on a device that no longer exists. When this happens,
         # we don't know what the device was renamed to, so re-gather the
         # entire list of devices and try again.
-        last_exception = Exception
         for _ in range(5):
             try:
                 for iface in get_devicelist():

cloudinit/netinfo.py

@@ -13,7 +13,7 @@
 import re
 from copy import copy, deepcopy
 from ipaddress import IPv4Network
-from typing import Dict, List, Union
+from typing import Dict, List, TypedDict
 
 from cloudinit import lifecycle, subp, util
 from cloudinit.net.network_state import net_prefix_to_ipv4_mask
@@ -41,6 +41,13 @@
 DEFAULT_NETDEV_INFO = {"ipv4": [], "ipv6": [], "hwaddr": "", "up": False}
 
 
+class Interface(TypedDict):
+    up: bool
+    hwaddr: str
+    ipv4: List[dict]
+    ipv6: List[dict]
+
+
 def _netdev_info_iproute_json(ipaddr_json):
     """Get network device dicts from ip route and ip link info.
 
@@ -285,7 +292,7 @@ def _netdev_info_ifconfig(ifconfig_data):
 
 def netdev_info(
     empty="",
-) -> Dict[str, Dict[str, Union[str, List[Dict[str, str]]]]]:
+) -> Dict[str, Dict[str, Interface]]:
     """return the instance's interfaces and interface data
 
     includes, interface name, link state, hardware address, and lists of ipv4

cloudinit/reporting/handlers.py

@@ -98,7 +98,7 @@ def __init__(
         self.ssl_details = util.fetch_ssl_details()
 
         self.flush_requested = Event()
-        self.queue = queue.Queue()
+        self.queue: queue.Queue = queue.Queue()
         self.event_processor = threading.Thread(target=self.process_requests)
         self.event_processor.daemon = True
         self.event_processor.start()
@@ -204,7 +204,7 @@ def __init__(self, kvp_file_path=KVP_POOL_FILE_GUEST, event_types=None):
         )
 
         self._event_types = event_types
-        self.q = queue.Queue()
+        self.q: queue.Queue = queue.Queue()
         self.incarnation_no = self._get_incarnation_no()
         self.event_key_prefix = "{0}|{1}".format(
             self.EVENT_PREFIX, self.incarnation_no

cloudinit/sources/__init__.py

@@ -203,7 +203,7 @@ class DataSource(CloudInitPickleMixin, metaclass=abc.ABCMeta):
     dsname = "_undef"
 
     # Cached cloud_name as determined by _get_cloud_name
-    _cloud_name = None
+    _cloud_name: Optional[str] = None
 
     # Cached cloud platform api type: e.g. ec2, openstack, kvm, lxd, azure etc.
     _platform_type = None
@@ -332,7 +332,7 @@ def __init__(self, sys_cfg, distro: Distro, paths: Paths, ud_proc=None):
         self.vendordata2 = None
         self.vendordata_raw = None
         self.vendordata2_raw = None
-        self.metadata_address = None
+        self.metadata_address: Optional[str] = None
         self.network_json: Optional[str] = UNSET
         self.ec2_metadata = UNSET
 

cloudinit/stages.py

@@ -145,7 +145,7 @@ def __init__(self, ds_deps: Optional[List[str]] = None, reporter=None):
         # Changed only when a fetch occurs
         self.datasource: Optional[sources.DataSource] = None
         self.ds_restored = False
-        self._previous_iid = None
+        self._previous_iid: Optional[str] = None
 
         if reporter is None:
             reporter = events.ReportEventStack(

pyproject.toml

@@ -57,7 +57,6 @@ module = [
     "cloudinit.config.cc_rsyslog",
     "cloudinit.config.cc_ubuntu_pro",
     "cloudinit.config.modules",
-    "cloudinit.config.schema",
     "cloudinit.distros",
     "cloudinit.distros.alpine",
     "cloudinit.distros.azurelinux",
@@ -69,21 +68,16 @@ module = [
     "cloudinit.distros.ubuntu",
     "cloudinit.distros.ug_util",
     "cloudinit.helpers",
-    "cloudinit.mergers",
-    "cloudinit.net",
     "cloudinit.net.cmdline",
-    "cloudinit.net.dhcp",
     "cloudinit.net.eni",
     "cloudinit.net.ephemeral",
     "cloudinit.net.freebsd",
     "cloudinit.net.netbsd",
-    "cloudinit.net.netplan",
     "cloudinit.net.network_manager",
     "cloudinit.net.network_state",
     "cloudinit.net.networkd",
     "cloudinit.net.sysconfig",
     "cloudinit.netinfo",
-    "cloudinit.reporting.handlers",
     "cloudinit.sources.DataSourceAzure",
     "cloudinit.sources.DataSourceBigstep",
     "cloudinit.sources.DataSourceCloudSigma",

tests/unittests/net/test_dhcp.py

@@ -957,7 +957,7 @@ def test_udhcpc_discovery(
                 ),
                 mock.call(
                     [
-                        "/sbin/udhcpc",
+                        "udhcpc",
                         "-O",
                         "staticroutes",
                         "-i",
@@ -1026,7 +1026,7 @@ def test_udhcpc_discovery_ib(
                 ),
                 mock.call(
                     [
-                        "/sbin/udhcpc",
+                        "udhcpc",
                         "-O",
                         "staticroutes",
                         "-i",
@@ -1354,7 +1354,7 @@ def test_dhcpcd_discovery_ib(
                 ),
                 mock.call(
                     [
-                        "/sbin/dhcpcd",
+                        "dhcpcd",
                         "--ipv4only",
                         "--waitip",
                         "--persistent",
@@ -1400,7 +1400,7 @@ def test_dhcpcd_discovery_timeout(
                 ),
                 mock.call(
                     [
-                        "/sbin/dhcpcd",
+                        "dhcpcd",
                         "--ipv4only",
                         "--waitip",
                         "--persistent",