1
# Copyright 2013, Nachi Ueno, NTT I3, Inc.
4
# Licensed under the Apache License, Version 2.0 (the "License"); you may
5
# not use this file except in compliance with the License. You may obtain
6
# a copy of the License at
8
# http://www.apache.org/licenses/LICENSE-2.0
10
# Unless required by applicable law or agreed to in writing, software
11
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13
# License for the specific language governing permissions and limitations
18
from neutron.common import constants
19
from neutron import context
20
from neutron import manager
21
from neutron.plugins.common import constants as p_constants
22
from neutron.tests.unit.openvswitch import test_agent_scheduler
23
from neutron.tests.unit import test_agent_ext_plugin
25
from neutron_vpnaas.db.vpn import vpn_validator
26
from neutron_vpnaas.services.vpn.service_drivers import ipsec as ipsec_driver
27
from neutron_vpnaas.tests.unit.db.vpn import test_db_vpnaas
29
FAKE_HOST = test_agent_ext_plugin.L3_HOSTA
30
VPN_DRIVER_CLASS = 'neutron_vpnaas.services.vpn.plugin.VPNDriverPlugin'
33
class TestVPNDriverPlugin(test_db_vpnaas.TestVpnaas,
34
test_agent_scheduler.AgentSchedulerTestMixIn,
35
test_agent_ext_plugin.AgentDBTestMixIn):
38
driver_cls_p = mock.patch(
39
'neutron_vpnaas.services.vpn.'
40
'service_drivers.ipsec.IPsecVPNDriver')
41
driver_cls = driver_cls_p.start()
42
self.driver = mock.Mock()
43
self.driver.service_type = ipsec_driver.IPSEC
44
self.driver.validator = vpn_validator.VpnReferenceValidator()
45
driver_cls.return_value = self.driver
46
super(TestVPNDriverPlugin, self).setUp(
47
vpnaas_plugin=VPN_DRIVER_CLASS)
48
# Note: Context must be created after BaseTestCase.setUp() so that
49
# config for policy is set.
50
self.adminContext = context.get_admin_context()
52
def test_create_ipsec_site_connection(self, **extras):
53
super(TestVPNDriverPlugin, self).test_create_ipsec_site_connection()
54
self.driver.create_ipsec_site_connection.assert_called_once_with(
56
self.driver.delete_ipsec_site_connection.assert_called_once_with(
59
def test_delete_vpnservice(self, **extras):
60
super(TestVPNDriverPlugin, self).test_delete_vpnservice()
61
self.driver.delete_vpnservice.assert_called_once_with(
64
def test_update_vpnservice(self, **extras):
65
super(TestVPNDriverPlugin, self).test_update_vpnservice()
66
self.driver.update_vpnservice.assert_called_once_with(
67
mock.ANY, mock.ANY, mock.ANY)
69
@contextlib.contextmanager
70
def vpnservice_set(self):
71
"""Test case to create a ipsec_site_connection."""
72
vpnservice_name = "vpn1"
73
ipsec_site_connection_name = "ipsec_site_connection"
74
ikename = "ikepolicy1"
75
ipsecname = "ipsecpolicy1"
76
description = "my-vpn-connection"
77
keys = {'name': vpnservice_name,
78
'description': "my-vpn-connection",
79
'peer_address': '192.168.1.10',
80
'peer_id': '192.168.1.10',
81
'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'],
82
'initiator': 'bi-directional',
87
'tenant_id': self._tenant_id,
89
'status': 'PENDING_CREATE',
90
'admin_state_up': True}
91
with self.ikepolicy(name=ikename) as ikepolicy:
92
with self.ipsecpolicy(name=ipsecname) as ipsecpolicy:
93
with self.subnet() as subnet:
94
with self.router() as router:
95
plugin = manager.NeutronManager.get_plugin()
96
agent = {'host': FAKE_HOST,
97
'agent_type': constants.AGENT_TYPE_L3,
98
'binary': 'fake-binary',
99
'topic': 'fake-topic'}
100
plugin.create_or_update_agent(self.adminContext, agent)
101
plugin.schedule_router(
102
self.adminContext, router['router']['id'])
103
with self.vpnservice(name=vpnservice_name,
105
router=router) as vpnservice1:
106
keys['ikepolicy_id'] = ikepolicy['ikepolicy']['id']
107
keys['ipsecpolicy_id'] = (
108
ipsecpolicy['ipsecpolicy']['id']
110
keys['vpnservice_id'] = (
111
vpnservice1['vpnservice']['id']
113
with self.ipsec_site_connection(
115
ipsec_site_connection_name,
116
keys['peer_address'],
123
keys['dpd_interval'],
128
keys['admin_state_up'],
129
description=description,
131
yield vpnservice1['vpnservice']
133
def test_get_agent_hosting_vpn_services(self):
134
with self.vpnservice_set():
135
service_plugin = manager.NeutronManager.get_service_plugins()[
137
vpnservices = service_plugin._get_agent_hosting_vpn_services(
138
self.adminContext, FAKE_HOST)
139
vpnservices = vpnservices.all()
140
self.assertEqual(1, len(vpnservices))
141
vpnservice_db = vpnservices[0]
142
self.assertEqual(1, len(vpnservice_db.ipsec_site_connections))
143
ipsec_site_connection = vpnservice_db.ipsec_site_connections[0]
144
self.assertIsNotNone(
145
ipsec_site_connection['ikepolicy'])
146
self.assertIsNotNone(
147
ipsec_site_connection['ipsecpolicy'])
149
def test_update_status(self):
150
with self.vpnservice_set() as vpnservice:
151
self._register_agent_states()
152
service_plugin = manager.NeutronManager.get_service_plugins()[
154
service_plugin.update_status_by_agent(
156
[{'status': 'ACTIVE',
157
'ipsec_site_connections': {},
158
'updated_pending_status': True,
159
'id': vpnservice['id']}])
160
vpnservices = service_plugin._get_agent_hosting_vpn_services(
161
self.adminContext, FAKE_HOST)
162
vpnservice_db = vpnservices[0]
163
self.assertEqual(p_constants.ACTIVE, vpnservice_db['status'])
added
removed
neutron_vpnaas/tests/unit/services/vpn/test_vpnaas_driver_plugin.py
