from neutron.db import portbindings_db
from neutron.extensions import portbindings
from neutron.extensions import providernet
+from neutron.openstack.common import excutils
from neutron.openstack.common import importutils
from neutron.openstack.common import log as logging
from neutron.openstack.common import uuidutils as uuidutils
self._send_create_network_request(context, net, segment_pairs)
except(cisco_exceptions.VSMError,
cisco_exceptions.VSMConnectionFailed):
- super(N1kvNeutronPluginV2, self).delete_network(context, net['id'])
+ with excutils.save_and_reraise_exception():
+ self._delete_network_db(context, net['id'])
else:
LOG.debug(_("Created network: %s"), net['id'])
return net
"""
session = context.session
with session.begin(subtransactions=True):
- binding = n1kv_db_v2.get_network_binding(session, id)
network = self.get_network(context, id)
if n1kv_db_v2.is_trunk_member(session, id):
msg = _("Cannot delete network '%s' "
msg = _("Cannot delete network '%s' that is a member of a "
"multi-segment network") % network['name']
raise n_exc.InvalidInput(error_message=msg)
+ self._delete_network_db(context, id)
+ # the network_binding record is deleted via cascade from
+ # the network record, so explicit removal is not necessary
+ self._send_delete_network_request(context, network)
+ LOG.debug("Deleted network: %s", id)
+
+ def _delete_network_db(self, context, id):
+ session = context.session
+ with session.begin(subtransactions=True):
+ binding = n1kv_db_v2.get_network_binding(session, id)
if binding.network_type == c_const.NETWORK_TYPE_OVERLAY:
n1kv_db_v2.release_vxlan(session, binding.segmentation_id)
elif binding.network_type == c_const.NETWORK_TYPE_VLAN:
n1kv_db_v2.release_vlan(session, binding.physical_network,
binding.segmentation_id)
- self._process_l3_delete(context, id)
super(N1kvNeutronPluginV2, self).delete_network(context, id)
- # the network_binding record is deleted via cascade from
- # the network record, so explicit removal is not necessary
- self._send_delete_network_request(context, network)
- LOG.debug(_("Deleted network: %s"), id)
def get_network(self, context, id, fields=None):
"""
"""
p_profile = None
port_count = None
+ vm_network = None
vm_network_name = None
profile_id_set = False
profile_id,
pt['network_id'])
port_count = 1
- n1kv_db_v2.add_vm_network(context.session,
- vm_network_name,
- profile_id,
- pt['network_id'],
- port_count)
+ vm_network = n1kv_db_v2.add_vm_network(context.session,
+ vm_network_name,
+ profile_id,
+ pt['network_id'],
+ port_count)
else:
# Update port count of the VM network.
vm_network_name = vm_network['name']
vm_network_name)
except(cisco_exceptions.VSMError,
cisco_exceptions.VSMConnectionFailed):
- super(N1kvNeutronPluginV2, self).delete_port(context, pt['id'])
+ with excutils.save_and_reraise_exception():
+ self._delete_port_db(context, pt, vm_network)
else:
LOG.debug(_("Created port: %s"), pt)
return pt
vm_network = n1kv_db_v2.get_vm_network(context.session,
port[n1kv.PROFILE_ID],
port['network_id'])
+ router_ids = self.disassociate_floatingips(
+ context, id, do_notify=False)
+ self._delete_port_db(context, port, vm_network)
+
+ # now that we've left db transaction, we are safe to notify
+ self.notify_routers_updated(context, router_ids)
+ self._send_delete_port_request(context, port, vm_network)
+
+ def _delete_port_db(self, context, port, vm_network):
+ with context.session.begin(subtransactions=True):
vm_network['port_count'] -= 1
n1kv_db_v2.update_vm_network_port_count(context.session,
vm_network['name'],
n1kv_db_v2.delete_vm_network(context.session,
port[n1kv.PROFILE_ID],
port['network_id'])
- router_ids = self.disassociate_floatingips(
- context, id, do_notify=False)
- super(N1kvNeutronPluginV2, self).delete_port(context, id)
-
- # now that we've left db transaction, we are safe to notify
- self.notify_routers_updated(context, router_ids)
-
- self._send_delete_port_request(context, port, vm_network)
+ super(N1kvNeutronPluginV2, self).delete_port(context, port['id'])
def get_port(self, context, id, fields=None):
"""
self._send_create_subnet_request(context, sub)
except(cisco_exceptions.VSMError,
cisco_exceptions.VSMConnectionFailed):
- super(N1kvNeutronPluginV2, self).delete_subnet(context, sub['id'])
+ with excutils.save_and_reraise_exception():
+ super(N1kvNeutronPluginV2,
+ self).delete_subnet(context, sub['id'])
else:
LOG.debug(_("Created subnet: %s"), sub['id'])
return sub
context.tenant_id)
except(cisco_exceptions.VSMError,
cisco_exceptions.VSMConnectionFailed):
- n1kv_db_v2.delete_profile_binding(context.session,
- context.tenant_id,
- net_p['id'])
+ with excutils.save_and_reraise_exception():
+ super(N1kvNeutronPluginV2,
+ self).delete_network_profile(context, net_p['id'])
try:
self._send_create_network_profile_request(context, net_p)
except(cisco_exceptions.VSMError,
cisco_exceptions.VSMConnectionFailed):
- n1kv_db_v2.delete_profile_binding(context.session,
- context.tenant_id,
- net_p['id'])
- self._send_delete_logical_network_request(net_p)
+ with excutils.save_and_reraise_exception():
+ super(N1kvNeutronPluginV2,
+ self).delete_network_profile(context, net_p['id'])
+ self._send_delete_logical_network_request(net_p)
return net_p
def delete_network_profile(self, context, id):
import neutron.db.api as db
from neutron.extensions import portbindings
from neutron import manager
+from neutron.plugins.cisco.common import cisco_constants as c_const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.db import n1kv_db_v2
+from neutron.plugins.cisco.db import n1kv_models_v2
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco import extensions
from neutron.plugins.cisco.extensions import n1kv
def _make_test_profile(self,
name='default_network_profile',
+ segment_type=c_const.NETWORK_TYPE_VLAN,
segment_range='386-400'):
"""
Create a profile record for testing purposes.
:param name: string representing the name of the network profile to
create. Default argument value chosen to correspond to the
default name specified in config.py file.
+ :param segment_type: string representing the type of network segment.
:param segment_range: string representing the segment range for network
profile.
"""
db_session = db.get_session()
profile = {'name': name,
- 'segment_type': 'vlan',
- 'physical_network': PHYS_NET,
+ 'segment_type': segment_type,
'tenant_id': self.tenant_id,
'segment_range': segment_range}
- net_p = n1kv_db_v2.create_network_profile(db_session, profile)
- n1kv_db_v2.sync_vlan_allocations(db_session, net_p)
+ if segment_type == c_const.NETWORK_TYPE_OVERLAY:
+ profile['sub_type'] = 'unicast'
+ profile['multicast_ip_range'] = '0.0.0.0'
+ net_p = n1kv_db_v2.create_network_profile(db_session, profile)
+ n1kv_db_v2.sync_vxlan_allocations(db_session, net_p)
+ elif segment_type == c_const.NETWORK_TYPE_VLAN:
+ profile['physical_network'] = PHYS_NET
+ net_p = n1kv_db_v2.create_network_profile(db_session, profile)
+ n1kv_db_v2.sync_vlan_allocations(db_session, net_p)
return net_p
def setUp(self):
PHYS_NET,
vlan)
+ def test_create_network_profile_rollback_profile_binding(self):
+ """Test rollback of profile binding if network profile create fails."""
+ db_session = db.get_session()
+ client_patch = mock.patch(n1kv_client.__name__ + ".Client",
+ new=fake_client.TestClientInvalidResponse)
+ client_patch.start()
+ net_p_dict = self._prepare_net_profile_data(c_const.NETWORK_TYPE_VLAN)
+ self.new_create_request('network_profiles', net_p_dict)
+ bindings = (db_session.query(n1kv_models_v2.ProfileBinding).filter_by(
+ profile_type="network"))
+ self.assertEqual(bindings.count(), 0)
+
class TestN1kvBasicGet(test_plugin.TestBasicGet,
N1kvPluginTestCase):
self.assertEqual(res.status_int, 500)
client_patch.stop()
+ def test_create_first_port_rollback_vmnetwork(self):
+ """Test whether VMNetwork is cleaned up if port create fails on VSM."""
+ db_session = db.get_session()
+ profile_obj = self._make_test_policy_profile(name='test_profile')
+ with self.network() as network:
+ client_patch = mock.patch(n1kv_client.__name__ + ".Client",
+ new=fake_client.
+ TestClientInvalidResponse)
+ client_patch.start()
+ data = {'port': {n1kv.PROFILE_ID: profile_obj.id,
+ 'tenant_id': self.tenant_id,
+ 'network_id': network['network']['id'],
+ }}
+ self.new_create_request('ports', data)
+ self.assertRaises(c_exc.VMNetworkNotFound,
+ n1kv_db_v2.get_vm_network,
+ db_session,
+ profile_obj.id,
+ network['network']['id'])
+ # Explicit stop of failure response mock from controller required
+ # for network object clean up to succeed.
+ client_patch.stop()
+
+ def test_create_next_port_rollback_vmnetwork_count(self):
+ """Test whether VMNetwork count if port create fails on VSM."""
+ db_session = db.get_session()
+ with self.port() as port:
+ pt = port['port']
+ old_vmn = n1kv_db_v2.get_vm_network(db_session,
+ pt['n1kv:profile_id'],
+ pt['network_id'])
+ client_patch = mock.patch(n1kv_client.__name__ + ".Client",
+ new=fake_client.
+ TestClientInvalidResponse)
+ client_patch.start()
+ data = {'port': {n1kv.PROFILE_ID: pt['n1kv:profile_id'],
+ 'tenant_id': pt['tenant_id'],
+ 'network_id': pt['network_id']}}
+ self.new_create_request('ports', data)
+ new_vmn = n1kv_db_v2.get_vm_network(db_session,
+ pt['n1kv:profile_id'],
+ pt['network_id'])
+ self.assertEqual(old_vmn.port_count, new_vmn.port_count)
+ # Explicit stop of failure response mock from controller required
+ # for network object clean up to succeed.
+ client_patch.stop()
+
class TestN1kvPolicyProfiles(N1kvPluginTestCase):
def test_populate_policy_profile(self):
# Network update should fail to update network profile id.
self.assertEqual(res.status_int, 400)
+ def test_create_network_rollback_deallocate_vlan_segment(self):
+ """Test vlan segment deallocation on network create failure."""
+ profile_obj = self._make_test_profile(name='test_profile',
+ segment_range='20-23')
+ data = self._prepare_net_data(profile_obj.id)
+ client_patch = mock.patch(n1kv_client.__name__ + ".Client",
+ new=fake_client.TestClientInvalidResponse)
+ client_patch.start()
+ self.new_create_request('networks', data)
+ db_session = db.get_session()
+ self.assertFalse(n1kv_db_v2.get_vlan_allocation(db_session,
+ PHYS_NET,
+ 20).allocated)
+
+ def test_create_network_rollback_deallocate_overlay_segment(self):
+ """Test overlay segment deallocation on network create failure."""
+ profile_obj = self._make_test_profile('test_np',
+ c_const.NETWORK_TYPE_OVERLAY,
+ '10000-10001')
+ data = self._prepare_net_data(profile_obj.id)
+ client_patch = mock.patch(n1kv_client.__name__ + ".Client",
+ new=fake_client.TestClientInvalidResponse)
+ client_patch.start()
+ self.new_create_request('networks', data)
+ db_session = db.get_session()
+ self.assertFalse(n1kv_db_v2.get_vxlan_allocation(db_session,
+ 10000).allocated)
+
class TestN1kvSubnets(test_plugin.TestSubnetsV2,
N1kvPluginTestCase):
Code Review / openstack-build / neutron-build.git / commitdiff