feat(opennebula): support ETHx_ROUTES static routes in network config

cloudinit/sources/DataSourceOpenNebula.py

@@ -228,6 +228,30 @@ def get_gateway6(self, dev: str) -> Optional[str]:
     def get_mask(self, dev: str) -> str:
         return self.get_field(dev, "mask", "255.255.255.0")
 
+    def get_routes(self, dev: str) -> List[Dict[str, str]]:
+        """Parse ETHx_ROUTES into a list of Netplan route dicts.
+
+        Expected format: "NETWORK via GATEWAY[, NETWORK via GATEWAY, ...]"
+        e.g. "10.0.0.0/8 via 192.168.1.1, 192.168.100.0/24 via 10.0.0.1"
+        Returns an empty list when the variable is absent or empty.
+        """
+        routes: List[Dict[str, str]] = []
+        for entry in self.get_field(dev, "routes", "").split(","):
+            entry = entry.strip()
+            if not entry:
+                continue
+            m = re.match(
+                r"\s*(?P<route_to>\S+)\s+via\s+(?P<route_via>\S+)\s*$",
+                entry,
+            )
+            if m:
+                routes.append({"to": m["route_to"], "via": m["route_via"]})
+            else:
+                LOG.warning(
+                    "Unparseable ETHx_ROUTES entry for %s: %r", dev, entry
+                )
+        return routes
+
     @overload
     def get_field(self, dev: str, name: str) -> Optional[str]: ...
     @overload
@@ -304,6 +328,11 @@ def gen_conf(self) -> Dict[str, Any]:
             if mtu:
                 devconf["mtu"] = mtu
 
+            # Set static routes
+            extra_routes: List[Dict[str, str]] = self.get_routes(c_dev)
+            if extra_routes:
+                devconf["routes"] = extra_routes
+
             ethernets[dev] = devconf
 
         netconf["ethernets"] = ethernets

doc/rtd/reference/datasources/opennebula.rst

@@ -72,9 +72,15 @@ the OpenNebula documentation.
     ETH<x>_IP6_ULA
     ETH<x>_IP6_PREFIX_LENGTH
     ETH<x>_IP6_GATEWAY
+    ETH<x>_ROUTES
 
 Static `network configuration`_.
 
+``ETH<x>_ROUTES`` is a comma-separated list of static routes in the form
+``NETWORK via GATEWAY``. For example::
+
+    ETH0_ROUTES="10.0.0.0/8 via 192.168.1.1, 172.16.0.0/12 via 192.168.1.254"
+
 ::
 
     SET_HOSTNAME
@@ -146,5 +152,5 @@ Example VM's context section
 .. _OpenNebula: http://opennebula.org/
 .. _contextualization overview: http://opennebula.org/documentation:documentation:context_overview
 .. _contextualizing VMs: http://opennebula.org/documentation:documentation:cong
-.. _network configuration: https://docs.opennebula.io/
+.. _network configuration: https://docs.opennebula.io/7.2/product/operation_references/configuration_references/template/#context-section
 .. _iso9660: https://en.wikipedia.org/wiki/ISO_9660

tests/unittests/sources/test_opennebula.py

@@ -975,6 +975,76 @@ def test_multiple_nics(self):
 
         assert expected == net.gen_conf()
 
+    # ------------------------------------------------------------------ #
+    # ETHx_ROUTES                                                          #
+    # ------------------------------------------------------------------ #
+
+    @pytest.mark.parametrize(
+        "context,expected",
+        [
+            pytest.param({}, [], id="absent"),
+            pytest.param({"ETH0_ROUTES": ""}, [], id="empty_string"),
+            pytest.param(
+                {"ETH0_ROUTES": "10.0.0.0/8 via 192.168.1.1"},
+                [{"to": "10.0.0.0/8", "via": "192.168.1.1"}],
+                id="single_entry",
+            ),
+            pytest.param(
+                {
+                    "ETH0_ROUTES": (
+                        "10.0.0.0/8 via 192.168.1.1,"
+                        " 172.16.0.0/12 via 192.168.1.254"
+                    )
+                },
+                [
+                    {"to": "10.0.0.0/8", "via": "192.168.1.1"},
+                    {"to": "172.16.0.0/12", "via": "192.168.1.254"},
+                ],
+                id="multiple_comma_separated_entries",
+            ),
+            pytest.param(
+                {"ETH0_ROUTES": "bad-entry, 10.0.0.0/8 via 192.168.1.1"},
+                [{"to": "10.0.0.0/8", "via": "192.168.1.1"}],
+                id="malformed_entry_skipped",
+            ),
+        ],
+    )
+    def test_get_routes(self, context, expected):
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        assert net.get_routes("eth0") == expected
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_routes(self, m_get_phys_by_mac):
+        """Routes from ETHx_ROUTES appear in gen_conf() output."""
+        self.maxDiff = None
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+            "ETH0_IP": "10.0.0.5",
+            "ETH0_MASK": "255.255.255.0",
+            "ETH0_GATEWAY": "10.0.0.1",
+            "ETH0_ROUTES": (
+                "192.168.0.0/16 via 10.0.0.1, 172.16.0.0/12 via 10.0.0.1"
+            ),
+        }
+        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()
+            routes = conf["ethernets"][nic].get("routes", [])
+            assert {"to": "192.168.0.0/16", "via": "10.0.0.1"} in routes
+            assert {"to": "172.16.0.0/12", "via": "10.0.0.1"} in routes
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_no_routes_key_when_absent(self, m_get_phys_by_mac):
+        """gen_conf() does not emit 'routes' key when ETHx_ROUTES is unset."""
+        context = {
+            "ETH0_MAC": "02:00:0a:12:01:01",
+        }
+        m_get_phys_by_mac.return_value = {MACADDR: "eth0"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        conf = net.gen_conf()
+        assert "routes" not in conf["ethernets"]["eth0"]
+
 
 class TestParseShellConfig:
     @pytest.mark.allow_subp_for("bash", "sh")