fix: typing for rsyslog, ubuntu_pro, power_state_change (canonical#5985)

Author: MostafaTarek124eru

cloudinit/config/cc_power_state_change.py

@@ -45,7 +45,10 @@ def givecmdline(pid):
             (output, _err) = subp.subp(["procstat", "-c", str(pid)])
             line = output.splitlines()[1]
             m = re.search(r"\d+ (\w|\.|-)+\s+(/\w.+)", line)
-            return m.group(2)
+            if m:
+                return m.group(2)
+            else:
+                return None
         else:
             return util.load_text_file("/proc/%s/cmdline" % pid)
     except IOError:

cloudinit/config/cc_rsyslog.py

@@ -246,10 +246,7 @@ def __init__(
         self.proto = proto
 
         self.addr = addr
-        if port:
-            self.port = int(port)
-        else:
-            self.port = None
+        self.port = int(port) if port is not None else None
 
     def validate(self):
         if self.port:

cloudinit/config/cc_ubuntu_pro.py

@@ -178,8 +178,8 @@ def configure_pro(token, enable=None):
         # Allow `ua attach` to fail in already attached machines
         subp.subp(attach_cmd, rcs={0, 2}, logstring=redacted_cmd)
     except subp.ProcessExecutionError as e:
-        err = str(e).replace(token, REDACTED)
-        msg = f"Failure attaching Ubuntu Pro:\n{err}"
+        er = str(e).replace(token, REDACTED)
+        msg = f"Failure attaching Ubuntu Pro:\n{er}"
         util.logexc(LOG, msg)
         raise RuntimeError(msg) from e
 

pyproject.toml

@@ -53,9 +53,6 @@ module = [
     "cloudinit.config.cc_ca_certs",
     "cloudinit.config.cc_growpart",
     "cloudinit.config.cc_ntp",
-    "cloudinit.config.cc_power_state_change",
-    "cloudinit.config.cc_rsyslog",
-    "cloudinit.config.cc_ubuntu_pro",
     "cloudinit.config.modules",
     "cloudinit.distros",
     "cloudinit.distros.alpine",
@@ -134,19 +131,16 @@ module = [
     "tests.unittests.config.test_cc_mcollective",
     "tests.unittests.config.test_cc_mounts",
     "tests.unittests.config.test_cc_phone_home",
-    "tests.unittests.config.test_cc_power_state_change",
     "tests.unittests.config.test_cc_puppet",
     "tests.unittests.config.test_cc_resizefs",
     "tests.unittests.config.test_cc_resolv_conf",
     "tests.unittests.config.test_cc_rh_subscription",
-    "tests.unittests.config.test_cc_rsyslog",
     "tests.unittests.config.test_cc_runcmd",
     "tests.unittests.config.test_cc_snap",
     "tests.unittests.config.test_cc_ssh",
     "tests.unittests.config.test_cc_timezone",
     "tests.unittests.config.test_cc_ubuntu_autoinstall",
     "tests.unittests.config.test_cc_ubuntu_drivers",
-    "tests.unittests.config.test_cc_ubuntu_pro",
     "tests.unittests.config.test_cc_update_etc_hosts",
     "tests.unittests.config.test_cc_users_groups",
     "tests.unittests.config.test_cc_wireguard",

tests/unittests/config/test_cc_power_state_change.py

@@ -47,7 +47,7 @@ def test_empty_mode(self):
         self.assertRaises(TypeError, psc.load_power_state, cfg, self.dist)
 
     def test_valid_modes(self):
-        cfg = {"power_state": {}}
+        cfg: dict = {"power_state": {}}
         for mode in ("halt", "poweroff", "reboot"):
             cfg["power_state"]["mode"] = mode
             check_lps_ret(psc.load_power_state(cfg, self.dist), mode=mode)

tests/unittests/config/test_cc_rsyslog.py

@@ -340,7 +340,7 @@ def test_install_rsyslog_on_freebsd(self, m_which):
             with mock.patch.object(
                 cloud.distro, "install_packages"
             ) as m_install:
-                handle("rsyslog", {"rsyslog": config}, cloud, None)
+                handle("rsyslog", {"rsyslog": config}, cloud, [])
             m_which.assert_called_with(config["check_exe"])
             m_install.assert_called_with(config["packages"])
 
@@ -356,6 +356,6 @@ def test_no_install_rsyslog_with_check_exe(self, m_which, m_isbsd):
         m_isbsd.return_value = False
         m_which.return_value = "/usr/sbin/rsyslogd"
         with mock.patch.object(cloud.distro, "install_packages") as m_install:
-            handle("rsyslog", {"rsyslog": config}, cloud, None)
+            handle("rsyslog", {"rsyslog": config}, cloud, [])
         m_which.assert_called_with(config["check_exe"])
         m_install.assert_not_called()

tests/unittests/config/test_cc_ubuntu_pro.py

@@ -57,19 +57,24 @@ def fake_uaclient(mocker):
     m_uaclient = mock.Mock()
 
     sys.modules["uaclient"] = m_uaclient
+    mock_exceptions_module = mock.Mock()
 
     # Exceptions
-    _exceptions = namedtuple(
-        "exceptions",
+    Exceptions = namedtuple(
+        "Exceptions",
         [
             "UserFacingError",
             "AlreadyAttachedError",
         ],
-    )(
+    )
+    mock_exceptions_module.UserFacingError = FakeUserFacingError
+    mock_exceptions_module.AlreadyAttachedError = FakeAlreadyAttachedError
+    sys.modules["uaclient.api.exceptions"] = mock_exceptions_module
+    _exceptions = Exceptions(
         FakeUserFacingError,
         FakeAlreadyAttachedError,
     )
-    sys.modules["uaclient.api.exceptions"] = _exceptions
+    return _exceptions
 
 
 @pytest.mark.usefixtures("fake_uaclient")
@@ -834,7 +839,7 @@ def test_handle_attach(
         caplog,
     ):
         """Non-Pro schemas and instance."""
-        handle("nomatter", cfg=cfg, cloud=cloud, args=None)
+        handle("nomatter", cfg=cfg, cloud=cloud, args=[])
         for record_tuple in log_record_tuples:
             assert record_tuple in caplog.record_tuples
         if maybe_install_call_args_list is not None:
@@ -961,7 +966,7 @@ def test_handle_auto_attach_vs_attach(
             m_auto_attach.side_effect = auto_attach_side_effect
 
         with expectation:
-            handle("nomatter", cfg=cfg, cloud=cloud, args=None)
+            handle("nomatter", cfg=cfg, cloud=cloud, args=[])
 
         for record_tuple in log_record_tuples:
             assert record_tuple in caplog.record_tuples
@@ -1006,7 +1011,7 @@ def test_no_fallback_attach(
         enable or disable pro auto-attach.
         """
         m_should_auto_attach.return_value = is_pro
-        handle("nomatter", cfg=cfg, cloud=self.cloud, args=None)
+        handle("nomatter", cfg=cfg, cloud=self.cloud, args=[])
         assert not m_attach.call_args_list
 
     @pytest.mark.parametrize(
@@ -1061,7 +1066,7 @@ def test_handle_errors(self, cfg, match):
                 "nomatter",
                 cfg=cfg,
                 cloud=self.cloud,
-                args=None,
+                args=[],
             )
 
     @mock.patch(f"{MPATH}.subp.subp")
@@ -1087,7 +1092,7 @@ def test_pro_config_error_invalid_url(self, m_subp, caplog):
                 "nomatter",
                 cfg=cfg,
                 cloud=self.cloud,
-                args=None,
+                args=[],
             )
         assert not caplog.text
 
@@ -1107,7 +1112,7 @@ def test_fallback_to_attach_no_token(
                 "nomatter",
                 cfg=cfg,
                 cloud=self.cloud,
-                args=None,
+                args=[],
             )
         assert [] == m_subp.call_args_list
         assert (