openstack/neutron-vpnaas
Code Issues Proposed changes

Merge "Cisco VPN device driver - support IPSec connection updates"

Browse Source
This commit is contained in:
Jenkins 2014-05-02 20:55:56 +00:00 committed by Gerrit Code Review
commit 1e0f4389f3
7 changed files with 312 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
neutron/services/vpn/device_drivers/cisco_csr_rest_client.py
View File

@ -214,9 +214,6 @@ class CsrRestClient(object):
base_conn_info = {u'vpn-type': u'site-to-site',
u'ip-version': u'ipv4'}
connection_info.update(base_conn_info)
# TODO(pcm) pass in value, when CSR is embedded as Neutron router.
# Currently, get this from .INI file.
connection_info[u'local-device'][u'tunnel-ip-address'] = self.tunnel_ip
return self.post_request('vpn-svc/site-to-site',
payload=connection_info)
@ -232,6 +229,14 @@ class CsrRestClient(object):
def delete_static_route(self, route_id):
return self.delete_request('routing-svc/static-routes/%s' % route_id)
def set_ipsec_connection_state(self, tunnel, admin_up=True):
"""Set the IPSec site-to-site connection (tunnel) admin state.
Note: When a tunnel is created, it will be admin up.
"""
info = {u'vpn-interface-name': tunnel, u'enabled': admin_up}
return self.put_request('vpn-svc/site-to-site/%s/state' % tunnel, info)
def delete_ipsec_connection(self, conn_id):
return self.delete_request('vpn-svc/site-to-site/%s' % conn_id)

66
neutron/services/vpn/device_drivers/cisco_ipsec.py
View File

@ -54,6 +54,11 @@ class CsrResourceCreateFailure(exceptions.NeutronException):
message = _("Cisco CSR failed to create %(resource)s (%(which)s)")
class CsrAdminStateChangeFailure(exceptions.NeutronException):
message = _("Cisco CSR failed to change %(tunnel)s admin state to "
"%(state)s")
class CsrDriverMismatchError(exceptions.NeutronException):
message = _("Required %(resource)s attribute %(attr)s mapping for Cisco "
"CSR is missing in device driver")
@ -240,36 +245,37 @@ class CiscoCsrIPsecDriver(device_drivers.DeviceDriver):
conn_id = conn_data['id']
conn_is_admin_up = conn_data[u'admin_state_up']
if conn_id in vpn_service.conn_state:
if conn_id in vpn_service.conn_state: # Existing connection...
ipsec_conn = vpn_service.conn_state[conn_id]
config_changed = ipsec_conn.check_for_changes(conn_data)
if config_changed:
LOG.debug(_("Update: Existing connection %s changed"), conn_id)
ipsec_conn.delete_ipsec_site_connection(context, conn_id)
ipsec_conn.create_ipsec_site_connection(context, conn_data)
ipsec_conn.conn_info = conn_data
if ipsec_conn.forced_down:
if vpn_service.is_admin_up and conn_is_admin_up:
LOG.debug(_("Update: Connection %s no longer admin down"),
conn_id)
# TODO(pcm) Do no shut on tunnel, once CSR supports
ipsec_conn.set_admin_state(is_up=True)
ipsec_conn.forced_down = False
ipsec_conn.create_ipsec_site_connection(context, conn_data)
else:
if not vpn_service.is_admin_up or not conn_is_admin_up:
LOG.debug(_("Update: Connection %s forced to admin down"),
conn_id)
# TODO(pcm) Do shut on tunnel, once CSR supports
ipsec_conn.set_admin_state(is_up=False)
ipsec_conn.forced_down = True
ipsec_conn.delete_ipsec_site_connection(context, conn_id)
else:
# TODO(pcm) FUTURE handle connection update
LOG.debug(_("Update: Ignoring existing connection %s"),
conn_id)
else: # New connection...
ipsec_conn = vpn_service.create_connection(conn_data)
ipsec_conn.create_ipsec_site_connection(context, conn_data)
if not vpn_service.is_admin_up or not conn_is_admin_up:
# TODO(pcm) Create, but set tunnel down, once CSR supports
LOG.debug(_("Update: Created new connection %s in admin down "
"state"), conn_id)
ipsec_conn.set_admin_state(is_up=False)
ipsec_conn.forced_down = True
else:
LOG.debug(_("Update: Created new connection %s"), conn_id)
ipsec_conn.create_ipsec_site_connection(context, conn_data)
ipsec_conn.is_dirty = False
ipsec_conn.last_status = conn_data['status']
@ -539,12 +545,33 @@ class CiscoCsrIPSecConnection(object):
"""State and actions for IPSec site-to-site connections."""
def __init__(self, conn_info, csr):
self.conn_id = conn_info['id']
self.conn_info = conn_info
self.csr = csr
self.steps = []
self.forced_down = False
self.is_admin_up = conn_info[u'admin_state_up']
self.tunnel = conn_info['cisco']['site_conn_id']
self.changed = False
@property
def conn_id(self):
return self.conn_info['id']
@property
def is_admin_up(self):
return self.conn_info['admin_state_up']
@is_admin_up.setter
def is_admin_up(self, is_up):
self.conn_info['admin_state_up'] = is_up
@property
def tunnel(self):
return self.conn_info['cisco']['site_conn_id']
def check_for_changes(self, curr_conn):
return not all([self.conn_info[attr] == curr_conn[attr]
for attr in ('mtu', 'psk', 'peer_address',
'peer_cidrs', 'ike_policy',
'ipsec_policy', 'cisco')])
def find_current_status_in(self, statuses):
if self.tunnel in statuses:
@ -683,7 +710,7 @@ class CiscoCsrIPSecConnection(object):
u'ip-address': u'GigabitEthernet3',
# TODO(pcm): FUTURE - Get IP address of router's public
# I/F, once CSR is used as embedded router.
u'tunnel-ip-address': u'172.24.4.23'
u'tunnel-ip-address': self.csr.tunnel_ip
# u'tunnel-ip-address': u'%s' % gw_ip
},
u'remote-device': {
@ -822,3 +849,12 @@ class CiscoCsrIPSecConnection(object):
LOG.info(_("SUCCESS: Deleted IPSec site-to-site connection %s"),
conn_id)
def set_admin_state(self, is_up):
"""Change the admin state for the IPSec connection."""
self.csr.set_ipsec_connection_state(self.tunnel, admin_up=is_up)
if self.csr.status != requests.codes.NO_CONTENT:
state = "UP" if is_up else "DOWN"
LOG.error(_("Unable to change %(tunnel)s admin state to "
"%(state)s"), {'tunnel': self.tunnel, 'state': state})
raise CsrAdminStateChangeFailure(tunnel=self.tunnel, state=state)

12
neutron/services/vpn/service_drivers/cisco_ipsec.py
View File

@ -41,10 +41,6 @@ class CsrValidationFailure(exceptions.BadRequest):
"with value '%(value)s'")
class CsrUnsupportedError(exceptions.NeutronException):
message = _("Cisco CSR does not currently support %(capability)s")
class CiscoCsrIPsecVpnDriverCallBack(object):
"""Handler for agent to plugin RPC messaging."""
@ -184,9 +180,11 @@ class CiscoCsrIPsecVPNDriver(service_drivers.VpnDriver):
def update_ipsec_site_connection(
self, context, old_ipsec_site_connection, ipsec_site_connection):
capability = _("update of IPSec connections. You can delete and "
"re-add, as a workaround.")
raise CsrUnsupportedError(capability=capability)
vpnservice = self.service_plugin._get_vpnservice(
context, ipsec_site_connection['vpnservice_id'])
self.agent_rpc.vpnservice_updated(
context, vpnservice['router_id'],
reason='ipsec-conn-update')
def delete_ipsec_site_connection(self, context, ipsec_site_connection):
vpnservice = self.service_plugin._get_vpnservice(

28
neutron/tests/unit/services/vpn/device_drivers/cisco_csr_mock.py
View File

@ -331,6 +331,34 @@ def get_unnumbered(url, request):
return httmock.response(requests.codes.OK, content=content)
@filter_request(['get'], 'vpn-svc/site-to-site/Tunnel')
@httmock.urlmatch(netloc=r'localhost')
def get_admin_down(url, request):
if not request.headers.get('X-auth-token', None):
return {'status_code': requests.codes.UNAUTHORIZED}
# URI has .../Tunnel#/state, so get number from 2nd to last element
tunnel = url.path.split('/')[-2]
content = {u'kind': u'object#vpn-site-to-site-state',
u'vpn-interface-name': u'%s' % tunnel,
u'line-protocol-state': u'down',
u'enabled': False}
return httmock.response(requests.codes.OK, content=content)
@filter_request(['get'], 'vpn-svc/site-to-site/Tunnel')
@httmock.urlmatch(netloc=r'localhost')
def get_admin_up(url, request):
if not request.headers.get('X-auth-token', None):
return {'status_code': requests.codes.UNAUTHORIZED}
# URI has .../Tunnel#/state, so get number from 2nd to last element
tunnel = url.path.split('/')[-2]
content = {u'kind': u'object#vpn-site-to-site-state',
u'vpn-interface-name': u'%s' % tunnel,
u'line-protocol-state': u'down',
u'enabled': True}
return httmock.response(requests.codes.OK, content=content)
@filter_request(['get'], 'vpn-svc/site-to-site')
@httmock.urlmatch(netloc=r'localhost')
def get_mtu(url, request):

41
neutron/tests/unit/services/vpn/device_drivers/notest_cisco_csr_rest.py
View File

@ -1012,6 +1012,47 @@ class TestCsrRestIPSecConnectionCreate(base.BaseTestCase):
expected_connection.update(connection_info)
self.assertEqual(expected_connection, content)
def test_set_ipsec_connection_admin_state_changes(self):
"""Create IPSec connection in admin down state."""
tunnel_id, ipsec_policy_id = self._prepare_for_site_conn_create()
tunnel = u'Tunnel%d' % tunnel_id
with httmock.HTTMock(csr_request.token, csr_request.post):
connection_info = {
u'vpn-interface-name': tunnel,
u'ipsec-policy-id': u'%d' % ipsec_policy_id,
u'mtu': 1500,
u'local-device': {u'ip-address': u'10.3.0.1/24',
u'tunnel-ip-address': u'10.10.10.10'},
u'remote-device': {u'tunnel-ip-address': u'10.10.10.20'}
}
location = self.csr.create_ipsec_connection(connection_info)
self.addCleanup(self._remove_resource_for_test,
self.csr.delete_ipsec_connection,
tunnel)
self.assertEqual(requests.codes.CREATED, self.csr.status)
self.assertIn('vpn-svc/site-to-site/%s' % tunnel, location)
state_uri = location + "/state"
# Note: When created, the tunnel will be in admin 'up' state
# Note: Line protocol state will be down, unless have an active conn.
expected_state = {u'kind': u'object#vpn-site-to-site-state',
u'vpn-interface-name': tunnel,
u'line-protocol-state': u'down',
u'enabled': False}
with httmock.HTTMock(csr_request.put, csr_request.get_admin_down):
self.csr.set_ipsec_connection_state(tunnel, admin_up=False)
self.assertEqual(requests.codes.NO_CONTENT, self.csr.status)
content = self.csr.get_request(state_uri, full_url=True)
self.assertEqual(requests.codes.OK, self.csr.status)
self.assertEqual(expected_state, content)
with httmock.HTTMock(csr_request.put, csr_request.get_admin_up):
self.csr.set_ipsec_connection_state(tunnel, admin_up=True)
self.assertEqual(requests.codes.NO_CONTENT, self.csr.status)
content = self.csr.get_request(state_uri, full_url=True)
self.assertEqual(requests.codes.OK, self.csr.status)
expected_state[u'enabled'] = True
self.assertEqual(expected_state, content)
def test_create_ipsec_connection_missing_ipsec_policy(self):
"""Negative test of connection create without IPSec policy."""
tunnel_id, ipsec_policy_id = self._prepare_for_site_conn_create(

268
neutron/tests/unit/services/vpn/device_drivers/test_cisco_ipsec.py
View File

@ -14,6 +14,7 @@
#
# @author: Paul Michali, Cisco Systems, Inc.
import copy
import httplib
import os
import tempfile
@ -78,6 +79,7 @@ class TestCiscoCsrIPSecConnection(base.BaseTestCase):
}
self.csr = mock.Mock(spec=csr_client.CsrRestClient)
self.csr.status = 201 # All calls to CSR REST API succeed
self.csr.tunnel_ip = '172.24.4.23'
self.ipsec_conn = ipsec_driver.CiscoCsrIPSecConnection(self.conn_info,
self.csr)
@ -219,8 +221,10 @@ class TestCiscoCsrIPsecConnectionCreateTransforms(base.BaseTestCase):
# TODO(pcm) get from vpnservice['external_ip']
'router_public_ip': '172.24.4.23'}
}
self.csr = mock.Mock(spec=csr_client.CsrRestClient)
self.csr.tunnel_ip = '172.24.4.23'
self.ipsec_conn = ipsec_driver.CiscoCsrIPSecConnection(self.conn_info,
mock.Mock())
self.csr)
def test_invalid_attribute(self):
"""Negative test of unknown attribute - programming error."""
@ -360,7 +364,7 @@ class TestCiscoCsrIPsecConnectionCreateTransforms(base.BaseTestCase):
u'ipsec-policy-id': 333,
u'local-device': {
u'ip-address': u'GigabitEthernet3',
u'tunnel-ip-address': u'172.24.4.23'
u'tunnel-ip-address': '172.24.4.23'
},
u'remote-device': {
u'tunnel-ip-address': '192.168.1.2'
@ -418,14 +422,36 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
self.conn_delete = mock.patch.object(
ipsec_driver.CiscoCsrIPSecConnection,
'delete_ipsec_site_connection').start()
self.admin_state = mock.patch.object(
ipsec_driver.CiscoCsrIPSecConnection,
'set_admin_state').start()
self.csr = mock.Mock()
self.driver.csrs['1.1.1.1'] = self.csr
self.service123_data = {u'id': u'123',
u'status': constants.DOWN,
u'admin_state_up': False,
u'external_ip': u'1.1.1.1'}
self.conn1_data = {u'id': u'1', u'status': constants.ACTIVE,
self.conn1_data = {u'id': u'1',
u'status': constants.ACTIVE,
u'admin_state_up': True,
u'mtu': 1500,
u'psk': u'secret',
u'peer_address': '192.168.1.2',
u'peer_cidrs': ['10.1.0.0/24', '10.2.0.0/24'],
u'ike_policy': {
u'auth_algorithm': u'sha1',
u'encryption_algorithm': u'aes-128',
u'pfs': u'Group5',
u'ike_version': u'v1',
u'lifetime_units': u'seconds',
u'lifetime_value': 3600},
u'ipsec_policy': {
u'transform_protocol': u'ah',
u'encryption_algorithm': u'aes-128',
u'auth_algorithm': u'sha1',
u'pfs': u'group5',
u'lifetime_units': u'seconds',
u'lifetime_value': 3600},
u'cisco': {u'site_conn_id': u'Tunnel0'}}
# NOTE: For sync, there is mark (trivial), update (tested),
@ -435,9 +461,8 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
"""Notified of connection create request - create."""
# Make the (existing) service
self.driver.create_vpn_service(self.service123_data)
conn_data = {u'id': u'1', u'status': constants.PENDING_CREATE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'status'] = constants.PENDING_CREATE
connection = self.driver.update_connection(self.context,
u'123', conn_data)
@ -446,17 +471,50 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
self.assertEqual(constants.PENDING_CREATE, connection.last_status)
self.assertEqual(1, self.conn_create.call_count)
def test_update_ipsec_connection_changed_settings(self):
"""Notified of connection changing config - update."""
# TODO(pcm) Place holder for this condition
# Make the (existing) service and connection
def test_detect_no_change_to_ipsec_connection(self):
"""No change to IPSec connection - nop."""
# Make existing service, and connection that was active
vpn_service = self.driver.create_vpn_service(self.service123_data)
# TODO(pcm) add info that indicates that the connection has changed
conn_data = {u'id': u'1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
vpn_service.create_connection(conn_data)
connection = vpn_service.create_connection(self.conn1_data)
self.assertFalse(connection.check_for_changes(self.conn1_data))
def test_detect_state_only_change_to_ipsec_connection(self):
"""Only IPSec connection state changed - update."""
# Make existing service, and connection that was active
vpn_service = self.driver.create_vpn_service(self.service123_data)
connection = vpn_service.create_connection(self.conn1_data)
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'admin_state_up'] = False
self.assertFalse(connection.check_for_changes(conn_data))
def test_detect_non_state_change_to_ipsec_connection(self):
"""Connection change instead of/in addition to state - update."""
# Make existing service, and connection that was active
vpn_service = self.driver.create_vpn_service(self.service123_data)
connection = vpn_service.create_connection(self.conn1_data)
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'ipsec_policy'][u'encryption_algorithm'] = u'aes-256'
self.assertTrue(connection.check_for_changes(conn_data))
def test_update_ipsec_connection_changed_admin_down(self):
"""Notified of connection state change - update.
For a connection that was previously created, expect to
force connection down on an admin down (only) change.
"""
# Make existing service, and connection that was active
vpn_service = self.driver.create_vpn_service(self.service123_data)
connection = vpn_service.create_connection(self.conn1_data)
# Simulate that notification of connection update received
self.driver.mark_existing_connections_as_dirty()
# Modify the connection data for the 'sync'
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'admin_state_up'] = False
connection = self.driver.update_connection(self.context,
'123', conn_data)
@ -464,7 +522,37 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.ACTIVE, connection.last_status)
self.assertFalse(self.conn_create.called)
# TODO(pcm) FUTURE - handling for update (delete/create?)
self.assertFalse(connection.is_admin_up)
self.assertTrue(connection.forced_down)
self.assertEqual(1, self.admin_state.call_count)
def test_update_ipsec_connection_changed_config(self):
"""Notified of connection changing config - update.
Goal here is to detect that the connection is deleted and then
created, but not that the specific values have changed, so picking
arbitrary value (MTU).
"""
# Make existing service, and connection that was active
vpn_service = self.driver.create_vpn_service(self.service123_data)
connection = vpn_service.create_connection(self.conn1_data)
# Simulate that notification of connection update received
self.driver.mark_existing_connections_as_dirty()
# Modify the connection data for the 'sync'
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'mtu'] = 9200
connection = self.driver.update_connection(self.context,
'123', conn_data)
self.assertFalse(connection.is_dirty)
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.ACTIVE, connection.last_status)
self.assertEqual(1, self.conn_create.call_count)
self.assertEqual(1, self.conn_delete.call_count)
self.assertTrue(connection.is_admin_up)
self.assertFalse(connection.forced_down)
self.assertFalse(self.admin_state.called)
def test_update_of_unknown_ipsec_connection(self):
"""Notified of update of unknown connection - create.
@ -472,15 +560,14 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
Occurs if agent restarts and receives a notification of change
to connection, but has no previous record of the connection.
Result will be to rebuild the connection.
This can also happen, if a connection is changed from admin
down to admin up (so don't need a separate test for admin up.
"""
# Will have previously created service, but don't know of connection
self.driver.create_vpn_service(self.service123_data)
conn_data = {u'id': u'1', u'status': constants.DOWN,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
# Simulate that notification of connection update received
self.driver.mark_existing_connections_as_dirty()
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'status'] = constants.DOWN
connection = self.driver.update_connection(self.context,
u'123', conn_data)
@ -488,91 +575,58 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.DOWN, connection.last_status)
self.assertEqual(1, self.conn_create.call_count)
def test_update_unchanged_ipsec_connection(self):
"""Unchanged state for connection during sync - nop."""
# Make the (existing) service and connection
vpn_service = self.driver.create_vpn_service(self.service123_data)
conn_data = {u'id': u'1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
vpn_service.create_connection(conn_data)
self.driver.mark_existing_connections_as_dirty()
# The notification (state) hasn't changed for the connection
connection = self.driver.update_connection(self.context,
'123', conn_data)
self.assertFalse(connection.is_dirty)
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.ACTIVE, connection.last_status)
self.assertFalse(self.conn_create.called)
def test_update_connection_admin_down(self):
"""Connection updated to admin down state - force down."""
# Make existing service, and connection that was active
vpn_service = self.driver.create_vpn_service(self.service123_data)
conn_data = {u'id': '1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
vpn_service.create_connection(conn_data)
self.driver.mark_existing_connections_as_dirty()
# Now simulate that the notification shows the connection admin down
conn_data[u'admin_state_up'] = False
conn_data[u'status'] = constants.DOWN
connection = self.driver.update_connection(self.context,
u'123', conn_data)
self.assertFalse(connection.is_dirty)
self.assertTrue(connection.forced_down)
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.DOWN, connection.last_status)
self.assertFalse(self.conn_create.called)
self.assertTrue(connection.is_admin_up)
self.assertFalse(connection.forced_down)
self.assertFalse(self.admin_state.called)
def test_update_missing_connection_admin_down(self):
"""Connection not present is in admin down state - nop.
If the agent has restarted, and a sync notification occurs with
a connection that is in admin down state, create the structures,
a connection that is in admin down state, recreate the connection,
but indicate that the connection is down.
"""
# Make existing service, but no connection
self.driver.create_vpn_service(self.service123_data)
conn_data = {u'id': '1', u'status': constants.DOWN,
u'admin_state_up': False,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
conn_data = copy.deepcopy(self.conn1_data)
conn_data.update({u'status': constants.DOWN,
u'admin_state_up': False})
connection = self.driver.update_connection(self.context,
u'123', conn_data)
self.assertIsNotNone(connection)
self.assertFalse(connection.is_dirty)
self.assertEqual(1, self.conn_create.call_count)
self.assertFalse(connection.is_admin_up)
self.assertTrue(connection.forced_down)
self.assertFalse(self.conn_create.called)
self.assertEqual(1, self.admin_state.call_count)
def test_update_connection_admin_up(self):
"""Connection updated to admin up state - record."""
# Make existing service, and connection that was admin down
conn_data = {u'id': '1', u'status': constants.DOWN,
u'admin_state_up': False,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
conn_data = copy.deepcopy(self.conn1_data)
conn_data.update({u'status': constants.DOWN, u'admin_state_up': False})
service_data = {u'id': u'123',
u'status': constants.DOWN,
u'external_ip': u'1.1.1.1',
u'admin_state_up': True,
u'ipsec_conns': [conn_data]}
self.driver.update_service(self.context, service_data)
# Simulate that notification of connection update received
self.driver.mark_existing_connections_as_dirty()
# Now simulate that the notification shows the connection admin up
conn_data[u'admin_state_up'] = True
conn_data[u'status'] = constants.DOWN
new_conn_data = copy.deepcopy(conn_data)
new_conn_data[u'admin_state_up'] = True
connection = self.driver.update_connection(self.context,
u'123', conn_data)
u'123', new_conn_data)
self.assertFalse(connection.is_dirty)
self.assertFalse(connection.forced_down)
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.DOWN, connection.last_status)
self.assertEqual(1, self.conn_create.call_count)
self.assertTrue(connection.is_admin_up)
self.assertFalse(connection.forced_down)
self.assertEqual(2, self.admin_state.call_count)
def test_update_for_vpn_service_create(self):
"""Creation of new IPSec connection on new VPN service - create.
@ -580,9 +634,8 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
Service will be created and marked as 'clean', and update
processing for connection will occur (create).
"""
conn_data = {u'id': u'1', u'status': constants.PENDING_CREATE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'status'] = constants.PENDING_CREATE
service_data = {u'id': u'123',
u'status': constants.PENDING_CREATE,
u'external_ip': u'1.1.1.1',
@ -597,15 +650,17 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
self.assertEqual(u'Tunnel0', connection.tunnel)
self.assertEqual(constants.PENDING_CREATE, connection.last_status)
self.assertEqual(1, self.conn_create.call_count)
self.assertTrue(connection.is_admin_up)
self.assertFalse(connection.forced_down)
self.assertFalse(self.admin_state.called)
def test_update_for_new_connection_on_existing_service(self):
"""Creating a new IPSec connection on an existing service."""
# Create the service before testing, and mark it dirty
prev_vpn_service = self.driver.create_vpn_service(self.service123_data)
self.driver.mark_existing_connections_as_dirty()
conn_data = {u'id': u'1', u'status': constants.PENDING_CREATE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
conn_data = copy.deepcopy(self.conn1_data)
conn_data[u'status'] = constants.PENDING_CREATE
service_data = {u'id': u'123',
u'status': constants.ACTIVE,
u'external_ip': u'1.1.1.1',
@ -631,17 +686,15 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
"""
# Create a service and add in a connection that is active
prev_vpn_service = self.driver.create_vpn_service(self.service123_data)
conn_data = {u'id': u'1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
prev_vpn_service.create_connection(conn_data)
prev_vpn_service.create_connection(self.conn1_data)
self.driver.mark_existing_connections_as_dirty()
# Create notification with conn unchanged and service already created
service_data = {u'id': u'123',
u'status': constants.ACTIVE,
u'external_ip': u'1.1.1.1',
u'admin_state_up': True,
u'ipsec_conns': [conn_data]}
u'ipsec_conns': [self.conn1_data]}
vpn_service = self.driver.update_service(self.context, service_data)
# Should reuse the entry and update the status
self.assertEqual(prev_vpn_service, vpn_service)
@ -661,15 +714,13 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
"""
# Create an "existing" service, prior to notification
prev_vpn_service = self.driver.create_vpn_service(self.service123_data)
self.driver.mark_existing_connections_as_dirty()
conn_data = {u'id': u'1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
service_data = {u'id': u'123',
u'status': constants.DOWN,
u'external_ip': u'1.1.1.1',
u'admin_state_up': False,
u'ipsec_conns': [conn_data]}
u'ipsec_conns': [self.conn1_data]}
vpn_service = self.driver.update_service(self.context, service_data)
self.assertEqual(prev_vpn_service, vpn_service)
self.assertFalse(vpn_service.is_dirty)
@ -688,14 +739,11 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
of a service that is in the admin down state. Structures will be
created, but forced down.
"""
conn_data = {u'id': u'1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'cisco': {u'site_conn_id': u'Tunnel0'}}
service_data = {u'id': u'123',
u'status': constants.DOWN,
u'external_ip': u'1.1.1.1',
u'admin_state_up': False,
u'ipsec_conns': [conn_data]}
u'ipsec_conns': [self.conn1_data]}
vpn_service = self.driver.update_service(self.context, service_data)
self.assertIsNotNone(vpn_service)
self.assertFalse(vpn_service.is_dirty)
@ -888,7 +936,7 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
self.assertEqual(1, self.conn_delete.call_count)
def test_sweep_multiple_services(self):
"""One service and conn udpated, one service and conn not."""
"""One service and conn updated, one service and conn not."""
# Create two services, each with a connection
vpn_service1 = self.driver.create_vpn_service(self.service123_data)
vpn_service1.create_connection(self.conn1_data)
@ -1315,9 +1363,41 @@ class TestCiscoCsrIPsecDeviceDriverSyncStatuses(base.BaseTestCase):
# Simulate one service with one connection up, one down
conn1_data = {u'id': u'1', u'status': constants.ACTIVE,
u'admin_state_up': True,
u'mtu': 1500,
u'psk': u'secret',
u'peer_address': '192.168.1.2',
u'peer_cidrs': ['10.1.0.0/24', '10.2.0.0/24'],
u'ike_policy': {u'auth_algorithm': u'sha1',
u'encryption_algorithm': u'aes-128',
u'pfs': u'Group5',
u'ike_version': u'v1',
u'lifetime_units': u'seconds',
u'lifetime_value': 3600},
u'ipsec_policy': {u'transform_protocol': u'ah',
u'encryption_algorithm': u'aes-128',
u'auth_algorithm': u'sha1',
u'pfs': u'group5',
u'lifetime_units': u'seconds',
u'lifetime_value': 3600},
u'cisco': {u'site_conn_id': u'Tunnel1'}}
conn2_data = {u'id': u'2', u'status': constants.DOWN,
u'admin_state_up': True,
u'mtu': 1500,
u'psk': u'secret',
u'peer_address': '192.168.1.2',
u'peer_cidrs': ['10.1.0.0/24', '10.2.0.0/24'],
u'ike_policy': {u'auth_algorithm': u'sha1',
u'encryption_algorithm': u'aes-128',
u'pfs': u'Group5',
u'ike_version': u'v1',
u'lifetime_units': u'seconds',
u'lifetime_value': 3600},
u'ipsec_policy': {u'transform_protocol': u'ah',
u'encryption_algorithm': u'aes-128',
u'auth_algorithm': u'sha1',
u'pfs': u'group5',
u'lifetime_units': u'seconds',
u'lifetime_value': 3600},
u'cisco': {u'site_conn_id': u'Tunnel2'}}
service_data = {u'id': u'123',
u'status': constants.ACTIVE,

12
neutron/tests/unit/services/vpn/service_drivers/test_cisco_ipsec.py
View File

@ -309,12 +309,12 @@ class TestCiscoIPsecDriver(base.BaseTestCase):
mock.patch.object(csr_db, 'create_tunnel_mapping').start()
self.context = n_ctx.Context('some_user', 'some_tenant')
def _test_update(self, func, args, reason=None):
def _test_update(self, func, args, additional_info=None):
with mock.patch.object(self.driver.agent_rpc, 'cast') as cast:
func(self.context, *args)
cast.assert_called_once_with(
self.context,
{'args': reason,
{'args': additional_info,
'namespace': None,
'method': 'vpnservice_updated'},
version='1.0',
@ -345,11 +345,9 @@ class TestCiscoIPsecDriver(base.BaseTestCase):
constants.ERROR)
def test_update_ipsec_site_connection(self):
# TODO(pcm) FUTURE - Update test, when supported
self.assertRaises(ipsec_driver.CsrUnsupportedError,
self._test_update,
self.driver.update_ipsec_site_connection,
[FAKE_VPN_CONNECTION, FAKE_VPN_CONNECTION])
self._test_update(self.driver.update_ipsec_site_connection,
[FAKE_VPN_CONNECTION, FAKE_VPN_CONNECTION],
{'reason': 'ipsec-conn-update'})
def test_delete_ipsec_site_connection(self):
self._test_update(self.driver.delete_ipsec_site_connection,