Preserve PXE flags after enrolment

Author: stevekeay

python/ironic-understack/ironic_understack/port_bios_name_hook.py

@@ -1,5 +1,6 @@
 from typing import ClassVar
 
+from ironic.common import exception
 from ironic.drivers.modules.inspector.hooks import base
 from oslo_log import log as logging
 
@@ -34,35 +35,32 @@ def __call__(self, task, inventory, plugin_data):
             i["mac_address"].upper(): i["name"] for i in inspected_interfaces
         }
 
-        pxe_nics = _enrolled_pxe_nics(task)
+        ports = list(ironic_ports_for_node(task.context, task.node.id))
+        initial_enroll = _is_initial_enroll(ports)
+        pxe_nics = _enrolled_pxe_nics(task) if initial_enroll else []
 
-        for baremetal_port in ironic_ports_for_node(task.context, task.node.id):
+        for baremetal_port in ports:
             mac = baremetal_port.address.upper()
             bios_name = interface_names.get(mac)
 
             _set_port_extra(baremetal_port, mac, bios_name)
             _set_port_name(baremetal_port, mac, bios_name, task.node.name)
 
-            is_pxe = bios_name is not None and any(
-                pxe_nic.startswith(bios_name) or bios_name.startswith(pxe_nic)
-                for pxe_nic in pxe_nics
-            )
-
-            if baremetal_port.pxe_enabled != is_pxe:
-                LOG.info(
-                    "Port %s (%s) pxe_enabled %s -> %s",
-                    mac,
-                    bios_name,
-                    baremetal_port.pxe_enabled,
-                    is_pxe,
+            if initial_enroll:
+                is_pxe = bios_name is not None and any(
+                    pxe_nic.startswith(bios_name) or bios_name.startswith(pxe_nic)
+                    for pxe_nic in pxe_nics
                 )
-                baremetal_port.pxe_enabled = is_pxe
-                baremetal_port.save()
+                _set_port_pxe_enabled(baremetal_port, mac, bios_name, is_pxe)
+            else:
+                is_pxe = baremetal_port.pxe_enabled
 
-            if is_pxe:
+            if initial_enroll and is_pxe:
                 _set_port_physical_network(baremetal_port, mac)
                 _set_port_local_link_connection(baremetal_port, mac)
 
+        _assert_has_pxe_port(task, ports)
+
 
 def _enrolled_pxe_nics(task) -> list[str]:
     """Read enrolled PXE NIC names from node.extra, or use broad prefixes."""
@@ -84,6 +82,34 @@ def _enrolled_pxe_nics(task) -> list[str]:
         return PXE_BIOS_NAME_PREFIXES
 
 
+def _is_initial_enroll(ports) -> bool:
+    return all(
+        not port.physical_network or "enrol" in port.physical_network for port in ports
+    )
+
+
+def _set_port_pxe_enabled(baremetal_port, mac, bios_name, is_pxe):
+    if baremetal_port.pxe_enabled != is_pxe:
+        LOG.info(
+            "Port %s (%s) pxe_enabled %s -> %s",
+            mac,
+            bios_name,
+            baremetal_port.pxe_enabled,
+            is_pxe,
+        )
+        baremetal_port.pxe_enabled = is_pxe
+        baremetal_port.save()
+
+
+def _assert_has_pxe_port(task, ports):
+    if any(port.pxe_enabled for port in ports):
+        return
+
+    msg = f"No PXE-enabled ports found for node {task.node.uuid}"
+    LOG.error(msg)
+    raise exception.InvalidNodeInventory(node=task.node.uuid, reason=msg)
+
+
 def _set_port_extra(baremetal_port, mac, required_bios_name):
     extra = baremetal_port.extra
     current_bios_name = extra.get("bios_name")

python/ironic-understack/ironic_understack/tests/test_port_bios_name_hook.py

@@ -1,5 +1,7 @@
 import logging
 
+import pytest
+from ironic.common import exception
 from oslo_utils import uuidutils
 
 from ironic_understack.port_bios_name_hook import PortBiosNameHook
@@ -137,6 +139,7 @@ def test_retaining_physical_network(mocker, caplog):
     port = _make_port(
         mocker,
         "11:11:11:11:11:11",
+        pxe_enabled=True,
         physical_network="previous_value",
         local_link_connection={
             "port_id": "Ethernet1/19",
@@ -156,13 +159,23 @@ def test_retaining_physical_network(mocker, caplog):
     assert port.local_link_connection["port_id"] == "Ethernet1/19"
 
 
-def test_clears_pxe_on_previously_enabled_port(mocker, caplog):
-    """Port that was pxe_enabled but no longer matches gets cleared."""
+def test_preserves_pxe_on_post_enroll_ports(mocker, caplog):
+    """Post-enrol inspection keeps the existing PXE decision."""
     caplog.set_level(logging.DEBUG)
     task = _make_task(mocker, enrolled_pxe_ports=["NIC.Integrated.1-2-1"])
 
-    port1 = _make_port(mocker, "11:11:11:11:11:11", pxe_enabled=True)
-    port2 = _make_port(mocker, "22:22:22:22:22:22")
+    port1 = _make_port(
+        mocker,
+        "11:11:11:11:11:11",
+        pxe_enabled=True,
+        physical_network="f20-1-network",
+    )
+    port2 = _make_port(
+        mocker,
+        "22:22:22:22:22:22",
+        pxe_enabled=False,
+        physical_network="f20-1-network",
+    )
 
     mocker.patch(
         "ironic_understack.port_bios_name_hook.ironic_ports_for_node",
@@ -171,28 +184,62 @@ def test_clears_pxe_on_previously_enabled_port(mocker, caplog):
 
     PortBiosNameHook().__call__(task, _INVENTORY, {})
 
-    assert port1.pxe_enabled is False
-    assert port2.pxe_enabled is True
+    assert port1.pxe_enabled is True
+    assert port2.pxe_enabled is False
+
+
+def test_errors_if_post_enroll_ports_have_no_pxe_enabled_port(mocker, caplog):
+    """Post-enrol inspection fails if no PXE-enabled port exists."""
+    caplog.set_level(logging.DEBUG)
+    task = _make_task(mocker, enrolled_pxe_ports=["NIC.Integrated.1-1-1"])
+
+    port1 = _make_port(
+        mocker,
+        "11:11:11:11:11:11",
+        pxe_enabled=False,
+        physical_network="f20-1-network",
+    )
+    port2 = _make_port(
+        mocker,
+        "22:22:22:22:22:22",
+        pxe_enabled=False,
+        physical_network="f20-1-network",
+    )
+
+    mocker.patch(
+        "ironic_understack.port_bios_name_hook.ironic_ports_for_node",
+        return_value=[port1, port2],
+    )
+
+    with pytest.raises(exception.InvalidNodeInventory, match="No PXE-enabled ports"):
+        PortBiosNameHook().__call__(task, _INVENTORY, {})
 
 
 def test_removing_bios_name(mocker, caplog):
     """Port with unknown MAC gets bios_name removed."""
     caplog.set_level(logging.DEBUG)
     task = _make_task(mocker, enrolled_pxe_ports=["NIC.Integrated.1-1-1"])
 
-    port = _make_port(
+    unknown_port = _make_port(
         mocker,
         "33:33:33:33:33:33",
         extra={"bios_name": "old_name_no_longer_valid"},
         name="original-name",
+        physical_network="f20-1-network",
+    )
+    pxe_port = _make_port(
+        mocker,
+        "11:11:11:11:11:11",
+        pxe_enabled=True,
+        physical_network="f20-1-network",
     )
 
     mocker.patch(
         "ironic_understack.port_bios_name_hook.ironic_ports_for_node",
-        return_value=[port],
+        return_value=[unknown_port, pxe_port],
     )
 
     PortBiosNameHook().__call__(task, _INVENTORY, {})
 
-    assert port.name == "original-name"
-    assert "bios_name" not in port.extra
+    assert unknown_port.name == "original-name"
+    assert "bios_name" not in unknown_port.extra