neutron/tests/tempest/api/test_routers.py

# Copyright 2013 OpenStack Foundation
# 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 netaddr
import six
from tempest.lib.common.utils import data_utils
from tempest import test

from neutron.tests.tempest.api import base_routers as base
from neutron.tests.tempest import config

CONF = config.CONF


class RoutersTest(base.BaseRouterTest):

    @classmethod
    @test.requires_ext(extension="router", service="network")
    def skip_checks(cls):
        super(RoutersTest, cls).skip_checks()

    @classmethod
    def resource_setup(cls):
        super(RoutersTest, cls).resource_setup()
        cls.tenant_cidr = (
            config.safe_get_config_value('network', 'project_network_cidr')
            if cls._ip_version == 4 else
            config.safe_get_config_value('network', 'project_network_v6_cidr'))

    @test.idempotent_id('c72c1c0c-2193-4aca-eeee-b1442640eeee')
    @test.requires_ext(extension="standard-attr-description",
                       service="network")
    def test_create_update_router_description(self):
        body = self.create_router(description='d1', router_name='test')
        self.assertEqual('d1', body['description'])
        body = self.client.show_router(body['id'])['router']
        self.assertEqual('d1', body['description'])
        body = self.client.update_router(body['id'], description='d2')
        self.assertEqual('d2', body['router']['description'])
        body = self.client.show_router(body['router']['id'])['router']
        self.assertEqual('d2', body['description'])

    @test.idempotent_id('847257cc-6afd-4154-b8fb-af49f5670ce8')
    @test.requires_ext(extension='ext-gw-mode', service='network')
    def test_create_router_with_default_snat_value(self):
        # Create a router with default snat rule
        name = data_utils.rand_name('router')
        router = self._create_router(
            name, external_network_id=CONF.network.public_network_id)
        self._verify_router_gateway(
            router['id'], {'network_id': CONF.network.public_network_id,
                           'enable_snat': True})

    @test.idempotent_id('ea74068d-09e9-4fd7-8995-9b6a1ace920f')
    @test.requires_ext(extension='ext-gw-mode', service='network')
    def test_create_router_with_snat_explicit(self):
        name = data_utils.rand_name('snat-router')
        # Create a router enabling snat attributes
        enable_snat_states = [False, True]
        for enable_snat in enable_snat_states:
            external_gateway_info = {
                'network_id': CONF.network.public_network_id,
                'enable_snat': enable_snat}
            create_body = self.admin_client.create_router(
                name, external_gateway_info=external_gateway_info)
            self.addCleanup(self.admin_client.delete_router,
                            create_body['router']['id'])
            # Verify snat attributes after router creation
            self._verify_router_gateway(create_body['router']['id'],
                                        exp_ext_gw_info=external_gateway_info)

    def _verify_router_gateway(self, router_id, exp_ext_gw_info=None):
        show_body = self.admin_client.show_router(router_id)
        actual_ext_gw_info = show_body['router']['external_gateway_info']
        if exp_ext_gw_info is None:
            self.assertIsNone(actual_ext_gw_info)
            return
        # Verify only keys passed in exp_ext_gw_info
        for k, v in six.iteritems(exp_ext_gw_info):
            self.assertEqual(v, actual_ext_gw_info[k])

    def _verify_gateway_port(self, router_id):
        list_body = self.admin_client.list_ports(
            network_id=CONF.network.public_network_id,
            device_id=router_id)
        self.assertEqual(len(list_body['ports']), 1)
        gw_port = list_body['ports'][0]
        fixed_ips = gw_port['fixed_ips']
        self.assertGreaterEqual(len(fixed_ips), 1)
        public_net_body = self.admin_client.show_network(
            CONF.network.public_network_id)
        public_subnet_id = public_net_body['network']['subnets'][0]
        self.assertIn(public_subnet_id,
                      [x['subnet_id'] for x in fixed_ips])

    @test.idempotent_id('b386c111-3b21-466d-880c-5e72b01e1a33')
    @test.requires_ext(extension='ext-gw-mode', service='network')
    def test_update_router_set_gateway_with_snat_explicit(self):
        router = self._create_router(data_utils.rand_name('router-'))
        self.admin_client.update_router_with_snat_gw_info(
            router['id'],
            external_gateway_info={
                'network_id': CONF.network.public_network_id,
                'enable_snat': True})
        self._verify_router_gateway(
            router['id'],
            {'network_id': CONF.network.public_network_id,
             'enable_snat': True})
        self._verify_gateway_port(router['id'])

    @test.idempotent_id('96536bc7-8262-4fb2-9967-5c46940fa279')
    @test.requires_ext(extension='ext-gw-mode', service='network')
    def test_update_router_set_gateway_without_snat(self):
        router = self._create_router(data_utils.rand_name('router-'))
        self.admin_client.update_router_with_snat_gw_info(
            router['id'],
            external_gateway_info={
                'network_id': CONF.network.public_network_id,
                'enable_snat': False})
        self._verify_router_gateway(
            router['id'],
            {'network_id': CONF.network.public_network_id,
             'enable_snat': False})
        self._verify_gateway_port(router['id'])

    @test.idempotent_id('f2faf994-97f4-410b-a831-9bc977b64374')
    @test.requires_ext(extension='ext-gw-mode', service='network')
    def test_update_router_reset_gateway_without_snat(self):
        router = self._create_router(
            data_utils.rand_name('router-'),
            external_network_id=CONF.network.public_network_id)
        self.admin_client.update_router_with_snat_gw_info(
            router['id'],
            external_gateway_info={
                'network_id': CONF.network.public_network_id,
                'enable_snat': False})
        self._verify_router_gateway(
            router['id'],
            {'network_id': CONF.network.public_network_id,
             'enable_snat': False})
        self._verify_gateway_port(router['id'])

    @test.idempotent_id('c86ac3a8-50bd-4b00-a6b8-62af84a0765c')
    @test.requires_ext(extension='extraroute', service='network')
    def test_update_extra_route(self):
        self.network = self.create_network()
        self.name = self.network['name']
        self.subnet = self.create_subnet(self.network)
        # Add router interface with subnet id
        self.router = self._create_router(
            data_utils.rand_name('router-'), True)
        self.create_router_interface(self.router['id'], self.subnet['id'])
        self.addCleanup(
            self._delete_extra_routes,
            self.router['id'])
        # Update router extra route, second ip of the range is
        # used as next hop
        cidr = netaddr.IPNetwork(self.subnet['cidr'])
        next_hop = str(cidr[2])
        destination = str(self.subnet['cidr'])
        extra_route = self.client.update_extra_routes(self.router['id'],
                                                      next_hop, destination)
        self.assertEqual(1, len(extra_route['router']['routes']))
        self.assertEqual(destination,
                         extra_route['router']['routes'][0]['destination'])
        self.assertEqual(next_hop,
                         extra_route['router']['routes'][0]['nexthop'])
        show_body = self.client.show_router(self.router['id'])
        self.assertEqual(destination,
                         show_body['router']['routes'][0]['destination'])
        self.assertEqual(next_hop,
                         show_body['router']['routes'][0]['nexthop'])

    def _delete_extra_routes(self, router_id):
        self.client.delete_extra_routes(router_id)

    @test.idempotent_id('01f185d1-d1a6-4cf9-abf7-e0e1384c169c')
    def test_network_attached_with_two_routers(self):
        network = self.create_network(data_utils.rand_name('network1'))
        self.create_subnet(network)
        port1 = self.create_port(network)
        port2 = self.create_port(network)
        router1 = self._create_router(data_utils.rand_name('router1'))
        router2 = self._create_router(data_utils.rand_name('router2'))
        self.client.add_router_interface_with_port_id(
            router1['id'], port1['id'])
        self.client.add_router_interface_with_port_id(
            router2['id'], port2['id'])
        self.addCleanup(self.client.remove_router_interface_with_port_id,
                        router1['id'], port1['id'])
        self.addCleanup(self.client.remove_router_interface_with_port_id,
                        router2['id'], port2['id'])
        body = self.client.show_port(port1['id'])
        port_show1 = body['port']
        body = self.client.show_port(port2['id'])
        port_show2 = body['port']
        self.assertEqual(port_show1['network_id'], network['id'])
        self.assertEqual(port_show2['network_id'], network['id'])
        self.assertEqual(port_show1['device_id'], router1['id'])
        self.assertEqual(port_show2['device_id'], router2['id'])


class RoutersIpV6Test(RoutersTest):
    _ip_version = 6


class DvrRoutersTest(base.BaseRouterTest):

    @classmethod
    @test.requires_ext(extension="dvr", service="network")
    def skip_checks(cls):
        super(DvrRoutersTest, cls).skip_checks()

    @test.idempotent_id('141297aa-3424-455d-aa8d-f2d95731e00a')
    def test_create_distributed_router(self):
        name = data_utils.rand_name('router')
        create_body = self.admin_client.create_router(
            name, distributed=True)
        self.addCleanup(self._delete_router,
                        create_body['router']['id'],
                        self.admin_client)
        self.assertTrue(create_body['router']['distributed'])

    @test.idempotent_id('644d7a4a-01a1-4b68-bb8d-0c0042cb1729')
    def test_convert_centralized_router(self):
        router = self._create_router(data_utils.rand_name('router'))
        self.assertNotIn('distributed', router)
        update_body = self.admin_client.update_router(router['id'],
                                                      distributed=True)
        self.assertTrue(update_body['router']['distributed'])
        show_body = self.admin_client.show_router(router['id'])
        self.assertTrue(show_body['router']['distributed'])
        show_body = self.client.show_router(router['id'])
        self.assertNotIn('distributed', show_body['router'])