Checking functionality of DVR
1) Check traffic pass when the SNAT port is enabled
Disable the port and check the traffic is still passing
2) Check centralized routers on DVR setup works through network
node
Update router to be distributed
Check traffic works when the SNAT port is down
3) Fix test_trunk as the function for creating router changed
Change-Id: I5022f42392c0a076aa4980082640392cad1688a0
diff --git a/neutron/tests/tempest/api/base.py b/neutron/tests/tempest/api/base.py
index 0d77064..49c48d6 100644
--- a/neutron/tests/tempest/api/base.py
+++ b/neutron/tests/tempest/api/base.py
@@ -317,15 +317,16 @@
return body['port']
@classmethod
- def create_router(cls, router_name=None, admin_state_up=False,
- external_network_id=None, enable_snat=None,
- **kwargs):
+ def _create_router_with_client(
+ cls, client, router_name=None, admin_state_up=False,
+ external_network_id=None, enable_snat=None, **kwargs
+ ):
ext_gw_info = {}
if external_network_id:
ext_gw_info['network_id'] = external_network_id
if enable_snat:
ext_gw_info['enable_snat'] = enable_snat
- body = cls.client.create_router(
+ body = client.create_router(
router_name, external_gateway_info=ext_gw_info,
admin_state_up=admin_state_up, **kwargs)
router = body['router']
@@ -333,6 +334,15 @@
return router
@classmethod
+ def create_router(cls, *args, **kwargs):
+ return cls._create_router_with_client(cls.client, *args, **kwargs)
+
+ @classmethod
+ def create_admin_router(cls, *args, **kwargs):
+ return cls._create_router_with_client(cls.admin_manager.network_client,
+ *args, **kwargs)
+
+ @classmethod
def create_floatingip(cls, external_network_id):
"""Wrapper utility that returns a test floating IP."""
body = cls.client.create_floatingip(
diff --git a/neutron/tests/tempest/scenario/base.py b/neutron/tests/tempest/scenario/base.py
index 56221e5..351e587 100644
--- a/neutron/tests/tempest/scenario/base.py
+++ b/neutron/tests/tempest/scenario/base.py
@@ -128,12 +128,15 @@
cls.create_secgroup_rules(rule_list, secgroup_id=secgroup_id)
@classmethod
- def create_router_and_interface(cls, subnet_id):
- router = cls.create_router(
- data_utils.rand_name('router'), admin_state_up=True,
- external_network_id=CONF.network.public_network_id)
+ def create_router_by_client(cls, is_admin=False, **kwargs):
+ kwargs.update({'router_name': data_utils.rand_name('router'),
+ 'admin_state_up': True,
+ 'external_network_id': CONF.network.public_network_id})
+ if not is_admin:
+ router = cls.create_router(**kwargs)
+ else:
+ router = cls.create_admin_router(**kwargs)
LOG.debug("Created router %s", router['name'])
- cls.create_router_interface(router['id'], subnet_id)
cls.routers.append(router)
return router
@@ -151,8 +154,8 @@
ssh_client.test_connection_auth()
@classmethod
- def setup_network_and_server(cls):
- """Creating network resources and a server.
+ def setup_network_and_server(cls, router=None, **kwargs):
+ """Create network resources and a server.
Creating a network, subnet, router, keypair, security group
and a server.
@@ -167,8 +170,9 @@
LOG.debug("Created security group %s",
secgroup['security_group']['name'])
cls.security_groups.append(secgroup['security_group'])
-
- cls.create_router_and_interface(cls.subnet['id'])
+ if not router:
+ router = cls.create_router_by_client(**kwargs)
+ cls.create_router_interface(router['id'], cls.subnet['id'])
cls.keypair = cls.create_keypair()
cls.create_loginable_secgroup_rule(
secgroup_id=secgroup['security_group']['id'])
diff --git a/neutron/tests/tempest/scenario/test_dvr.py b/neutron/tests/tempest/scenario/test_dvr.py
new file mode 100644
index 0000000..9150c01
--- /dev/null
+++ b/neutron/tests/tempest/scenario/test_dvr.py
@@ -0,0 +1,81 @@
+# Copyright 2016 Red Hat, Inc.
+# All Rights Reserved.
+#
+# 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.
+from tempest import test
+
+from neutron.tests.tempest import config
+from neutron.tests.tempest.scenario import base
+from neutron_lib import constants
+
+CONF = config.CONF
+
+
+class NetworkDvrTest(base.BaseTempestTestCase):
+ credentials = ['primary', 'admin']
+ force_tenant_isolation = False
+
+ @classmethod
+ @test.requires_ext(extension="dvr", service="network")
+ def skip_checks(cls):
+ super(NetworkDvrTest, cls).skip_checks()
+
+ def _check_connectivity(self):
+ self.check_connectivity(self.fip['floating_ip_address'],
+ CONF.validation.image_ssh_user,
+ self.keypair['private_key'])
+
+ def _check_snat_port_connectivity(self):
+ self._check_connectivity()
+
+ # Put the Router_SNAT port down to make sure the traffic flows through
+ # Compute node.
+ self._put_snat_port_down(self.network['id'])
+ self._check_connectivity()
+
+ def _put_snat_port_down(self, network_id):
+ port_id = self.client.list_ports(
+ network_id=network_id,
+ device_owner=constants.DEVICE_OWNER_ROUTER_SNAT)['ports'][0]['id']
+ self.admin_manager.network_client.update_port(
+ port_id, admin_state_up=False)
+
+ @test.idempotent_id('3d73ec1a-2ec6-45a9-b0f8-04a283d9d344')
+ def test_vm_reachable_through_compute(self):
+ """Check that the VM is reachable through compute node.
+
+ The test is done by putting the SNAT port down on controller node.
+ """
+ router = self.create_router_by_client(
+ distributed=True, tenant_id=self.client.tenant_id, is_admin=True)
+ self.setup_network_and_server(router=router)
+ self._check_snat_port_connectivity()
+
+ @test.idempotent_id('23724222-483a-4129-bc15-7a9278f3828b')
+ def test_update_centralized_router_to_dvr(self):
+ """Check that updating centralized router to be distributed works.
+ """
+ # Created a centralized router on a DVR setup
+ router = self.create_router_by_client(
+ distributed=False, tenant_id=self.client.tenant_id, is_admin=True)
+ self.setup_network_and_server(router=router)
+ self._check_connectivity()
+
+ # Update router to be distributed
+ self.admin_manager.network_client.update_router(
+ router_id=router['id'], admin_state_up=False)
+ self.admin_manager.network_client.update_router(
+ router_id=router['id'], distributed=True)
+ self.admin_manager.network_client.update_router(
+ router_id=router['id'], admin_state_up=True)
+ self._check_snat_port_connectivity()
diff --git a/neutron/tests/tempest/scenario/test_trunk.py b/neutron/tests/tempest/scenario/test_trunk.py
index b350392..408ef79 100644
--- a/neutron/tests/tempest/scenario/test_trunk.py
+++ b/neutron/tests/tempest/scenario/test_trunk.py
@@ -37,7 +37,8 @@
# setup basic topology for servers we can log into
cls.network = cls.create_network()
cls.subnet = cls.create_subnet(cls.network)
- cls.create_router_and_interface(cls.subnet['id'])
+ router = cls.create_router_by_client()
+ cls.create_router_interface(router['id'], cls.subnet['id'])
cls.keypair = cls.create_keypair()
cls.secgroup = cls.manager.network_client.create_security_group(
name=data_utils.rand_name('secgroup-'))