ML2 plugin cannot raise NoResultFound exception

The ML2 plugin cannot raise NoResultFound exception because it does not
use the correct sqlalchemy library:
'from sqlalchemy import exc as ...' instead of 'from sqlalchemy.orm
import exc as ...'

Closes-Bug: #1277914
Change-Id: If3819adc62b9254f0c08eea6dcfcf5f06288e20e
This commit is contained in:
Édouard Thuleau 2014-02-08 18:28:19 +01:00
parent 6df373d9c0
commit e62e5f3763
2 changed files with 26 additions and 2 deletions

View File

@ -15,6 +15,7 @@
from oslo.config import cfg from oslo.config import cfg
from sqlalchemy import exc as sql_exc from sqlalchemy import exc as sql_exc
from sqlalchemy.orm import exc as sa_exc
from neutron.agent import securitygroups_rpc as sg_rpc from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
@ -595,7 +596,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
port_db = (session.query(models_v2.Port). port_db = (session.query(models_v2.Port).
enable_eagerloads(False). enable_eagerloads(False).
filter_by(id=id).with_lockmode('update').one()) filter_by(id=id).with_lockmode('update').one())
except sql_exc.NoResultFound: except sa_exc.NoResultFound:
raise exc.PortNotFound(port_id=id) raise exc.PortNotFound(port_id=id)
original_port = self._make_port_dict(port_db) original_port = self._make_port_dict(port_db)
updated_port = super(Ml2Plugin, self).update_port(context, id, updated_port = super(Ml2Plugin, self).update_port(context, id,
@ -653,7 +654,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
port_db = (session.query(models_v2.Port). port_db = (session.query(models_v2.Port).
enable_eagerloads(False). enable_eagerloads(False).
filter_by(id=id).with_lockmode('update').one()) filter_by(id=id).with_lockmode('update').one())
except sql_exc.NoResultFound: except sa_exc.NoResultFound:
# the port existed when l3plugin.prevent_l3_port_deletion # the port existed when l3plugin.prevent_l3_port_deletion
# was called but now is already gone # was called but now is already gone
LOG.debug(_("The port '%s' was deleted"), id) LOG.debug(_("The port '%s' was deleted"), id)

View File

@ -13,10 +13,16 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import mock
from neutron.common import exceptions as exc
from neutron import context
from neutron.extensions import multiprovidernet as mpnet from neutron.extensions import multiprovidernet as mpnet
from neutron.extensions import portbindings from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet from neutron.extensions import providernet as pnet
from neutron import manager
from neutron.plugins.ml2 import config from neutron.plugins.ml2 import config
from neutron.plugins.ml2 import plugin as ml2_plugin
from neutron.tests.unit import _test_extension_portbindings as test_bindings from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit import test_db_plugin as test_plugin from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import test_extension_extradhcpopts as test_dhcpopts from neutron.tests.unit import test_extension_extradhcpopts as test_dhcpopts
@ -90,6 +96,23 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
self.assertEqual(port['port']['status'], 'DOWN') self.assertEqual(port['port']['status'], 'DOWN')
self.assertEqual(self.port_create_status, 'DOWN') self.assertEqual(self.port_create_status, 'DOWN')
def test_update_non_existent_port(self):
ctx = context.get_admin_context()
plugin = manager.NeutronManager.get_plugin()
data = {'port': {'admin_state_up': False}}
self.assertRaises(exc.PortNotFound, plugin.update_port, ctx,
'invalid-uuid', data)
def test_delete_non_existent_port(self):
ctx = context.get_admin_context()
plugin = manager.NeutronManager.get_plugin()
with mock.patch.object(ml2_plugin.LOG, 'debug') as log_debug:
plugin.delete_port(ctx, 'invalid-uuid', l3_port_check=False)
log_debug.assert_has_calls([
mock.call(_("Deleting port %s"), 'invalid-uuid'),
mock.call(_("The port '%s' was deleted"), 'invalid-uuid')
])
class TestMl2PortBinding(Ml2PluginV2TestCase, class TestMl2PortBinding(Ml2PluginV2TestCase,
test_bindings.PortBindingsTestCase): test_bindings.PortBindingsTestCase):