openstack/neutron
Code Issues Proposed changes

[FT] Use a trivial active wait in `test_create_bridges`

Browse Source 
In the ``test_create_bridges`` test case, in order to check the status
of the bridge monitor, the test access to the internal variable used to
store the bridge names to monitor.

Related-Bug: #2114909
Signed-off-by: Rodolfo Alonso Hernandez <ralonsoh@redhat.com>
Change-Id: Iafbb3de3dbded523d66814ceca534d5da52a6a2a
This commit is contained in:
Rodolfo Alonso Hernandez
2025-09-12 08:55:21 +00:00
committed by Rodolfo Alonso
parent ae016de771
commit ff64d0d784
1 changed files with 9 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
neutron/tests/functional/agent/ovsdb/native/test_connection.py
View File

@@ -13,39 +13,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import threading
from oslo_utils import uuidutils
from ovsdbapp import event
from neutron.agent.common import ovs_lib
from neutron.common import utils as common_utils
from neutron.tests.functional import base
class WaitForBridgesEvent(event.RowEvent):
event_name = 'WaitForBridgesEvent'
ONETIME = True
def __init__(self, bridges, timeout=20):
self.bridges_not_seen = set(bridges)
self.timeout = timeout
self.event = threading.Event()
super().__init__(
(self.ROW_CREATE,), 'Bridge', None)
def matches(self, event, row, old=None):
if event not in self.events or row._table.name != self.table:
return False
self.bridges_not_seen.discard(row.name)
return not self.bridges_not_seen
def run(self, event, row, old):
self.event.set()
def wait(self):
return self.event.wait(self.timeout)
class BridgeMonitorTestCase(base.BaseSudoTestCase):
def _delete_bridges(self, bridges):
@@ -66,11 +40,13 @@ class BridgeMonitorTestCase(base.BaseSudoTestCase):
self.ovs.ovsdb.idl_monitor.start_bridge_monitor(bridges_to_monitor)
self.addCleanup(self._delete_bridges, bridges_to_create)
event = WaitForBridgesEvent(bridges_to_monitor)
self.ovs.ovsdb.idl_monitor.notify_handler.watch_event(event)
for bridge in bridges_to_create:
self.ovs.add_bridge(bridge)
self.assertTrue(event.wait())
self.assertEqual(bridges_to_monitor,
self.ovs.ovsdb.idl_monitor.bridges_added)
self.assertEqual([], self.ovs.ovsdb.idl_monitor.bridges_added)
_idl_mon = self.ovs.ovsdb.idl_monitor
common_utils.wait_until_true(
lambda: set(bridges_to_monitor) ==
set(_idl_mon._bridges_added_list),
timeout=20)
self.assertEqual(bridges_to_monitor, _idl_mon.bridges_added)
self.assertEqual([], _idl_mon.bridges_added)