feat(opennebula): support ETHx_METHOD and ETHx_IP6_METHOD

cloudinit/sources/DataSourceOpenNebula.py

@@ -252,6 +252,23 @@ def get_routes(self, dev: str) -> List[Dict[str, str]]:
                 )
         return routes
 
+    def get_method(self, dev: str) -> str:
+        """Return IPv4 config method: static | dhcp | skip."""
+        return self.get_field(dev, "method", "static").lower()
+
+    def get_ip6_method(self, dev: str) -> str:
+        """Return IPv6 config method: static | dhcp | auto | disable | skip.
+
+        Defaults to 'static' when ETHx_IP6 is set, otherwise 'disable'.
+        """
+        val = self.get_field(dev, "ip6_method", "")
+        if val:
+            return val.lower()
+        # infer from presence of IPv6 address or gateway
+        if self.get_ip6(dev) or self.get_gateway6(dev):
+            return "static"
+        return "disable"
+
     @overload
     def get_field(self, dev: str, name: str) -> Optional[str]: ...
     @overload
@@ -294,29 +311,36 @@ def gen_conf(self) -> Dict[str, Any]:
             # Set MAC address
             devconf["match"] = {"macaddress": mac}
 
-            # Set IPv4 address
-            devconf["addresses"] = []
-            mask = self.get_mask(c_dev)
-            prefix = str(net.ipv4_mask_to_net_prefix(mask))
-            devconf["addresses"].append(self.get_ip(c_dev, mac) + "/" + prefix)
-
-            # Set IPv6 Global and ULA address
-            addresses6 = self.get_ip6(c_dev)
-            if addresses6:
-                prefix6 = self.get_ip6_prefix(c_dev)
-                devconf["addresses"].extend(
-                    [i + "/" + prefix6 for i in addresses6]
-                )
-
-            # Set IPv4 default gateway
-            gateway = self.get_gateway(c_dev)
-            if gateway:
-                devconf["gateway4"] = gateway
-
-            # Set IPv6 default gateway
-            gateway6 = self.get_gateway6(c_dev)
-            if gateway6:
-                devconf["gateway6"] = gateway6
+            # Set IPv4 address / method
+            method: str = self.get_method(c_dev)
+            if method == "skip":
+                continue
+            elif method == "dhcp":
+                devconf["dhcp4"] = True
+            else:  # static (default)
+                mask = self.get_mask(c_dev)
+                prefix = str(net.ipv4_mask_to_net_prefix(mask))
+                devconf["addresses"] = [self.get_ip(c_dev, mac) + "/" + prefix]
+                gateway = self.get_gateway(c_dev)
+                if gateway:
+                    devconf["gateway4"] = gateway
+
+            # Set IPv6 address / method
+            ip6_method: str = self.get_ip6_method(c_dev)
+            if ip6_method in ("dhcp", "dhcp6"):
+                devconf["dhcp6"] = True
+            elif ip6_method in ("auto", "slaac"):
+                devconf["accept-ra"] = True
+            elif ip6_method not in ("disable", "skip"):  # static
+                addresses6 = self.get_ip6(c_dev)
+                if addresses6:
+                    prefix6 = self.get_ip6_prefix(c_dev)
+                    devconf.setdefault("addresses", []).extend(
+                        [i + "/" + prefix6 for i in addresses6]
+                    )
+                gateway6 = self.get_gateway6(c_dev)
+                if gateway6:
+                    devconf["gateway6"] = gateway6
 
             # Set DNS servers and search domains
             nameservers = self.get_nameservers(c_dev)

doc/rtd/reference/datasources/opennebula.rst

@@ -68,14 +68,28 @@ the OpenNebula documentation.
     ETH<x>_DNS
     ETH<x>_SEARCH_DOMAIN
     ETH<x>_MTU
+    ETH<x>_METHOD
     ETH<x>_IP6
     ETH<x>_IP6_ULA
     ETH<x>_IP6_PREFIX_LENGTH
     ETH<x>_IP6_GATEWAY
+    ETH<x>_IP6_METHOD
     ETH<x>_ROUTES
 
 Static `network configuration`_.
 
+``ETH<x>_METHOD`` controls the IPv4 configuration method for interface
+``ETH<x>``. Accepted values are ``static`` (default), ``dhcp``, and
+``skip``. When set to ``dhcp``, ``dhcp4: true`` is emitted and no static
+address is configured. When set to ``skip``, the interface is omitted
+from the Netplan output entirely.
+
+``ETH<x>_IP6_METHOD`` controls the IPv6 configuration method. Accepted
+values are ``static`` (default when ``ETH<x>_IP6`` or ``ETH<x>_IP6_GATEWAY``
+is present), ``dhcp`` / ``dhcp6``, ``auto`` / ``slaac``, ``disable``, and
+``skip``. ``dhcp``/``dhcp6`` emits ``dhcp6: true``; ``auto``/``slaac``
+emits ``accept-ra: true``; ``disable`` suppresses all IPv6 configuration.
+
 ``ETH<x>_ROUTES`` is a comma-separated list of static routes in the form
 ``NETWORK via GATEWAY``. For example::
 

tests/unittests/sources/test_opennebula.py

@@ -975,6 +975,122 @@ def test_multiple_nics(self):
 
         assert expected == net.gen_conf()
 
+    # --- ETHx_METHOD / ETHx_IP6_METHOD tests ---
+
+    def test_get_method_default(self):
+        """METHOD absent → default 'static'."""
+        net = ds.OpenNebulaNetwork({}, mock.Mock())
+        assert net.get_method("eth0") == "static"
+
+    def test_get_method_dhcp(self):
+        """METHOD=dhcp → 'dhcp'."""
+        net = ds.OpenNebulaNetwork({"ETH0_METHOD": "dhcp"}, mock.Mock())
+        assert net.get_method("eth0") == "dhcp"
+
+    def test_get_method_skip(self):
+        """METHOD=skip → 'skip'."""
+        net = ds.OpenNebulaNetwork({"ETH0_METHOD": "skip"}, mock.Mock())
+        assert net.get_method("eth0") == "skip"
+
+    def test_get_ip6_method_default_no_ip6(self):
+        """IP6_METHOD absent, no ETHx_IP6 → 'disable'."""
+        net = ds.OpenNebulaNetwork({}, mock.Mock())
+        assert net.get_ip6_method("eth0") == "disable"
+
+    def test_get_ip6_method_default_with_ip6(self):
+        """IP6_METHOD absent, ETHx_IP6 present → 'static'."""
+        net = ds.OpenNebulaNetwork({"ETH0_IP6": IP6_GLOBAL}, mock.Mock())
+        assert net.get_ip6_method("eth0") == "static"
+
+    def test_get_ip6_method_dhcp(self):
+        """IP6_METHOD=dhcp → 'dhcp'."""
+        net = ds.OpenNebulaNetwork({"ETH0_IP6_METHOD": "dhcp"}, mock.Mock())
+        assert net.get_ip6_method("eth0") == "dhcp"
+
+    def test_get_ip6_method_auto(self):
+        """IP6_METHOD=auto → 'auto'."""
+        net = ds.OpenNebulaNetwork({"ETH0_IP6_METHOD": "auto"}, mock.Mock())
+        assert net.get_ip6_method("eth0") == "auto"
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_method_dhcp4(self, m_get_phys_by_mac):
+        """METHOD=dhcp → dhcp4: true, no addresses key."""
+        context = {"ETH0_MAC": MACADDR, "ETH0_METHOD": "dhcp"}
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+            eth = conf["ethernets"][nic]
+            assert eth.get("dhcp4") is True
+            assert "addresses" not in eth
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_method_skip(self, m_get_phys_by_mac):
+        """METHOD=skip → interface absent from ethernets output."""
+        context = {"ETH0_MAC": MACADDR, "ETH0_METHOD": "skip"}
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+            assert nic not in conf["ethernets"]
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_method_static_no_regression(self, m_get_phys_by_mac):
+        """METHOD absent (static) produces same output as before."""
+        context = {
+            "ETH0_MAC": MACADDR,
+            "ETH0_IP": PUBLIC_IP,
+            "ETH0_MASK": "255.255.255.0",
+        }
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+            eth = conf["ethernets"][nic]
+            assert PUBLIC_IP + "/24" in eth["addresses"]
+            assert "dhcp4" not in eth
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_ip6_method_dhcp(self, m_get_phys_by_mac):
+        """IP6_METHOD=dhcp → dhcp6: true, no IPv6 addresses."""
+        context = {"ETH0_MAC": MACADDR, "ETH0_IP6_METHOD": "dhcp"}
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+            eth = conf["ethernets"][nic]
+            assert eth.get("dhcp6") is True
+            assert not any(
+                "/" in a and ":" in a for a in eth.get("addresses", [])
+            )
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_ip6_method_auto(self, m_get_phys_by_mac):
+        """IP6_METHOD=auto → accept-ra: true."""
+        context = {"ETH0_MAC": MACADDR, "ETH0_IP6_METHOD": "auto"}
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+            assert conf["ethernets"][nic].get("accept-ra") is True
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_ip6_method_disable(self, m_get_phys_by_mac):
+        """IP6_METHOD=disable (or default with no IP6) → no IPv6 keys."""
+        context = {
+            "ETH0_MAC": MACADDR,
+            "ETH0_IP6": IP6_GLOBAL,
+            "ETH0_IP6_METHOD": "disable",
+        }
+        for nic in self.system_nics:
+            m_get_phys_by_mac.return_value = {MACADDR: nic}
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+            eth = conf["ethernets"][nic]
+            assert "dhcp6" not in eth
+            assert "accept-ra" not in eth
+            assert not any(":" in a for a in eth.get("addresses", []))
+
     # ------------------------------------------------------------------ #
     # ETHx_ROUTES                                                          #
     # ------------------------------------------------------------------ #