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

.github/workflows/23-pr-unit-distro.yml

@@ -42,7 +42,7 @@ jobs:
           lxc exec alpine -- ping -c 1 dl-cdn.alpinelinux.org || true
 
       - name: Install dependencies
-        run: lxc exec alpine -- apk add py3-tox git tzdata
+        run: lxc exec alpine -- apk add py3-tox py3-python-discovery git tzdata
 
       - name: Mount source into container directory
         run: lxc config device add alpine gitdir disk source=$(pwd) path=/root/cloud-init-ro

cloudinit/cmd/devel/make_mime.py

@@ -32,11 +32,13 @@ 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!") % (
+            err_msg = (
+                "content type %r for attachment %s may be incorrect!"
+            ) % (
                 content_type,
                 i + 1,
             )
-            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
@@ -83,6 +83,13 @@ class SubcommandAwareArgumentParser(argparse.ArgumentParser):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._raw_args = None
+        self._subparsers_action = None
+
+    def add_subparsers(self, **kwargs):
+        """Override to capture the subparsers action for use in error()."""
+        action = super().add_subparsers(**kwargs)
+        self._subparsers_action = action
+        return action
 
     def parse_args(self, args=None, namespace=None):
         """Override parse_args to store raw arguments for error handling."""
@@ -98,18 +105,20 @@ def error(self, message):
         if not self._raw_args:
             self._raw_args = sys.argv[1:]
         subcommand = None
-        if self._raw_args:
-            for arg in self._raw_args:
-                if arg in self._subparsers._group_actions[0].choices:
-                    subcommand = arg
-                    break
-        # Check if the subcommand exists and show its help
+        if not self._subparsers_action:
+            self.print_help(file=sys.stderr)
+            sys.exit(2)
+
+        choices = self._subparsers_action.choices
+        for arg in self._raw_args:
+            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
+            # Print subcommand help to stderr
+            choices[subcommand].print_help(file=sys.stderr)
         else:
             self.print_help(file=sys.stderr)
         sys.exit(2)
@@ -546,6 +555,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 +629,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 +935,13 @@ def status_wrapper(name, args):
             "Invalid cloud init mode specified '{0}'".format(mode)
         )
 
-    nullstatus = {
+    nullstatus: dict[str, list[Any] | dict[str, Any] | None] = {
         "errors": [],
         "recoverable_errors": {},
         "start": None,
         "finished": None,
     }
-    status = {
+    status: dict[str, Any] = {
         "v1": {
             "datasource": None,
             "init": nullstatus.copy(),
@@ -955,6 +978,8 @@ def status_wrapper(name, args):
             lambda h: isinstance(h, loggers.LogExporter), root_logger.handlers
         )
     )
+    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 +1077,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[Optional[Mapping[str, Any]]] = None):
         if not config:
             config = {}
         self.eni_path = config.get("eni_path", "etc/network/interfaces")

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",