openstack/neutron-vpnaas
Code Issues Proposed changes

Merge "VPNaaS: Store local side's tunnel IP for later retrieval"

Browse Source
This commit is contained in:
Jenkins
2015-08-13 18:37:53 +00:00
committed by Gerrit Code Review
parent c1aa06ee96 b6cf05de21
commit 7b6f715ffc
10 changed files with 424 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
neutron_vpnaas/db/migration/alembic_migrations/versions/HEADS
View File

@@ -1,3 +1,3 @@
30018084ed99
56893333aa52
24f28869838b
333dfd6afaa2
kilo

85
neutron_vpnaas/db/migration/alembic_migrations/versions/liberty/contract/333dfd6afaa2_populate_vpn_service_table_fields.py Normal file
View File

@@ -0,0 +1,85 @@
# Copyright 2015 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""Populate VPN service table fields
Revision ID: 333dfd6afaa2
Revises: 56893333aa52
Create Date: 2015-07-27 16:43:59.123456
"""
# revision identifiers, used by Alembic.
revision = '333dfd6afaa2'
down_revision = '56893333aa52'
depends_on = '24f28869838b'
from alembic import op
import netaddr
import sqlalchemy as sa
VPNService = sa.Table('vpnservices', sa.MetaData(),
sa.Column('router_id', sa.String(36), nullable=False),
sa.Column('external_v4_ip', sa.String(16)),
sa.Column('external_v6_ip', sa.String(64)),
sa.Column('id', sa.String(36), nullable=False,
primary_key=True))
Router = sa.Table('routers', sa.MetaData(),
sa.Column('gw_port_id', sa.String(36)),
sa.Column('id', sa.String(36), nullable=False,
primary_key=True))
Port = sa.Table('ports', sa.MetaData(),
sa.Column('id', sa.String(36), nullable=False,
primary_key=True))
IPAllocation = sa.Table('ipallocations', sa.MetaData(),
sa.Column('ip_address', sa.String(64),
nullable=False, primary_key=True),
sa.Column('port_id', sa.String(36)))
def _migrate_external_ips(engine):
"""Use router external IPs to populate external_v*_ip entries.
For each service, look through the associated router's
gw_port['fixed_ips'] list and store any IPv4 and/or IPv6
addresses into the new fields. If there are multiple
addresses for an IP version, then only the first one will
be stored (the same as the reference driver does).
"""
session = sa.orm.Session(bind=engine.connect())
services = session.query(VPNService).all()
for service in services:
addresses = session.query(IPAllocation.c.ip_address).filter(
service.router_id == Router.c.id,
Router.c.gw_port_id == Port.c.id,
Port.c.id == IPAllocation.c.port_id).all()
have_version = []
for address in addresses:
version = netaddr.IPAddress(address[0]).version
if version in have_version:
continue
have_version.append(version)
update = {'external_v%s_ip' % version: address[0]}
op.execute(VPNService.update().where(
VPNService.c.id == service.id).values(update))
session.commit()
def upgrade():
# Use the router to populate the fields
for_engine = op.get_bind()
_migrate_external_ips(for_engine)

37
neutron_vpnaas/db/migration/alembic_migrations/versions/liberty/expand/24f28869838b_add_fields_to_vpn_service_table.py Normal file
View File

@@ -0,0 +1,37 @@
# Copyright 2015 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""Add fields to VPN service table
Revision ID: 24f28869838b
Revises: 30018084ed99
Create Date: 2015-07-06 14:52:24.339246
"""
# revision identifiers, used by Alembic.
revision = '24f28869838b'
down_revision = '30018084ed99'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('vpnservices',
sa.Column('external_v4_ip', sa.String(16), nullable=True))
op.add_column('vpnservices',
sa.Column('external_v6_ip', sa.String(64), nullable=True))

13
neutron_vpnaas/db/vpn/vpn_db.py
View File

@@ -150,6 +150,8 @@ class VPNService(model_base.BASEV2, models_v2.HasId, models_v2.HasTenant):
description = sa.Column(sa.String(255))
status = sa.Column(sa.String(16), nullable=False)
admin_state_up = sa.Column(sa.Boolean(), nullable=False)
external_v4_ip = sa.Column(sa.String(16))
external_v6_ip = sa.Column(sa.String(64))
subnet_id = sa.Column(sa.String(36), sa.ForeignKey('subnets.id'),
nullable=False)
router_id = sa.Column(sa.String(36), sa.ForeignKey('routers.id'),
@@ -555,6 +557,8 @@ class VPNPluginDb(vpnaas.VPNPluginBase, base_db.CommonDbMixin):
'subnet_id': vpnservice['subnet_id'],
'router_id': vpnservice['router_id'],
'admin_state_up': vpnservice['admin_state_up'],
'external_v4_ip': vpnservice['external_v4_ip'],
'external_v6_ip': vpnservice['external_v6_ip'],
'status': vpnservice['status']}
return self._fields(res, fields)
@@ -575,6 +579,15 @@ class VPNPluginDb(vpnaas.VPNPluginBase, base_db.CommonDbMixin):
context.session.add(vpnservice_db)
return self._make_vpnservice_dict(vpnservice_db)
def set_external_tunnel_ips(self, context, vpnservice_id, v4_ip=None,
v6_ip=None):
"""Update the external tunnel IP(s) for service."""
vpns = {'external_v4_ip': v4_ip, 'external_v6_ip': v6_ip}
with context.session.begin(subtransactions=True):
vpns_db = self._get_resource(context, VPNService, vpnservice_id)
vpns_db.update(vpns)
return self._make_vpnservice_dict(vpns_db)
def update_vpnservice(self, context, vpnservice_id, vpnservice):
vpns = vpnservice['vpnservice']
with context.session.begin(subtransactions=True):

4
neutron_vpnaas/extensions/vpnaas.py
View File

@@ -142,6 +142,10 @@ RESOURCE_ATTRIBUTE_MAP = {
'default': True,
'convert_to': attr.convert_to_boolean,
'is_visible': True},
'external_v4_ip': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'external_v6_ip': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'status': {'allow_post': False, 'allow_put': False,
'is_visible': True}
},

7
neutron_vpnaas/services/vpn/plugin.py
View File

@@ -94,6 +94,13 @@ class VPNDriverPlugin(VPNPlugin, vpn_db.VPNPluginRpcDbMixin):
context, old_ipsec_site_connection, ipsec_site_connection)
return ipsec_site_connection
def create_vpnservice(self, context, vpnservice):
vpnservice = super(
VPNDriverPlugin, self).create_vpnservice(context, vpnservice)
driver = self._get_driver_for_vpnservice(vpnservice)
driver.create_vpnservice(context, vpnservice)
return vpnservice
def update_vpnservice(self, context, vpnservice_id, vpnservice):
old_vpn_service = self.get_vpnservice(context, vpnservice_id)
new_vpn_service = super(

72
neutron_vpnaas/services/vpn/service_drivers/base_ipsec.py
View File

@@ -114,8 +114,38 @@ class BaseIPsecVPNDriver(service_drivers.VpnDriver):
def update_ipsecpolicy(self, context, old_ipsec_policy, ipsecpolicy):
pass
def create_vpnservice(self, context, vpnservice):
pass
def _get_gateway_ips(self, router):
"""Obtain the IPv4 and/or IPv6 GW IP for the router.
If there are multiples, (arbitrarily) use the first one.
"""
v4_ip = v6_ip = None
for fixed_ip in router.gw_port['fixed_ips']:
addr = fixed_ip['ip_address']
vers = netaddr.IPAddress(addr).version
if vers == 4:
if v4_ip is None:
v4_ip = addr
elif v6_ip is None:
v6_ip = addr
return v4_ip, v6_ip
def create_vpnservice(self, context, vpnservice_dict):
"""Get the gateway IP(s) and save for later use.
For the reference implementation, this side's tunnel IP (external_ip)
will be the router's GW IP. IPSec connections will use a GW IP of
the same version, as is used for the peer, so we will collect the
first IP for each version (if they exist) and save them.
"""
vpnservice = self.service_plugin._get_vpnservice(context,
vpnservice_dict['id'])
v4_ip, v6_ip = self._get_gateway_ips(vpnservice.router)
vpnservice_dict['external_v4_ip'] = v4_ip
vpnservice_dict['external_v6_ip'] = v6_ip
self.service_plugin.set_external_tunnel_ips(context,
vpnservice_dict['id'],
v4_ip=v4_ip, v6_ip=v6_ip)
def update_vpnservice(self, context, old_vpnservice, vpnservice):
self.agent_rpc.vpnservice_updated(context, vpnservice['router_id'])
@@ -123,21 +153,17 @@ class BaseIPsecVPNDriver(service_drivers.VpnDriver):
def delete_vpnservice(self, context, vpnservice):
self.agent_rpc.vpnservice_updated(context, vpnservice['router_id'])
def assign_ipsec_sitecon_external_ip(self, vpnservice, ipsec_site_con):
"""Assign external_ip to ipsec siteconn.
We need to assign ip from gateway based on the
ip version of peer_address.
"""
ip_version = netaddr.IPAddress(ipsec_site_con['peer_address']).version
# *Swan use same ip version for left and right gateway ips
# If gateway has multiple GUA, we assign the first one, as
# any GUA can be used to reach the external network
for fixed_ip in vpnservice.router.gw_port['fixed_ips']:
addr = fixed_ip['ip_address']
if ip_version == netaddr.IPAddress(addr).version:
ipsec_site_con['external_ip'] = addr
break
def get_external_ip_based_on_peer(self, vpnservice, ipsec_site_con):
"""Use service's external IP, based on peer IP version."""
vers = netaddr.IPAddress(ipsec_site_con['peer_address']).version
if vers == 4:
ip_to_use = vpnservice.external_v4_ip
else:
ip_to_use = vpnservice.external_v6_ip
# TODO(pcm): Add validator to check that connection's peer address has
# a version that is available in service table, so can fail early and
# don't need a check here.
return ip_to_use
def make_vpnservice_dict(self, vpnservice):
"""Convert vpnservice information for vpn agent.
@@ -149,9 +175,10 @@ class BaseIPsecVPNDriver(service_drivers.VpnDriver):
vpnservice_dict['subnet'] = dict(
vpnservice.subnet)
# Not removing external_ip from vpnservice_dict, as some providers
# may be still using it from vpnservice_dict
vpnservice_dict['external_ip'] = vpnservice.router.gw_port[
'fixed_ips'][0]['ip_address']
# may be still using it from vpnservice_dict. Will use whichever IP
# is specified.
vpnservice_dict['external_ip'] = (
vpnservice.external_v4_ip or vpnservice.external_v6_ip)
for ipsec_site_connection in vpnservice.ipsec_site_connections:
ipsec_site_connection_dict = dict(ipsec_site_connection)
try:
@@ -169,6 +196,7 @@ class BaseIPsecVPNDriver(service_drivers.VpnDriver):
peer_cidr.cidr
for peer_cidr in ipsec_site_connection.peer_cidrs]
ipsec_site_connection_dict['peer_cidrs'] = peer_cidrs
self.assign_ipsec_sitecon_external_ip(
vpnservice, ipsec_site_connection_dict)
ipsec_site_connection_dict['external_ip'] = (
self.get_external_ip_based_on_peer(vpnservice,
ipsec_site_connection_dict))
return vpnservice_dict

155
neutron_vpnaas/tests/unit/db/vpn/test_vpn_db.py
View File

@@ -18,7 +18,9 @@ import os
import mock
from neutron.api import extensions as api_extensions
from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import constants as l3_constants
from neutron import context
from neutron.db import agentschedulers_db
from neutron.db import l3_agentschedulers_db
@@ -1608,3 +1610,156 @@ class TestVpnaas(VPNPluginDbTestCase):
mock.ANY, mock.ANY, mock.ANY, **kwargs))
vpn_plugin.check_router_in_use.assert_called_once_with(
mock.ANY, 'foo_id')
# Note: Below are new database related tests that only exercise the database
# instead of going through the client API. The intent here is to (eventually)
# convert all the database tests to this method, for faster, more granular
# tests.
# TODO(pcm): Put helpers in another module for sharing
class NeutronResourcesMixin(object):
def create_network(self, overrides=None):
"""Create datatbase entry for network."""
network_info = {'network': {'name': 'my-net',
'tenant_id': self.tenant_id,
'admin_state_up': True,
'shared': False}}
if overrides:
network_info['network'].update(overrides)
return self.core_plugin.create_network(self.context, network_info)
def create_subnet(self, overrides=None):
"""Create database entry for subnet."""
subnet_info = {'subnet': {'name': 'my-subnet',
'tenant_id': self.tenant_id,
'ip_version': 4,
'enable_dhcp': True,
'dns_nameservers': None,
'host_routes': None,
'allocation_pools': None}}
if overrides:
subnet_info['subnet'].update(overrides)
return self.core_plugin.create_subnet(self.context, subnet_info)
def create_router(self, overrides=None, gw=None):
"""Create database entry for router with optional gateway."""
router_info = {
'router': {
'name': 'my-router',
'tenant_id': self.tenant_id,
'admin_state_up': True,
}
}
if overrides:
router_info['router'].update(overrides)
if gw:
gw_info = {
'external_gateway_info': {
'network_id': gw['net_id'],
'external_fixed_ips': [{'subnet_id': gw['subnet_id'],
'ip_address': gw['ip']}],
}
}
router_info['router'].update(gw_info)
return self.l3_plugin.create_router(self.context, router_info)
def create_router_port_for_subnet(self, router, subnet):
"""Creates port on router for subnet specified."""
port = {'port': {
'tenant_id': self.tenant_id,
'network_id': subnet['network_id'],
'fixed_ips': [
{'ip_address': subnet['gateway_ip'],
'subnet_id': subnet['id']}
],
'mac_address': attributes.ATTR_NOT_SPECIFIED,
'admin_state_up': True,
'device_id': router['id'],
'device_owner': l3_constants.DEVICE_OWNER_ROUTER_INTF,
'name': ''
}}
return self.core_plugin.create_port(self.context, port)
def create_basic_topology(self):
"""Setup networks, subnets, and a router for testing VPN."""
public_net = self.create_network(overrides={'name': 'public',
'router:external': True})
private_net = self.create_network(overrides={'name': 'private'})
overrides = {'name': 'private-subnet',
'cidr': '10.2.0.0/24',
'gateway_ip': '10.2.0.1',
'network_id': private_net['id']}
private_subnet = self.create_subnet(overrides=overrides)
overrides = {'name': 'public-subnet',
'cidr': '192.168.100.0/24',
'gateway_ip': '192.168.100.1',
'allocation_pools': [{'start': '192.168.100.2',
'end': '192.168.100.254'}],
'network_id': public_net['id']}
public_subnet = self.create_subnet(overrides=overrides)
gw_info = {'net_id': public_net['id'],
'subnet_id': public_subnet['id'],
'ip': '192.168.100.5'}
router = self.create_router(gw=gw_info)
self.create_router_port_for_subnet(router, private_subnet)
return (private_subnet, router)
class TestVpnDatabase(base.NeutronDbPluginV2TestCase, NeutronResourcesMixin):
def setUp(self):
# Setup the core plugin
self.plugin_str = ('neutron_vpnaas.tests.unit.db.vpn.'
'test_vpn_db.TestVpnCorePlugin')
super(TestVpnDatabase, self).setUp(self.plugin_str)
# Get the plugins
self.core_plugin = manager.NeutronManager.get_plugin()
self.l3_plugin = manager.NeutronManager.get_service_plugins().get(
constants.L3_ROUTER_NAT)
# Create VPN database instance
self.plugin = vpn_db.VPNPluginDb()
self.tenant_id = uuidutils.generate_uuid()
self.context = context.get_admin_context()
def prepare_service_info(self, private_subnet, router):
return {'vpnservice': {'name': 'my-service',
'description': 'new service',
'subnet_id': private_subnet['id'],
'router_id': router['id'],
'admin_state_up': True}}
def test_create_vpnservice(self):
private_subnet, router = self.create_basic_topology()
info = self.prepare_service_info(private_subnet, router)
expected = {'admin_state_up': True,
'external_v4_ip': None,
'external_v6_ip': None,
'status': 'PENDING_CREATE'}
expected.update(info['vpnservice'])
new_service = self.plugin.create_vpnservice(self.context, info)
self.assertDictSupersetOf(expected, new_service)
def test_update_external_tunnel_ips(self):
"""Verify that external tunnel IPs can be set."""
private_subnet, router = self.create_basic_topology()
info = self.prepare_service_info(private_subnet, router)
expected = {'admin_state_up': True,
'external_v4_ip': None,
'external_v6_ip': None,
'status': 'PENDING_CREATE'}
expected.update(info['vpnservice'])
new_service = self.plugin.create_vpnservice(self.context, info)
self.assertDictSupersetOf(expected, new_service)
external_v4_ip = '192.168.100.5'
external_v6_ip = 'fd00:1000::4'
expected.update({'external_v4_ip': external_v4_ip,
'external_v6_ip': external_v6_ip})
mod_service = self.plugin.set_external_tunnel_ips(self.context,
new_service['id'],
v4_ip=external_v4_ip,
v6_ip=external_v6_ip)
self.assertDictSupersetOf(expected, mod_service)

80
neutron_vpnaas/tests/unit/services/vpn/service_drivers/test_ipsec.py
View File

@@ -44,6 +44,7 @@ FAKE_ROUTER = {l3_db.EXTERNAL_GW_INFO: FAKE_ROUTER_ID}
FAKE_SUBNET_ID = _uuid()
IPV4 = 4
IPV6 = 6
FAKE_CONN_ID = _uuid()
IPSEC_SERVICE_DRIVER = ('neutron_vpnaas.services.vpn.service_drivers.'
'ipsec.IPsecVPNDriver')
@@ -308,13 +309,13 @@ class TestIPsecDriver(base.BaseTestCase):
'neutron.manager.NeutronManager.get_service_plugins')
get_service_plugin = service_plugin_p.start()
get_service_plugin.return_value = {constants.L3_ROUTER_NAT: plugin}
service_plugin = mock.Mock()
service_plugin.get_l3_agents_hosting_routers.return_value = [l3_agent]
self.svc_plugin = mock.Mock()
self.svc_plugin.get_l3_agents_hosting_routers.return_value = [l3_agent]
self._fake_vpn_router_id = _uuid()
service_plugin._get_vpnservice.return_value = {
self.svc_plugin._get_vpnservice.return_value = {
'router_id': self._fake_vpn_router_id
}
self.driver = ipsec_driver.IPsecVPNDriver(service_plugin)
self.driver = ipsec_driver.IPsecVPNDriver(self.svc_plugin)
def _test_update(self, func, args, additional_info=None):
ctxt = n_ctx.Context('', 'somebody')
@@ -377,6 +378,8 @@ class TestIPsecDriver(base.BaseTestCase):
description='foo-vpn-service',
admin_state_up=True,
status='active',
external_v4_ip=fake_external_ip,
external_v6_ip=None,
subnet_id='foo-subnet-id',
router_id='foo-router-id')
fake_vpnservice.subnet = fake_subnet
@@ -388,6 +391,8 @@ class TestIPsecDriver(base.BaseTestCase):
'description': 'foo-vpn-service',
'admin_state_up': True,
'status': 'active',
'external_v4_ip': fake_external_ip,
'external_v6_ip': None,
'subnet_id': 'foo-subnet-id',
'router_id': 'foo-router-id',
'subnet': {'id': 'foo-subnet-id',
@@ -419,18 +424,65 @@ class TestIPsecDriver(base.BaseTestCase):
def test_make_vpnservice_dict_peer_id_is_string(self):
self._test_make_vpnservice_dict_helper('foo.peer.id', '@foo.peer.id')
def test_assign_ipsec_sitecon_external_ip_with_ipv4(self):
def test_get_external_ip_based_on_ipv4_peer(self):
vpnservice = mock.Mock()
vpnservice.router.gw_port = {'fixed_ips': [{'ip_address': '10.0.0.99'},
{'ip_address': '2001::1'}]}
ipsec_sitecon = {'peer_address': '10.0.0.9'}
self.driver.assign_ipsec_sitecon_external_ip(vpnservice, ipsec_sitecon)
self.assertEqual('10.0.0.99', ipsec_sitecon['external_ip'])
vpnservice.external_v4_ip = '10.0.0.99'
vpnservice.external_v6_ip = '2001::1'
ipsec_sitecon = {'id': FAKE_CONN_ID, 'peer_address': '10.0.0.9'}
ip_to_use = self.driver.get_external_ip_based_on_peer(vpnservice,
ipsec_sitecon)
self.assertEqual('10.0.0.99', ip_to_use)
def test_assign_ipsec_sitecon_external_ip_with_ipv6(self):
def test_get_external_ip_based_on_ipv6_peer(self):
vpnservice = mock.Mock()
vpnservice.external_v4_ip = '10.0.0.99'
vpnservice.external_v6_ip = '2001::1'
ipsec_sitecon = {'id': FAKE_CONN_ID, 'peer_address': '2001::5'}
ip_to_use = self.driver.get_external_ip_based_on_peer(vpnservice,
ipsec_sitecon)
self.assertEqual('2001::1', ip_to_use)
def test_get_ipv4_gw_ip(self):
vpnservice = mock.Mock()
vpnservice.router.gw_port = {'fixed_ips':
[{'ip_address': '10.0.0.99'}]}
v4_ip, v6_ip = self.driver._get_gateway_ips(vpnservice.router)
self.assertEqual('10.0.0.99', v4_ip)
self.assertIsNone(v6_ip)
def test_get_ipv6_gw_ip(self):
vpnservice = mock.Mock()
vpnservice.router.gw_port = {'fixed_ips': [{'ip_address': '2001::1'}]}
v4_ip, v6_ip = self.driver._get_gateway_ips(vpnservice.router)
self.assertIsNone(v4_ip)
self.assertEqual('2001::1', v6_ip)
def test_get_both_gw_ips(self):
vpnservice = mock.Mock()
vpnservice.router.gw_port = {'fixed_ips': [{'ip_address': '10.0.0.99'},
{'ip_address': '2001::1'}]}
ipsec_sitecon = {'peer_address': '2001::5'}
self.driver.assign_ipsec_sitecon_external_ip(vpnservice, ipsec_sitecon)
self.assertEqual('2001::1', ipsec_sitecon['external_ip'])
v4_ip, v6_ip = self.driver._get_gateway_ips(vpnservice.router)
self.assertEqual('10.0.0.99', v4_ip)
self.assertEqual('2001::1', v6_ip)
def test_use_first_gw_ips_when_multiples(self):
vpnservice = mock.Mock()
vpnservice.router.gw_port = {'fixed_ips': [{'ip_address': '10.0.0.99'},
{'ip_address': '20.0.0.99'},
{'ip_address': '2001::1'},
{'ip_address': 'fd00::4'}]}
v4_ip, v6_ip = self.driver._get_gateway_ips(vpnservice.router)
self.assertEqual('10.0.0.99', v4_ip)
self.assertEqual('2001::1', v6_ip)
def test_store_gw_ips_on_service_create(self):
vpnservice = mock.Mock()
self.svc_plugin._get_vpnservice.return_value = vpnservice
vpnservice.router.gw_port = {'fixed_ips': [{'ip_address': '10.0.0.99'},
{'ip_address': '2001::1'}]}
ctxt = n_ctx.Context('', 'somebody')
vpnservice_dict = {'id': FAKE_SERVICE_ID,
'router_id': FAKE_ROUTER_ID}
self.driver.create_vpnservice(ctxt, vpnservice_dict)
self.svc_plugin.set_external_tunnel_ips.assert_called_once_with(
ctxt, FAKE_SERVICE_ID, v4_ip='10.0.0.99', v6_ip='2001::1')

5
neutron_vpnaas/tests/unit/services/vpn/test_plugin.py
View File

@@ -57,6 +57,11 @@ class TestVPNDriverPlugin(test_db_vpnaas.TestVpnaas,
self.driver.delete_ipsec_site_connection.assert_called_once_with(
mock.ANY, mock.ANY)
def test_create_vpnservice(self):
super(TestVPNDriverPlugin, self).test_create_vpnservice()
self.driver.create_vpnservice.assert_called_once_with(
mock.ANY, mock.ANY)
def test_delete_vpnservice(self, **extras):
super(TestVPNDriverPlugin, self).test_delete_vpnservice()
self.driver.delete_vpnservice.assert_called_once_with(