fix(neutron-understack): select the ACTIVE port binding, not bindings[0]

docs/design-guide/neutron-networking.md

@@ -499,6 +499,49 @@ Neutron based on certain data provided to port creation and update API calls.
 - the port has a binding host
 - the port is either unbound or has previously failed to bind
 
+### Port Representations
+
+Neutron exposes a port in three different shapes and you will see all of them
+in the Neutron source and in our drivers:
+
+| Representation | How you get it | Notes |
+| -------------- | -------------- | ----- |
+| Dict | `core_plugin.get_port()`, `PortContext.current` | API-shaped. Binding fields appear flattened as `binding:vif_type`, `binding:vnic_type`, `binding:host_id`, etc. |
+| OVO | `neutron.objects.ports.Port.get_object()` | Typed versioned object. Bindings are in `port.bindings`, binding levels in `port.binding_levels`. |
+| SQLAlchemy model | `payload.metadata["port_db"]`, `ml2.db.get_port()` | ML2 internal. Bindings are in `port.port_bindings`. Don't go looking for these; use them when Neutron hands one to you. |
+
+Use the representation that matches where the port came from, and don't
+re-fetch a port just to change its shape:
+
+- If Neutron hands you a dict (`PortContext.current`) or a SQLAlchemy model
+  (for example `payload.metadata["port_db"]` in a registry callback), work with
+  that directly.
+- When we have only a port ID, as with a trunk's parent port, fetch the **OVO**
+  with `utils.fetch_port_object()`. It is typed, and it is what Neutron's own
+  newer code uses when it needs binding details. We don't use `get_port()` for
+  this.
+
+#### Selecting a binding
+
+A port can have several bindings (one `ACTIVE`, any number of `INACTIVE` ones
+during live migration), so never index with `port.bindings[0]`. Select the
+binding with `get_port_binding_by_status_and_host()` from
+`neutron_lib.plugins.utils`, then read `vif_type`, `vnic_type`, `profile` and
+`host` from the result. For an OVO port, `utils.active_port_binding()` wraps
+this and returns the `ACTIVE` binding, or `None` if there isn't one:
+
+```python
+binding = utils.active_port_binding(port)
+if binding and binding.vnic_type == portbindings.VNIC_BAREMETAL:
+    ...
+```
+
+For a SQLAlchemy model, call the neutron-lib helper directly with
+`port.port_bindings` (the model's attribute name) instead of `port.bindings`.
+The helper only subscripts each binding (`binding["status"]`,
+`binding["host"]`), so it works on either. Pass `raise_if_not_found=True` to get
+`PortBindingNotFound` instead of `None`.
+
 ### Router Interface Lifecycle
 
 When a subnet is attached to a router, the understack ML2 driver sets up a path

python/neutron-understack/neutron_understack/trunk.py

@@ -160,8 +160,9 @@ def _handle_segment_allocation(
     def _add_subports_networks_to_parent_port_switchport(
         self, parent_port: Port, subports: list[SubPort]
     ) -> None:
-        binding_profile = parent_port.bindings[0].profile
-        binding_host = parent_port.bindings[0].host
+        parent_binding = utils.active_port_binding(parent_port)
+        binding_profile = parent_binding.profile
+        binding_host = parent_binding.host
 
         physnet = binding_profile.get("physical_network")
         if not physnet:
@@ -182,7 +183,8 @@ def _clean_parent_port_switchport_config(
         parent_port_obj = utils.fetch_port_object(trunk.port_id)
         if not utils.parent_port_is_bound(parent_port_obj):
             return
-        self._handle_segment_deallocation(subports, parent_port_obj.bindings[0].host)
+        parent_binding = utils.active_port_binding(parent_port_obj)
+        self._handle_segment_deallocation(subports, parent_binding.host)
 
     def _delete_unused_segment(self, segment_id: str) -> NetworkSegment:
         network_segment = utils.network_segment_by_id(segment_id)

python/neutron-understack/neutron_understack/undersync_mech.py

@@ -144,7 +144,7 @@ def _validate_parent_physnet(self, resource, event, trunk_plugin, payload):
         parent_port = utils.fetch_port_object(trunk.port_id)
         if not utils.parent_port_is_bound(parent_port):
             return
-        if not parent_port.bindings[0].profile.get("physical_network"):
+        if not utils.active_port_binding(parent_port).profile.get(api.PHYSICAL_NETWORK):
             raise exc.BadRequest(
                 resource="port", msg=_missing_physnet_msg(parent_port.id)
             )
@@ -165,7 +165,9 @@ def _sync_trunk_parent(self, trunk) -> None:
         if not utils.parent_port_is_bound(parent_port):
             return
 
-        physnet = parent_port.bindings[0].profile.get("physical_network")
+        physnet = utils.active_port_binding(parent_port).profile.get(
+            api.PHYSICAL_NETWORK
+        )
         if not physnet:
             # _validate_parent_physnet rejects this while it is still
             # abortable, so the binding profile changed underneath us. Raising
@@ -192,7 +194,7 @@ def bind_port(self, context: PortContext) -> None:
         binding_profile = port.get(portbindings.PROFILE) or {}
         if not binding_profile.get("physical_network"):
             LOG.debug(
-                "Skipping port %s without a physical_network in its " "binding_profile",
+                "Skipping port %s without a physical_network in its binding_profile",
                 port["id"],
             )
             return

python/neutron-understack/neutron_understack/utils.py

@@ -15,6 +15,7 @@
 from neutron_lib import context as n_context
 from neutron_lib.api.definitions import segment as segment_def
 from neutron_lib.plugins import directory
+from neutron_lib.plugins import utils as plugin_utils
 from neutron_lib.plugins.ml2 import api
 from oslo_config import cfg
 
@@ -472,8 +473,15 @@ def release_segment_if_unused(segment: NetworkSegment) -> None:
         release_dynamic_segment(segment.id)
 
 
+def active_port_binding(port: port_obj.Port) -> port_obj.PortBinding | None:
+    """The port's ACTIVE binding, or None if it has none."""
+    return plugin_utils.get_port_binding_by_status_and_host(
+        port.bindings, constants.ACTIVE, port_id=port.id
+    )
+
+
 def parent_port_is_bound(port: port_obj.Port) -> bool:
-    port_binding = port.bindings[0]
+    port_binding = active_port_binding(port)
     return bool(
         port_binding
         and port_binding.vif_type == portbindings.VIF_TYPE_OTHER