VPNaaS: Cisco fix validation for GW IP
The validation to check that the router has a GW IP for the Cisco service driver was trying to directly get the gw_port information, which is not available, as the validator only has a dict of the vpn_service info. Modified validator to get the router object (not a dict representation), using the vpn_service's router_id and then check the attributes for the router directly. Change-Id: I48286cda23d9737ddd90251dc0d1db1c310e7784 Closes-Bug: 1356127
This commit is contained in:
parent
2aaf57e06b
commit
4df8a35086
@ -73,9 +73,9 @@ class CiscoCsrVpnValidator(vpn_validator.VpnReferenceValidator):
|
||||
key='mtu',
|
||||
value=mtu)
|
||||
|
||||
def validate_public_ip_present(self, vpn_service):
|
||||
def validate_public_ip_present(self, router):
|
||||
"""Ensure there is one gateway IP specified for the router used."""
|
||||
gw_port = vpn_service.router.gw_port
|
||||
gw_port = router.gw_port
|
||||
if not gw_port or len(gw_port.fixed_ips) != 1:
|
||||
raise CsrValidationFailure(resource='IPSec Connection',
|
||||
key='router:gw_port:ip_address',
|
||||
@ -106,11 +106,11 @@ class CiscoCsrVpnValidator(vpn_validator.VpnReferenceValidator):
|
||||
context, ipsec_sitecon['ipsecpolicy_id'])
|
||||
vpn_service = self.service_plugin.get_vpnservice(
|
||||
context, ipsec_sitecon['vpnservice_id'])
|
||||
router = self.l3_plugin._get_router(context, vpn_service['router_id'])
|
||||
self.validate_lifetime('IKE Policy', ike_policy)
|
||||
self.validate_lifetime('IPSec Policy', ipsec_policy)
|
||||
self.validate_ike_version(ike_policy)
|
||||
self.validate_mtu(ipsec_sitecon)
|
||||
self.validate_public_ip_present(vpn_service)
|
||||
self.validate_public_ip_present(router)
|
||||
self.validate_peer_id(ipsec_sitecon)
|
||||
LOG.debug("IPSec connection %s validated for Cisco CSR",
|
||||
ipsec_sitecon['id'])
|
||||
LOG.debug("IPSec connection validated for Cisco CSR")
|
||||
|
@ -83,7 +83,8 @@ class TestCiscoIPsecDriverValidation(base.BaseTestCase):
|
||||
mock.patch('neutron.manager.NeutronManager.get_plugin',
|
||||
return_value=self.core_plugin).start()
|
||||
self.context = n_ctx.Context('some_user', 'some_tenant')
|
||||
self.vpn_service = mock.Mock()
|
||||
self.vpn_service = {'router_id': '123'}
|
||||
self.router = mock.Mock()
|
||||
self.service_plugin = mock.Mock()
|
||||
self.validator = validator.CiscoCsrVpnValidator(self.service_plugin)
|
||||
|
||||
@ -167,23 +168,25 @@ class TestCiscoIPsecDriverValidation(base.BaseTestCase):
|
||||
"""Helper function indicating that tunnel has a gateway IP."""
|
||||
def have_one():
|
||||
return 1
|
||||
self.vpn_service.router.gw_port.fixed_ips.__len__ = have_one
|
||||
self.router.gw_port.fixed_ips.__len__ = have_one
|
||||
ip_addr_mock = mock.Mock()
|
||||
self.vpn_service.router.gw_port.fixed_ips = [ip_addr_mock]
|
||||
return ip_addr_mock
|
||||
self.router.gw_port.fixed_ips = [ip_addr_mock]
|
||||
|
||||
def test_have_public_ip_for_router(self):
|
||||
"""Ensure that router for IPSec connection has gateway IP."""
|
||||
self.simulate_gw_ip_available()
|
||||
self.validator.validate_public_ip_present(self.vpn_service)
|
||||
try:
|
||||
self.validator.validate_public_ip_present(self.router)
|
||||
except Exception:
|
||||
self.fail("Unexpected exception on validation")
|
||||
|
||||
def test_router_with_missing_gateway_ip(self):
|
||||
"""Failure test of IPSec connection with missing gateway IP."""
|
||||
self.simulate_gw_ip_available()
|
||||
self.vpn_service.router.gw_port = None
|
||||
self.router.gw_port = None
|
||||
self.assertRaises(validator.CsrValidationFailure,
|
||||
self.validator.validate_public_ip_present,
|
||||
self.vpn_service)
|
||||
self.router)
|
||||
|
||||
def test_peer_id_is_an_ip_address(self):
|
||||
"""Ensure peer ID is an IP address for IPsec connection create."""
|
||||
@ -206,6 +209,7 @@ class TestCiscoIPsecDriverValidation(base.BaseTestCase):
|
||||
return_value={'lifetime': {'units': 'seconds', 'value': 120}})
|
||||
self.service_plugin.get_vpnservice = mock.Mock(
|
||||
return_value=self.vpn_service)
|
||||
self.l3_plugin._get_router = mock.Mock(return_value=self.router)
|
||||
# Provide the minimum needed items to validate
|
||||
ipsec_sitecon = {'id': '1',
|
||||
'vpnservice_id': FAKE_SERVICE_ID,
|
||||
|
Loading…
Reference in New Issue
Block a user