# Copyright 2016 VMware 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. import time from tempest.api.network import base from tempest import config from tempest.lib.common.utils import data_utils from tempest.lib import decorators from tempest import test from vmware_nsx_tempest_plugin.common import constants from vmware_nsx_tempest_plugin.services import nsxp_client from vmware_nsx_tempest_plugin.services import nsxv3_client CONF = config.CONF class NSXv3RoutersTest(base.BaseAdminNetworkTest): """Test L3 Router and realization on NSX backend When test L3 Router feature, we need to test both REST API call from neutron and realization state on backend. Two tests have been added in this class: - Test create and update router - Test delete router """ @classmethod def skip_checks(cls): super(NSXv3RoutersTest, cls).skip_checks() if not test.is_extension_enabled('router', 'network'): msg = "router extension not enabled." raise cls.skipException(msg) @classmethod def resource_setup(cls): super(NSXv3RoutersTest, cls).resource_setup() cls.nsx = nsxv3_client.NSXV3Client(CONF.nsxv3.nsx_manager, CONF.nsxv3.nsx_user, CONF.nsxv3.nsx_password) cls.nsxp = nsxp_client.NSXPClient(CONF.nsxv3.nsx_manager, CONF.nsxv3.nsx_user, CONF.nsxv3.nsx_password) @decorators.attr(type='nsxv3') @decorators.idempotent_id('0e9938bc-d2a3-4a9a-a4f9-7a93ee8bb344') def test_create_update_nsx_router(self): # Create a router router_name = data_utils.rand_name('router-') router = self.create_router(router_name, admin_state_up=True) self.addCleanup(self._delete_router, router['id']) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(router['name'], router['id']) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) nsx_router = self.nsx.get_logical_router(router['name'], router['id']) self.assertEqual(router['name'], router_name) self.assertEqual(router['admin_state_up'], True) self.assertIsNotNone(nsx_router) # Update the name of router and verify if it is updated on both # neutron and nsx backend updated_name = 'updated ' + router_name update_body = self.routers_client.update_router(router['id'], name=updated_name) updated_router = update_body['router'] if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(updated_router['name'], updated_router['id']) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) nsx_router = self.nsx.get_logical_router(updated_router['name'], updated_router['id']) self.assertEqual(updated_router['name'], updated_name) self.assertIsNotNone(nsx_router) @decorators.attr(type='nsxv3') @decorators.idempotent_id('6f49b69c-0800-4c83-b1f8-595ae5bfeea7') def test_delete_nsx_router(self): # Create a router router_name = data_utils.rand_name('router-') router = self.create_router(router_name, admin_state_up=True) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(router['name'], router['id']) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) nsx_router = self.nsx.get_logical_router(router['name'], router['id']) self.assertEqual(router['name'], router_name) self.assertIsNotNone(nsx_router) # Delete the router and verify it is deleted on nsx backend self.routers_client.delete_router(router['id']) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(router['name'], router['id']) self.assertIsNone(nsx_router) nsx_router = self.nsx.get_logical_router(router['name'], router['id']) self.assertIsNone(nsx_router) def _delete_router(self, router_id): # Delete the router in case the test exits with any exception list_body = self.routers_client.list_routers() for router in list_body.get('router', []): if router['id'] == router_id: self.routers_client.delete_router(router_id) @decorators.attr(type='nsxv3') @decorators.idempotent_id('6f49b69c-0830-4c83-b1f8-5953e5bfeea5') def test_deploy_router_ha_with_relocation_enabled(self): # Check Standby relocation is enabled on backend after T1 Router got # deployed. router_name = data_utils.rand_name('router-') body = self.create_router( router_name=router_name, admin_state_up=True, external_network_id=CONF.network.public_network_id) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) self.assertEqual(nsx_router.get('enable_standby_relocation'), True) nsx_router = self.nsx.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) self.assertEqual(nsx_router.get('allocation_profile').get( 'enable_standby_relocation'), True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('6f49b69c-0830-4c83-b1f8-5953e6bfeea5') def test_deploy_router_ha_with_relocation_enable_disable(self): # Check Standby relocation is enabled on backend after T1 Router got # deployed. router_name = data_utils.rand_name('router-') body = self.create_router( router_name=router_name, admin_state_up=True) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) self.assertEqual(nsx_router.get( 'enable_standby_relocation'), False) nsx_router = self.nsx.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) product_version = self.nsx.get_product_version('/node/version') if product_version >= 3.2: self.assertEqual('enable_standby_relocation' in nsx_router.get( 'allocation_profile').keys(), False) else: self.assertEqual(nsx_router.get('allocation_profile').get( 'enable_standby_relocation'), False) public_network_info = {"external_gateway_info": dict(network_id=CONF.network.public_network_id)} self.routers_client.update_router(body['id'], **public_network_info) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) self.assertEqual(nsx_router.get('enable_standby_relocation'), True) nsx_router = self.nsx.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) self.assertEqual(nsx_router.get('allocation_profile').get( 'enable_standby_relocation'), True) public_network_info = {"external_gateway_info": {}} self.routers_client.update_router(body['id'], **public_network_info) if CONF.network.backend == 'nsxp': time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL) nsx_router = self.nsxp.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) if 'realization_id' in nsx_router.keys(): self.assertIsNotNone(nsx_router['realization_id']) r_state = self.nsxp.verify_realized_state(nsx_router) self.assertTrue(r_state) self.assertEqual(nsx_router.get( 'enable_standby_relocation'), False) nsx_router = self.nsx.get_logical_router(body['name'], body['id']) self.assertEqual(body['name'], router_name) self.assertIsNotNone(nsx_router) self.assertEqual(nsx_router.get('allocation_profile').get( 'enable_standby_relocation'), False)