chore(opennebula): address review feedback on type annotations

Add type annotations to parse_shell_config, get_field (with @overload),
and all OpenNebulaNetwork.get_* methods as requested. Replace assert-based
type narrowing with proper guards. Remove leaked global assignment in test
and revert unrelated cosmetic changes.

Refs GH-6810

Author: mcanevet

cloudinit/sources/DataSourceOpenNebula.py

@@ -20,7 +20,7 @@
 import re
 import shlex
 import textwrap
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, overload
 
 from cloudinit import atomic_helper, net, sources, subp, util
 
@@ -42,6 +42,7 @@
 
 
 class DataSourceOpenNebula(sources.DataSource):
+
     dsname = "OpenNebula"
 
     def __init__(self, sys_cfg, distro, paths):
@@ -94,11 +95,9 @@ def _get_data(self):
                 LOG.debug("found datasource in %s", cdev)
                 break
 
-        if not seed:
+        if not seed or results is None:
             return False
 
-        assert results is not None
-
         # merge fetched metadata with datasource defaults
         md = results["metadata"]
         md = util.mergemanydict([md, defaults])
@@ -119,7 +118,10 @@ def _get_data(self):
 
     def _get_subplatform(self):
         """Return the subplatform metadata source details."""
-        assert self.seed is not None
+        if self.seed is None:
+            raise RuntimeError(
+                "_get_subplatform called before datasource was read"
+            )
         if self.seed_dir in self.seed:
             subplatform_type = "seed-dir"
         else:
@@ -151,33 +153,40 @@ class BrokenContextDiskDir(Exception):
 
 
 class OpenNebulaNetwork:
-    def __init__(self, context, distro, system_nics_by_mac=None):
+    def __init__(
+        self,
+        context: Dict[str, str],
+        distro: Any,
+        system_nics_by_mac: Optional[Dict[str, str]] = None,
+    ) -> None:
         self.context = context
         if system_nics_by_mac is None:
             system_nics_by_mac = get_physical_nics_by_mac(distro)
-        self.ifaces = collections.OrderedDict(
-            [
-                k
-                for k in sorted(
-                    system_nics_by_mac.items(),
-                    key=lambda k: net.natural_sort_key(k[1]),
-                )
-            ]
+        self.ifaces: collections.OrderedDict[str, str] = (
+            collections.OrderedDict(
+                [
+                    k
+                    for k in sorted(
+                        system_nics_by_mac.items(),
+                        key=lambda k: net.natural_sort_key(k[1]),
+                    )
+                ]
+            )
         )
 
         # OpenNebula 4.14+ provide macaddr for ETHX in variable ETH_MAC.
         # context_devname provides {mac.lower():ETHX, mac2.lower():ETHX}
-        self.context_devname = {}
+        self.context_devname: Dict[str, str] = {}
         for k, v in context.items():
             m = re.match(r"^(.+)_MAC$", k)
             if m:
                 self.context_devname[v.lower()] = m.group(1)
 
-    def mac2ip(self, mac):
+    def mac2ip(self, mac: str) -> str:
         return ".".join([str(int(c, 16)) for c in mac.split(":")[2:]])
 
-    def get_nameservers(self, dev):
-        nameservers = {}
+    def get_nameservers(self, dev: str) -> Dict[str, List[str]]:
+        nameservers: Dict[str, List[str]] = {}
         dns = self.get_field(dev, "dns", "").split()
         dns.extend(self.context.get("DNS", "").split())
         if dns:
@@ -187,14 +196,14 @@ def get_nameservers(self, dev):
             nameservers["search"] = search_domain
         return nameservers
 
-    def get_mtu(self, dev):
+    def get_mtu(self, dev: str) -> Optional[str]:
         return self.get_field(dev, "mtu")
 
-    def get_ip(self, dev, mac):
+    def get_ip(self, dev: str, mac: str) -> str:
         return self.get_field(dev, "ip", self.mac2ip(mac))
 
-    def get_ip6(self, dev):
-        addresses6 = []
+    def get_ip6(self, dev: str) -> List[str]:
+        addresses6: List[str] = []
         ip6 = self.get_field(dev, "ip6")
         if ip6:
             addresses6.append(ip6)
@@ -203,24 +212,35 @@ def get_ip6(self, dev):
             addresses6.append(ip6_ula)
         return addresses6
 
-    def get_ip6_prefix(self, dev):
+    def get_ip6_prefix(self, dev: str) -> str:
         return self.get_field(dev, "ip6_prefix_length", "64")
 
-    def get_gateway(self, dev):
+    def get_gateway(self, dev: str) -> Optional[str]:
         return self.get_field(dev, "gateway")
 
-    def get_gateway6(self, dev):
+    def get_gateway6(self, dev: str) -> Optional[str]:
         # OpenNebula 6.1.80 introduced new context parameter ETHx_IP6_GATEWAY
         # to replace old ETHx_GATEWAY6. Old ETHx_GATEWAY6 will be removed in
         # OpenNebula 6.4.0 (https://github.com/OpenNebula/one/issues/5536).
-        return self.get_field(
-            dev, "ip6_gateway", self.get_field(dev, "gateway6")
-        )
+        ip6_gateway = self.get_field(dev, "ip6_gateway")
+        if ip6_gateway is not None:
+            return ip6_gateway
+        return self.get_field(dev, "gateway6")
 
-    def get_mask(self, dev):
+    def get_mask(self, dev: str) -> str:
         return self.get_field(dev, "mask", "255.255.255.0")
 
-    def get_field(self, dev, name, default=None):
+    @overload
+    def get_field(self, dev: str, name: str) -> Optional[str]: ...
+    @overload
+    def get_field(
+        self, dev: str, name: str, default: None
+    ) -> Optional[str]: ...
+    @overload
+    def get_field(self, dev: str, name: str, default: str) -> str: ...
+    def get_field(
+        self, dev: str, name: str, default: Optional[str] = None
+    ) -> Optional[str]:
         """return the field name in context for device dev.
 
         context stores <dev>_<NAME> (example: eth0_DOMAIN).
@@ -324,7 +344,9 @@ def varprinter():
     )
 
 
-def parse_shell_config(content, asuser=None):
+def parse_shell_config(
+    content: str, asuser: Optional[str] = None
+) -> Dict[str, str]:
     """run content and return environment variables which changed
 
     WARNING: the special variable _start_ is used to delimit content
@@ -398,10 +420,14 @@ def parse_shell_config(content, asuser=None):
 def read_context_disk_dir(
     source_dir: str, distro: Any, asuser: Optional[str] = None
 ) -> Dict[str, Any]:
-    """
-    read_context_disk_dir(source_dir):
-    read source_dir and return a tuple with metadata dict and user-data
-    string populated.  If not a valid dir, raise a NonContextDiskDir
+    """Read ``source_dir`` and return a dictionary containing context data.
+
+    The returned dictionary always includes ``"metadata"`` and
+    ``"userdata"`` keys, and may also include ``"network-interfaces"``
+    when network configuration can be generated from the context.
+
+    If ``source_dir`` is not a valid context directory, raise
+    ``NonContextDiskDir``.
     """
     found: Dict[str, str] = {}
     for af in CONTEXT_DISK_FILES:
@@ -493,7 +519,7 @@ def read_context_disk_dir(
     return results
 
 
-def get_physical_nics_by_mac(distro):
+def get_physical_nics_by_mac(distro: Any) -> Dict[str, str]:
     devs = net.get_interfaces_by_mac()
     return dict(
         [(m, n) for m, n in devs.items() if distro.networking.is_physical(n)]

tests/unittests/sources/test_opennebula.py

@@ -365,12 +365,14 @@ def my_devs_with(criteria):
             }.get(criteria, [])
 
         m_find_devs_with.side_effect = my_devs_with
+        # Signature of my_devs_with doesn't match util.find_devs_with
         util.find_devs_with = my_devs_with  # type: ignore[assignment]
         assert ["/dev/sdb", "/dev/sr0", "/dev/vdb"] == ds.find_candidate_devs()
 
 
 @mock.patch(DS_PATH + ".net.get_interfaces_by_mac", mock.Mock(return_value={}))
 class TestOpenNebulaNetwork:
+
     system_nics = ("eth0", "ens3")
 
     def test_context_devname(self):
@@ -575,8 +577,9 @@ def test_get_field_nonecontext(self):
         Verify get_field('device', 'name') returns None if context value is
         None.
         """
+        # Deliberately passing None value to test defensive handling in get_field
         context = {"ETH9_DUMMY": None}
-        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        net = ds.OpenNebulaNetwork(context, mock.Mock())  # type: ignore[arg-type]
         val = net.get_field("eth9", "dummy")
         assert None is val
 
@@ -1015,8 +1018,8 @@ class TestGetPhysicalNicsByMac:
     )
     def test(self, interfaces_by_mac, physical_devs, expected_return):
         distro = mock.Mock()
-        distro.networking.is_physical.side_effect = lambda devname: (
-            devname in physical_devs
+        distro.networking.is_physical.side_effect = (
+            lambda devname: devname in physical_devs
         )
         with mock.patch(
             DS_PATH + ".net.get_interfaces_by_mac",