OVS firewall: only remove security group when truly unused

SG was removed only if both its members and ports were empty.
Change to require neither members nor ports remain before
cleanup, preventing removal of still‐active IPs in SG.

Closes-Bug: #2112648
Change-Id: I57ae71a76c742a7d698d347a72f98e1cff469055
Signed-off-by: Jaroslav Pulchart <[email protected]>
Signed-off-by: Brian Haley <[email protected]>
(cherry picked from commit 34225a6)

Author: pulchart

neutron/agent/linux/openvswitch_firewall/firewall.py

@@ -872,7 +872,7 @@ def _cleanup_stale_sg(self):
 
         for sg_id in sg_to_delete:
             sec_group = self.sg_port_map.get_sg(sg_id)
-            if sec_group.members and sec_group.ports:
+            if sec_group.members or sec_group.ports:
                 # sec_group is still in use
                 continue
 

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

@@ -1090,6 +1090,49 @@ def test__cleanup_stale_sg(self):
             sg_removed_mock.assert_called_once_with(1)
             delete_sg_mock.assert_called_once_with(1)
 
+    def test__cleanup_stale_sg_members_and_ports(self):
+        self._prepare_security_group()
+        self.firewall.sg_to_delete = {1}
+        new_members = {constants.IPv4: [1]}
+        self.firewall.update_security_group_members(1, new_members)
+        port_dict = {'device': 'port-id',
+                     'security_groups': [1]}
+        self.firewall.prepare_port_filter(port_dict)
+        with mock.patch.object(self.firewall.conj_ip_manager,
+                               'sg_removed') as sg_removed_mock,\
+            mock.patch.object(self.firewall.sg_port_map,
+                              'delete_sg') as delete_sg_mock:
+            self.firewall._cleanup_stale_sg()
+            sg_removed_mock.assert_not_called()
+            delete_sg_mock.assert_not_called()
+
+    def test__cleanup_stale_sg_just_members(self):
+        self._prepare_security_group()
+        self.firewall.sg_to_delete = {1}
+        new_members = {constants.IPv4: [1]}
+        self.firewall.update_security_group_members(1, new_members)
+        with mock.patch.object(self.firewall.conj_ip_manager,
+                               'sg_removed') as sg_removed_mock,\
+            mock.patch.object(self.firewall.sg_port_map,
+                              'delete_sg') as delete_sg_mock:
+            self.firewall._cleanup_stale_sg()
+            sg_removed_mock.assert_not_called()
+            delete_sg_mock.assert_not_called()
+
+    def test__cleanup_stale_sg_just_ports(self):
+        self._prepare_security_group()
+        self.firewall.sg_to_delete = {1}
+        port_dict = {'device': 'port-id',
+                     'security_groups': [1]}
+        self.firewall.prepare_port_filter(port_dict)
+        with mock.patch.object(self.firewall.conj_ip_manager,
+                               'sg_removed') as sg_removed_mock,\
+            mock.patch.object(self.firewall.sg_port_map,
+                              'delete_sg') as delete_sg_mock:
+            self.firewall._cleanup_stale_sg()
+            sg_removed_mock.assert_not_called()
+            delete_sg_mock.assert_not_called()
+
     def test_get_ovs_port(self):
         ovs_port = self.firewall.get_ovs_port('port_id')
         self.assertEqual(self.fake_ovs_port, ovs_port)