chore(mypy): Fix failures on newer versions of mypy (canonical#5607)

cloudinit/config/schema.py

@@ -46,9 +46,7 @@
 )
 
 try:
-    from jsonschema import ValidationError as _ValidationError
-
-    ValidationError = _ValidationError
+    from jsonschema import ValidationError
 except ImportError:
     ValidationError = Exception  # type: ignore
 

cloudinit/distros/aosc.py

@@ -107,7 +107,7 @@ def package_command(self, command, args=None, pkgs=None):
     def install_packages(self, pkglist: PackageList):
         self.package_command("install", pkgs=pkglist)
 
-    def update_package_sources(self):
+    def update_package_sources(self, *, force=False):
         self._runner.run(
             "update-sources",
             self.package_command,

cloudinit/net/__init__.py

@@ -555,10 +555,8 @@ def find_fallback_nic_on_linux() -> Optional[str]:
     return None
 
 
-def generate_fallback_config(config_driver=None):
+def generate_fallback_config(config_driver=None) -> Optional[dict]:
     """Generate network cfg v2 for dhcp on the NIC most likely connected."""
-    if not config_driver:
-        config_driver = False
 
     target_name = find_fallback_nic()
     if not target_name:
@@ -572,16 +570,16 @@ def generate_fallback_config(config_driver=None):
         match = {
             "macaddress": read_sys_net_safe(target_name, "address").lower()
         }
+    if config_driver:
+        driver = device_driver(target_name)
+        if driver:
+            match["driver"] = driver
     cfg = {
         "dhcp4": True,
         "dhcp6": True,
         "set-name": target_name,
         "match": match,
     }
-    if config_driver:
-        driver = device_driver(target_name)
-        if driver:
-            cfg["match"]["driver"] = driver
     nconf = {"ethernets": {target_name: cfg}, "version": 2}
     return nconf
 

cloudinit/net/netops/__init__.py

@@ -1,18 +1,22 @@
+from abc import abstractmethod
 from typing import Optional
 
 from cloudinit.subp import SubpResult
 
 
 class NetOps:
     @staticmethod
+    @abstractmethod
     def link_up(interface: str) -> SubpResult:
         pass
 
     @staticmethod
+    @abstractmethod
     def link_down(interface: str) -> SubpResult:
         pass
 
     @staticmethod
+    @abstractmethod
     def link_rename(current_name: str, new_name: str):
         pass
 
@@ -41,6 +45,7 @@ def del_route(
         pass
 
     @staticmethod
+    @abstractmethod
     def get_default_route() -> str:
         pass
 

cloudinit/net/openbsd.py

@@ -10,7 +10,7 @@
 
 
 class Renderer(cloudinit.net.bsd.BSDRenderer):
-    def write_config(self):
+    def write_config(self, target=None):
         for device_name, v in self.interface_configurations.items():
             if_file = "etc/hostname.{}".format(device_name)
             fn = subp.target_path(self.target, if_file)

cloudinit/safeyaml.py

@@ -162,8 +162,8 @@ def _get_nested_path_prefix(self, node):
                     return f"{mark.path}."
         return ""
 
-    def construct_mapping(self, node):
-        mapping = super().construct_mapping(node)
+    def construct_mapping(self, node, deep=False):
+        mapping = super().construct_mapping(node, deep=deep)
         nested_path_prefix = self._get_nested_path_prefix(node)
         for key_node, value_node in node.value:
             node_key_path = f"{nested_path_prefix}{key_node.value}"

cloudinit/url_helper.py

@@ -293,8 +293,7 @@ def __init__(self, response: requests.Response):
     @property
     def contents(self) -> bytes:
         if self._response.content is None:
-            # typeshed bug: https://github.com/python/typeshed/pull/12180
-            return b""  # type: ignore
+            return b""
         return self._response.content
 
     @property

pyproject.toml

@@ -31,6 +31,7 @@ module = [
     "paramiko.*",
     "pip.*",
     "pycloudlib.*",
+    "responses",
     "serial",
     "tests.integration_tests.user_settings",
     "uaclient.*",

tests/unittests/config/test_cc_write_files.py

@@ -138,13 +138,14 @@ def test_all_decodings(self):
             b64 = (base64.b64encode(data), b64_aliases)
             for content, aliases in (gz, gz_b64, b64):
                 for enc in aliases:
+                    path = "/tmp/file-%s-%s" % (name, enc)
                     cur = {
                         "content": content,
-                        "path": "/tmp/file-%s-%s" % (name, enc),
+                        "path": path,
                         "encoding": enc,
                     }
                     files.append(cur)
-                    expected.append((cur["path"], data))
+                    expected.append((path, data))
 
         write_files("test_decoding", files, self.owner)
 

tests/unittests/helpers.py

@@ -599,11 +599,16 @@ def skipIfAptPkg():
 
 
 try:
+    import importlib.metadata
+
     import jsonschema
 
     assert jsonschema  # avoid pyflakes error F401: import unused
     _jsonschema_version = tuple(
-        int(part) for part in jsonschema.__version__.split(".")  # type: ignore
+        int(part)
+        for part in importlib.metadata.metadata("jsonschema")
+        .get("Version", "")
+        .split(".")
     )
     _missing_jsonschema_dep = False
 except ImportError:

tests/unittests/test_net.py

@@ -3551,14 +3551,12 @@ def test_render(
         )
 
         if network_cfg is None:
-            network_cfg = net.generate_fallback_config()
+            parsed_cfg = net.generate_fallback_config()
         else:
-            network_cfg = yaml.safe_load(network_cfg)
-        assert isinstance(network_cfg, dict)
+            parsed_cfg = yaml.safe_load(network_cfg)
+        assert isinstance(parsed_cfg, dict)
 
-        ns = network_state.parse_net_config_data(
-            network_cfg, skip_broken=False
-        )
+        ns = network_state.parse_net_config_data(parsed_cfg, skip_broken=False)
 
         render_dir = os.path.join(tmp_dir, "render")
         os.makedirs(render_dir)