|
26 | 26 | from neutron.tests.functional import base
|
27 | 27 |
|
28 | 28 |
|
29 |
| -class WaitForLogicalSwitchPortUpdateEvent(event.WaitEvent): |
30 |
| - event_name = 'WaitForDataPathBindingCreateEvent' |
| 29 | +class WaitForPortBindingDeleteEvent(event.WaitEvent): |
| 30 | + event_name = 'WaitForPortBindingDeleteEvent' |
31 | 31 |
|
32 |
| - def __init__(self): |
33 |
| - table = 'Logical_Switch_Port' |
34 |
| - events = (self.ROW_UPDATE,) |
35 |
| - super().__init__(events, table, None, timeout=15) |
| 32 | + def __init__(self, port_id): |
| 33 | + table = 'Port_Binding' |
| 34 | + events = (self.ROW_DELETE, ) |
| 35 | + conditions = (('logical_port', '=', port_id), ) |
| 36 | + super().__init__(events, table, conditions, timeout=10) |
36 | 37 |
|
37 | 38 |
|
38 | 39 | class TestOVNTrunkDriver(base.TestOVNFunctionalBase):
|
@@ -127,14 +128,14 @@ def test_subport_add(self):
|
127 | 128 | def test_subport_delete(self):
|
128 | 129 | with self.subport() as subport:
|
129 | 130 | with self.trunk([subport]) as trunk:
|
130 |
| - lsp_event = WaitForLogicalSwitchPortUpdateEvent() |
131 |
| - self.mech_driver.nb_ovn.idl.notify_handler.watch_events( |
132 |
| - (lsp_event,)) |
| 131 | + pb_event = WaitForPortBindingDeleteEvent(subport['port_id']) |
| 132 | + self.mech_driver.sb_ovn.idl.notify_handler.watch_event( |
| 133 | + pb_event) |
133 | 134 | self.trunk_plugin.remove_subports(self.context, trunk['id'],
|
134 | 135 | {'sub_ports': [subport]})
|
135 | 136 | new_trunk = self.trunk_plugin.get_trunk(self.context,
|
136 | 137 | trunk['id'])
|
137 |
| - self.assertTrue(lsp_event.wait()) |
| 138 | + self.assertTrue(pb_event.wait()) |
138 | 139 | self._verify_trunk_info(new_trunk, has_items=False)
|
139 | 140 |
|
140 | 141 | def test_trunk_delete(self):
|
|
0 commit comments