openstack
/
neutron-vpnaas
Code Issues Proposed changes

Merge "Check for VPN Objects when deleting interfaces"

This commit is contained in:
Jenkins 2014-10-21 17:37:57 +00:00 committed by Gerrit Code Review
parent 87722a7d1d fb96cb1eef
commit 9fc08d9095
2 changed files with 59 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
neutron/db/vpn/vpn_db.py
View File

@ -602,6 +602,16 @@ class VPNPluginDb(vpnaas.VPNPluginBase, base_db.CommonDbMixin):
router_id=router_id,
vpnservice_id=vpnservices[0]['id'])
def check_subnet_in_use(self, context, subnet_id):
with context.session.begin(subtransactions=True):
vpnservices = context.session.query(VPNService).filter_by(
subnet_id=subnet_id
).first()
if vpnservices:
raise vpnaas.SubnetInUseByVPNService(
subnet_id=subnet_id,
vpnservice_id=vpnservices['id'])
class VPNPluginRpcDbMixin():
def _get_agent_hosting_vpn_services(self, context, host):

49
neutron/tests/unit/db/vpn/test_db_vpnaas.py
View File

@ -867,6 +867,55 @@ class TestVpnaas(VPNPluginDbTestCase):
if k in expected),
expected)
def test_delete_router_interface_in_use_by_vpnservice(self):
"""Test delete router interface in use by vpn service."""
with self.subnet(cidr='10.2.0.0/24') as subnet:
with self.router() as router:
with self.vpnservice(subnet=subnet,
router=router):
self._router_interface_action('remove',
router['router']['id'],
subnet['subnet']['id'],
None,
expected_code=webob.exc.
HTTPConflict.code)
def test_delete_external_gateway_interface_in_use_by_vpnservice(self):
"""Test delete external gateway interface in use by vpn service."""
with self.subnet(cidr='10.2.0.0/24') as subnet:
with self.router() as router:
with self.subnet(cidr='11.0.0.0/24') as public_sub:
self._set_net_external(
public_sub['subnet']['network_id'])
self._add_external_gateway_to_router(
router['router']['id'],
public_sub['subnet']['network_id'])
with self.vpnservice(subnet=subnet,
router=router):
self._remove_external_gateway_from_router(
router['router']['id'],
public_sub['subnet']['network_id'],
expected_code=webob.exc.HTTPConflict.code)
def test_router_update_after_ipsec_site_connection(self):
"""Test case to update router after vpn connection."""
rname1 = "router_one"
rname2 = "router_two"
with self.subnet(cidr='10.2.0.0/24') as subnet:
with self.router(name=rname1) as r:
with self.vpnservice(subnet=subnet,
router=r
) as vpnservice:
self.ipsec_site_connection(
name='connection1', vpnservice=vpnservice
)
body = self._show('routers', r['router']['id'])
self.assertEqual(body['router']['name'], rname1)
body = self._update('routers', r['router']['id'],
{'router': {'name': rname2}})
body = self._show('routers', r['router']['id'])
self.assertEqual(body['router']['name'], rname2)
def test_update_vpnservice(self):
"""Test case to update a vpnservice."""
name = 'new_vpnservice1'