openstacksdk/shade/tests/unit/test_router.py

# Copyright 2017 OVH SAS
# 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 copy

from shade import exc
from shade.tests.unit import base


class TestRouter(base.RequestsMockTestCase):

    router_name = 'goofy'
    router_id = '57076620-dcfb-42ed-8ad6-79ccb4a79ed2'
    subnet_id = '1f1696eb-7f47-47f6-835c-4889bff88604'

    mock_router_rep = {
        'admin_state_up': True,
        'availability_zone_hints': [],
        'availability_zones': [],
        'description': u'',
        'distributed': False,
        'external_gateway_info': None,
        'flavor_id': None,
        'ha': False,
        'id': router_id,
        'name': router_name,
        'project_id': u'861808a93da0484ea1767967c4df8a23',
        'routes': [],
        'status': u'ACTIVE',
        'tenant_id': u'861808a93da0484ea1767967c4df8a23'
    }

    mock_router_interface_rep = {
        'network_id': '53aee281-b06d-47fc-9e1a-37f045182b8e',
        'subnet_id': '1f1696eb-7f47-47f6-835c-4889bff88604',
        'tenant_id': '861808a93da0484ea1767967c4df8a23',
        'subnet_ids': [subnet_id],
        'port_id': '23999891-78b3-4a6b-818d-d1b713f67848',
        'id': '57076620-dcfb-42ed-8ad6-79ccb4a79ed2',
        'request_ids': ['req-f1b0b1b4-ae51-4ef9-b371-0cc3c3402cf7']
    }

    def test_get_router(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': [self.mock_router_rep]})
        ])
        r = self.cloud.get_router(self.router_name)
        self.assertIsNotNone(r)
        self.assertDictEqual(self.mock_router_rep, r)
        self.assert_calls()

    def test_get_router_not_found(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': []})
        ])
        r = self.cloud.get_router('mickey')
        self.assertIsNone(r)
        self.assert_calls()

    def test_create_router(self):
        self.register_uris([
            dict(method='POST',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'router': self.mock_router_rep},
                 validate=dict(
                     json={'router': {
                         'name': self.router_name,
                         'admin_state_up': True}}))
        ])
        new_router = self.cloud.create_router(name=self.router_name,
                                              admin_state_up=True)
        self.assertDictEqual(self.mock_router_rep, new_router)
        self.assert_calls()

    def test_create_router_specific_tenant(self):
        new_router_tenant_id = "project_id_value"
        mock_router_rep = copy.copy(self.mock_router_rep)
        mock_router_rep['tenant_id'] = new_router_tenant_id
        mock_router_rep['project_id'] = new_router_tenant_id
        self.register_uris([
            dict(method='POST',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'router': mock_router_rep},
                 validate=dict(
                     json={'router': {
                         'name': self.router_name,
                         'admin_state_up': True,
                         'tenant_id': new_router_tenant_id}}))
        ])

        self.cloud.create_router(self.router_name,
                                 project_id=new_router_tenant_id)
        self.assert_calls()

    def test_create_router_with_enable_snat_True(self):
        """Do not send enable_snat when same as neutron default."""
        self.register_uris([
            dict(method='POST',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'router': self.mock_router_rep},
                 validate=dict(
                     json={'router': {
                         'name': self.router_name,
                         'admin_state_up': True}}))
        ])
        self.cloud.create_router(
            name=self.router_name, admin_state_up=True, enable_snat=True)
        self.assert_calls()

    def test_create_router_with_enable_snat_False(self):
        """Send enable_snat when it is False."""
        self.register_uris([
            dict(method='POST',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'router': self.mock_router_rep},
                 validate=dict(
                     json={'router': {
                         'name': self.router_name,
                         'external_gateway_info': {'enable_snat': False},
                         'admin_state_up': True}}))
        ])
        self.cloud.create_router(
            name=self.router_name, admin_state_up=True, enable_snat=False)
        self.assert_calls()

    def test_add_router_interface(self):
        self.register_uris([
            dict(method='PUT',
                 uri=self.get_mock_url(
                     'network', 'public',
                     append=['v2.0', 'routers', self.router_id,
                             'add_router_interface.json']),
                 json={'port': self.mock_router_interface_rep},
                 validate=dict(
                     json={'subnet_id': self.subnet_id}))
        ])
        self.cloud.add_router_interface(
            {'id': self.router_id}, subnet_id=self.subnet_id)
        self.assert_calls()

    def test_remove_router_interface(self):
        self.register_uris([
            dict(method='PUT',
                 uri=self.get_mock_url(
                     'network', 'public',
                     append=['v2.0', 'routers', self.router_id,
                             'remove_router_interface.json']),
                 json={'port': self.mock_router_interface_rep},
                 validate=dict(
                     json={'subnet_id': self.subnet_id}))
        ])
        self.cloud.remove_router_interface(
            {'id': self.router_id}, subnet_id=self.subnet_id)
        self.assert_calls()

    def test_remove_router_interface_missing_argument(self):
        self.assertRaises(ValueError, self.cloud.remove_router_interface,
                          {'id': '123'})

    def test_update_router(self):
        new_router_name = "mickey"
        expected_router_rep = copy.copy(self.mock_router_rep)
        expected_router_rep['name'] = new_router_name
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': [self.mock_router_rep]}),
            dict(method='PUT',
                 uri=self.get_mock_url(
                     'network', 'public',
                     append=['v2.0', 'routers', '%s.json' % self.router_id]),
                 json={'router': expected_router_rep},
                 validate=dict(
                     json={'router': {
                         'name': new_router_name}}))
        ])
        new_router = self.cloud.update_router(
            self.router_id, name=new_router_name)
        self.assertDictEqual(expected_router_rep, new_router)
        self.assert_calls()

    def test_delete_router(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': [self.mock_router_rep]}),
            dict(method='DELETE',
                 uri=self.get_mock_url(
                     'network', 'public',
                     append=['v2.0', 'routers', '%s.json' % self.router_id]),
                 json={})
        ])
        self.assertTrue(self.cloud.delete_router(self.router_name))
        self.assert_calls()

    def test_delete_router_not_found(self):
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': []}),
        ])
        self.assertFalse(self.cloud.delete_router(self.router_name))
        self.assert_calls()

    def test_delete_router_multiple_found(self):
        router1 = dict(id='123', name='mickey')
        router2 = dict(id='456', name='mickey')
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': [router1, router2]}),
        ])
        self.assertRaises(exc.OpenStackCloudException,
                          self.cloud.delete_router,
                          'mickey')
        self.assert_calls()

    def test_delete_router_multiple_using_id(self):
        router1 = dict(id='123', name='mickey')
        router2 = dict(id='456', name='mickey')
        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'routers.json']),
                 json={'routers': [router1, router2]}),
            dict(method='DELETE',
                 uri=self.get_mock_url(
                     'network', 'public',
                     append=['v2.0', 'routers', '123.json']),
                 json={})
        ])
        self.assertTrue(self.cloud.delete_router("123"))
        self.assert_calls()

    def _test_list_router_interfaces(self, router, interface_type,
                                     expected_result=None):
        internal_port = {
            'id': 'internal_port_id',
            'fixed_ips': [{
                'subnet_id': 'internal_subnet_id',
                'ip_address': "10.0.0.1"
            }],
            'device_id': self.router_id
        }
        external_port = {
            'id': 'external_port_id',
            'fixed_ips': [{
                'subnet_id': 'external_subnet_id',
                'ip_address': "1.2.3.4"
            }],
            'device_id': self.router_id
        }
        if expected_result is None:
            if interface_type == "internal":
                expected_result = [internal_port]
            elif interface_type == "external":
                expected_result = [external_port]
            else:
                expected_result = [internal_port, external_port]

        self.register_uris([
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'ports.json'],
                     qs_elements=["device_id=%s" % self.router_id]),
                 json={'ports': [internal_port, external_port]})
        ])
        ret = self.cloud.list_router_interfaces(router, interface_type)
        self.assertEqual(expected_result, ret)
        self.assert_calls()

    def test_list_router_interfaces_no_gw(self):
        """
        If a router does not have external_gateway_info, do not fail.
        """
        router = {
            'id': self.router_id
        }
        self._test_list_router_interfaces(router,
                                          interface_type="external",
                                          expected_result=[])

    def test_list_router_interfaces_gw_none(self):
        """
        If a router does have external_gateway_info set to None, do not fail.
        """
        router = {
            'id': self.router_id,
            'external_gateway_info': None
        }
        self._test_list_router_interfaces(router,
                                          interface_type="external",
                                          expected_result=[])

    def test_list_router_interfaces_all(self):
        router = {
            'id': self.router_id,
            'external_gateway_info': {
                'external_fixed_ips': [{
                    'subnet_id': 'external_subnet_id',
                    'ip_address': '1.2.3.4'}]
            }
        }
        self._test_list_router_interfaces(router,
                                          interface_type=None)

    def test_list_router_interfaces_internal(self):
        router = {
            'id': self.router_id,
            'external_gateway_info': {
                'external_fixed_ips': [{
                    'subnet_id': 'external_subnet_id',
                    'ip_address': '1.2.3.4'}]
            }
        }
        self._test_list_router_interfaces(router,
                                          interface_type="internal")

    def test_list_router_interfaces_external(self):
        router = {
            'id': self.router_id,
            'external_gateway_info': {
                'external_fixed_ips': [{
                    'subnet_id': 'external_subnet_id',
                    'ip_address': '1.2.3.4'}]
            }
        }
        self._test_list_router_interfaces(router,
                                          interface_type="external")