# under the License.
from oslo import messaging
+from sqlalchemy.orm import exc
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.api.rpc.handlers import dvr_rpc
return {'device': device,
'exists': port_exists}
- port_exists = bool(plugin.update_port_status(rpc_context, port_id,
- q_const.PORT_STATUS_DOWN,
- host))
+ try:
+ port_exists = bool(plugin.update_port_status(
+ rpc_context, port_id, q_const.PORT_STATUS_DOWN, host))
+ except exc.StaleDataError:
+ port_exists = False
+ LOG.debug("delete_port and update_device_down are being executed "
+ "concurrently. Ignoring StaleDataError.")
return {'device': device,
'exists': port_exists}
import mock
from oslo_context import context as oslo_context
+from sqlalchemy.orm import exc
from neutron.agent import rpc as agent_rpc
from neutron.common import constants
'fake_context', 'fake_port_id', constants.PORT_STATUS_DOWN,
'fake_host')
+ def test_update_device_down_call_update_port_status_failed(self):
+ self.plugin.update_port_status.side_effect = exc.StaleDataError
+ self.assertEqual({'device': 'fake_device', 'exists': False},
+ self.callbacks.update_device_down(
+ 'fake_context', device='fake_device'))
+
class RpcApiTestCase(base.BaseTestCase):