fix(cmd): enable mypy strict checking for cloudinit.cmd

cloudinit/cmd/devel/make_mime.py

@@ -32,11 +32,11 @@ def create_mime_message(files):
         )
         content_type = sub_message.get_content_type().lower()
         if content_type not in get_content_types():
-            msg = ("content type %r for attachment %s may be incorrect!") % (
-                content_type,
-                i + 1,
+            err_msg = (
+                f"content type {content_type!r} for attachment"
+                f" {i + 1} may be incorrect!"
             )
-            errors.append(msg)
+            errors.append(err_msg)
         sub_messages.append(sub_message)
     combined_message = MIMEMultipart()
     for msg in sub_messages:

cloudinit/cmd/devel/net_convert.py

@@ -18,6 +18,7 @@
     network_manager,
     network_state,
     networkd,
+    renderer,
     sysconfig,
 )
 
@@ -158,13 +159,14 @@ def handle_args(name, args):
             apply_network_config_for_secondary_ips=True,
         )
     elif args.kind == "vmware-imc":
-        config = guestcust_util.Config(
+        vmware_config = guestcust_util.Config(
             guestcust_util.ConfigFile(args.network_data.name)
         )
         pre_ns = guestcust_util.get_network_data_from_vmware_cust_cfg(
-            config, False
+            vmware_config, False
         )
 
+    r_cls: type[renderer.Renderer]
     distro_cls = distros.fetch(args.distro)
     distro = distro_cls(args.distro, {}, None)
     if args.output_kind == "eni":

cloudinit/cmd/main.py

@@ -19,7 +19,7 @@
 import traceback
 import logging
 import yaml
-from typing import Optional, Tuple, Callable, Union
+from typing import Any, Optional, Tuple, Callable, Union
 
 from cloudinit import features, netinfo
 from cloudinit import signal_handler
@@ -98,18 +98,20 @@ def error(self, message):
         if not self._raw_args:
             self._raw_args = sys.argv[1:]
         subcommand = None
+        subparsers_action = (
+            self._subparsers._group_actions[0] if self._subparsers else None
+        )
+        choices = getattr(subparsers_action, "choices", None) or {}
         if self._raw_args:
             for arg in self._raw_args:
-                if arg in self._subparsers._group_actions[0].choices:
+                if arg in choices:
                     subcommand = arg
                     break
         # Check if the subcommand exists and show its help
 
         if subcommand:
-            subparser = self._subparsers._group_actions[0].choices[subcommand]
-            subparser.print_help(
-                file=sys.stderr
-            )  # Print subcommand help to stderr
+            subparser = choices[subcommand]
+            subparser.print_help(file=sys.stderr)
         else:
             self.print_help(file=sys.stderr)
         sys.exit(2)
@@ -546,6 +548,13 @@ def main_init(name, args):
     bring_up_interfaces = _should_bring_up_interfaces(init, args)
     try:
         init.fetch(existing=existing)
+        if init.datasource is None:
+            LOG.debug(
+                "[%s] Exiting. datasource is None after fetch,"
+                " cannot continue.",
+                mode,
+            )
+            return (None, [])
         # if in network mode, and the datasource is local
         # then work was done at that stage.
         if mode == sources.DSMODE_NETWORK and init.datasource.dsmode != mode:
@@ -613,6 +622,13 @@ def main_init(name, args):
                 )
                 util.write_file(init.paths.get_runpath(".skip-network"), "")
 
+        if init.datasource is None:
+            LOG.debug(
+                "[%s] Exiting. datasource is None in local mode,"
+                " cannot check dsmode.",
+                mode,
+            )
+            return (None, [])
         if init.datasource.dsmode != mode:
             LOG.debug(
                 "[%s] Exiting. datasource %s not in local mode.",
@@ -912,13 +928,13 @@ def status_wrapper(name, args):
             "Invalid cloud init mode specified '{0}'".format(mode)
         )
 
-    nullstatus = {
+    nullstatus: dict[str, Union[list[Any], dict[str, Any], None]] = {
         "errors": [],
         "recoverable_errors": {},
         "start": None,
         "finished": None,
     }
-    status = {
+    status: dict[str, Any] = {
         "v1": {
             "datasource": None,
             "init": nullstatus.copy(),
@@ -951,10 +967,15 @@ def status_wrapper(name, args):
 
     v1[mode]["start"] = float(util.uptime())
     handler = next(
-        filter(
-            lambda h: isinstance(h, loggers.LogExporter), root_logger.handlers
-        )
+        (
+            h
+            for h in root_logger.handlers
+            if isinstance(h, loggers.LogExporter)
+        ),
+        None,
     )
+    if not isinstance(handler, loggers.LogExporter):
+        raise RuntimeError("LogExporter handler not found in root logger")
     preexisting_recoverable_errors = handler.export_logs()
 
     # Write status.json prior to running init / module code
@@ -1052,7 +1073,7 @@ def _maybe_set_hostname(init, stage, retry_stage):
     )
     if hostname:  # meta-data or user-data hostname content
         try:
-            cc_set_hostname.handle("set_hostname", init.cfg, cloud, None)
+            cc_set_hostname.handle("set_hostname", init.cfg, cloud, [])
         except cc_set_hostname.SetHostnameError as e:
             LOG.debug(
                 "Failed setting hostname in %s stage. Will"

cloudinit/net/eni.py

@@ -7,7 +7,7 @@
 import os
 import re
 from contextlib import suppress
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Mapping, Optional
 
 from cloudinit import performance, subp, util
 from cloudinit.net import (
@@ -453,7 +453,7 @@ def has_same_ip_version(addr_or_net: str, is_ipv6: bool) -> bool:
 class Renderer(renderer.Renderer):
     """Renders network information in a /etc/network/interfaces format."""
 
-    def __init__(self, config: Optional[dict] = None):
+    def __init__(self, config: Optional[Mapping[str, Any]] = None):
         if not config:
             config = {}
         self.eni_path = config.get("eni_path", "etc/network/interfaces")

cloudinit/stages.py

@@ -558,10 +558,8 @@ def is_new_instance(self):
             or previous != self.ds.get_instance_id()
         )
 
-    def fetch(self, existing="check"):
-        """optionally load datasource from cache, otherwise discover
-        datasource
-        """
+    def fetch(self, existing="check") -> sources.DataSource:
+        """Load datasource from cache, otherwise discover datasource"""
         return self._get_data_source(existing=existing)
 
     def instancify(self):

pyproject.toml

@@ -42,9 +42,6 @@ no_implicit_optional = true
 # See GH-5445
 [[tool.mypy.overrides]]
 module = [
-    "cloudinit.cmd.devel.make_mime",
-    "cloudinit.cmd.devel.net_convert",
-    "cloudinit.cmd.main",
     "cloudinit.config.cc_apt_configure",
     "cloudinit.config.cc_ca_certs",
     "cloudinit.config.cc_growpart",