

def test_get_events(self):
    utils.wait_until_true((lambda : self.monitor.has_updates))
    devices = self.monitor.get_events()
    self.assertTrue(devices.get('added'), 'Initial call should always be true')
    br = self.useFixture(net_helpers.OVSBridgeFixture())
    p1 = self.useFixture(net_helpers.OVSPortFixture(br.bridge))
    p2 = self.useFixture(net_helpers.OVSPortFixture(br.bridge))
    added_devices = [p1.port.name, p2.port.name]
    utils.wait_until_true((lambda : self._expected_devices_events(added_devices, 'added')))
    br.bridge.delete_port(p1.port.name)
    br.bridge.delete_port(p2.port.name)
    removed_devices = [p1.port.name, p2.port.name]
    utils.wait_until_true((lambda : self._expected_devices_events(removed_devices, 'removed')))
    self.monitor.stop(block=True)
    self.monitor.start(block=True, timeout=60)
    devices = self.monitor.get_events()
    self.assertTrue(devices.get('added'), 'Initial call should always be true')

dataset/ETHPy150Open openstack/neutron/neutron/tests/functional/agent/linux/test_ovsdb_monitor.py TestSimpleInterfaceMonitor.test_get_events/original