Merge "Always get local vlan from port other_config" into stable/2024.1

neutron/agent/linux/openvswitch_firewall/firewall.py

@@ -705,6 +705,22 @@ class OVSFirewallDriver(firewall.FirewallDriver):
         port_id = port['device']
         return self.sg_port_map.ports.get(port_id)
 
+    def _create_of_port(self, port, ovs_port):
+        # Should always try to get the local vlan tag from
+        # the OVSDB Port other_config, since the ovs-agent's
+        # LocalVlanManager always allocated/updated it and then
+        # set_db_attribute to Port other_config before this.
+        port_vlan_id = self._get_port_vlan_tag(ovs_port.port_name)
+        segment_id = self._get_port_segmentation_id(
+            ovs_port.port_name)
+        network_type = self._get_port_network_type(
+            ovs_port.port_name)
+        physical_network = self._get_port_physical_network(
+            ovs_port.port_name)
+        return OFPort(port, ovs_port, port_vlan_id,
+                      segment_id,
+                      network_type, physical_network)
+
     def get_or_create_ofport(self, port):
         """Get ofport specified by port['device'], checking and reflecting
         ofport changes.
@@ -715,22 +731,12 @@ class OVSFirewallDriver(firewall.FirewallDriver):
         try:
             of_port = self.sg_port_map.ports[port_id]
         except KeyError:
-            port_vlan_id = self._get_port_vlan_tag(ovs_port.port_name)
+            of_port = self._create_of_port(port, ovs_port)
-            segment_id = self._get_port_segmentation_id(
-                ovs_port.port_name)
-            network_type = self._get_port_network_type(
-                ovs_port.port_name)
-            physical_network = self._get_port_physical_network(
-                ovs_port.port_name)
-            of_port = OFPort(port, ovs_port, port_vlan_id,
-                             segment_id,
-                             network_type, physical_network)
             self.sg_port_map.create_port(of_port, port)
         else:
             if of_port.ofport != ovs_port.ofport:
                 self.sg_port_map.remove_port(of_port)
-                of_port = OFPort(port, ovs_port, of_port.vlan_tag,
+                of_port = self._create_of_port(port, ovs_port)
-                                 of_port.segment_id)
                 self.sg_port_map.create_port(of_port, port)
             else:
                 self.sg_port_map.update_port(of_port, port)

neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py

@@ -658,6 +658,21 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         self.assertIn(of_port.id, self.firewall.sg_port_map.ports.keys())
         self.assertEqual(port.ofport, 2)
 
+    def test_get_or_create_ofport_changed_and_local_vlan_changed(self):
+        port_dict = {
+            'device': 'port-id',
+            'security_groups': [123, 456]}
+        of_port = create_ofport(port_dict)
+        self.firewall.sg_port_map.ports[of_port.id] = of_port
+        fake_ovs_port = FakeOVSPort('port', 2, '00:00:00:00:00:00')
+        self.mock_bridge.br.get_vif_port_by_id.return_value = \
+            fake_ovs_port
+        self.mock_bridge.br.db_get_val.return_value = {"tag": 10}
+        port = self.firewall.get_or_create_ofport(port_dict)
+        self.assertIn(of_port.id, self.firewall.sg_port_map.ports.keys())
+        self.assertEqual(port.ofport, 2)
+        self.assertEqual(port.vlan_tag, 10)
+
     def test_get_or_create_ofport_missing(self):
         port_dict = {
             'device': 'port-id',