kuryr-kubernetes/kuryr_kubernetes/tests/unit/test_os_vif_util.py

# Copyright (c) 2016 Mirantis, 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 mock
import uuid

from os_vif.objects import fixed_ip as osv_fixed_ip
from os_vif.objects import network as osv_network
from os_vif.objects import route as osv_route
from os_vif.objects import subnet as osv_subnet
from oslo_config import cfg as o_cfg

from kuryr_kubernetes import constants as const
from kuryr_kubernetes import exceptions as k_exc
from kuryr_kubernetes import os_vif_util as ovu
from kuryr_kubernetes.tests import base as test_base


# REVISIT(ivc): move to kuryr-lib along with 'os_vif_util'


class TestOSVIFUtils(test_base.TestCase):
    def test_neutron_to_osvif_network(self):
        network_id = str(uuid.uuid4())
        network_name = 'test-net'
        network_mtu = 1500
        neutron_network = {
            'id': network_id,
            'name': network_name,
            'mtu': network_mtu,
        }

        network = ovu.neutron_to_osvif_network(neutron_network)

        self.assertEqual(network_id, network.id)
        self.assertEqual(network_name, network.label)
        self.assertEqual(network_mtu, network.mtu)

    def test_neutron_to_osvif_network_no_name(self):
        network_id = str(uuid.uuid4())
        network_mtu = 1500
        neutron_network = {
            'id': network_id,
            'mtu': network_mtu,
        }

        network = ovu.neutron_to_osvif_network(neutron_network)

        self.assertFalse(network.obj_attr_is_set('label'))

    def test_neutron_to_osvif_network_no_mtu(self):
        network_id = str(uuid.uuid4())
        network_name = 'test-net'
        neutron_network = {
            'id': network_id,
            'name': network_name,
        }

        network = ovu.neutron_to_osvif_network(neutron_network)

        self.assertIsNone(network.mtu)

    @mock.patch('kuryr_kubernetes.os_vif_util._neutron_to_osvif_routes')
    def test_neutron_to_osvif_subnet(self, m_conv_routes):
        gateway = '1.1.1.1'
        cidr = '1.1.1.1/8'
        dns = ['2.2.2.2', '3.3.3.3']
        host_routes = mock.sentinel.host_routes
        route_list = osv_route.RouteList(objects=[
            osv_route.Route(cidr='4.4.4.4/8', gateway='5.5.5.5')])
        m_conv_routes.return_value = route_list
        neutron_subnet = {
            'cidr': cidr,
            'dns_nameservers': dns,
            'host_routes': host_routes,
            'gateway_ip': gateway,
        }

        subnet = ovu.neutron_to_osvif_subnet(neutron_subnet)

        self.assertEqual(cidr, str(subnet.cidr))
        self.assertEqual(route_list, subnet.routes)
        self.assertEqual(set(dns), set([str(addr) for addr in subnet.dns]))
        self.assertEqual(gateway, str(subnet.gateway))
        m_conv_routes.assert_called_once_with(host_routes)

    @mock.patch('kuryr_kubernetes.os_vif_util._neutron_to_osvif_routes')
    def test_neutron_to_osvif_subnet_no_gateway(self, m_conv_routes):
        cidr = '1.1.1.1/8'
        route_list = osv_route.RouteList()
        m_conv_routes.return_value = route_list
        neutron_subnet = {
            'cidr': cidr,
            'dns_nameservers': [],
            'host_routes': [],
        }

        subnet = ovu.neutron_to_osvif_subnet(neutron_subnet)

        self.assertFalse(subnet.obj_attr_is_set('gateway'))

    def test_neutron_to_osvif_routes(self):
        routes_map = {'%s.0.0.0/8' % i: '10.0.0.%s' % i for i in range(3)}
        routes = [{'destination': k, 'nexthop': v}
                  for k, v in routes_map.items()]

        route_list = ovu._neutron_to_osvif_routes(routes)

        self.assertEqual(len(routes), len(route_list.objects))
        for route in route_list.objects:
            self.assertEqual(routes_map[str(route.cidr)], str(route.gateway))

    @mock.patch('kuryr_kubernetes.os_vif_util._VIF_MANAGERS')
    def test_neutron_to_osvif_vif(self, m_mgrs):
        vif_plugin = mock.sentinel.vif_plugin
        port = mock.sentinel.port
        subnets = mock.sentinel.subnets
        m_mgr = mock.Mock()
        m_mgrs.__getitem__.return_value = m_mgr

        ovu.neutron_to_osvif_vif(vif_plugin, port, subnets)

        m_mgrs.__getitem__.assert_called_with(vif_plugin)
        m_mgr.driver.assert_called_with(vif_plugin, port, subnets)

    @mock.patch('stevedore.driver.DriverManager')
    @mock.patch('kuryr_kubernetes.os_vif_util._VIF_MANAGERS')
    def test_neutron_to_osvif_vif_load(self, m_mgrs, m_stv_drm):
        vif_plugin = mock.sentinel.vif_plugin
        port = mock.sentinel.port
        subnets = mock.sentinel.subnets
        m_mgr = mock.Mock()
        m_mgrs.__getitem__.side_effect = KeyError
        m_stv_drm.return_value = m_mgr

        ovu.neutron_to_osvif_vif(vif_plugin, port, subnets)

        m_stv_drm.assert_called_once_with(
            namespace=ovu._VIF_TRANSLATOR_NAMESPACE,
            name=vif_plugin,
            invoke_on_load=False)
        m_mgrs.__setitem__.assert_called_once_with(vif_plugin, m_mgr)
        m_mgr.driver.assert_called_once_with(vif_plugin, port, subnets)

    @mock.patch('kuryr_kubernetes.os_vif_util._get_ovs_hybrid_bridge_name')
    @mock.patch('kuryr_kubernetes.os_vif_util._get_vif_name')
    @mock.patch('kuryr_kubernetes.os_vif_util._is_port_active')
    @mock.patch('kuryr_kubernetes.os_vif_util._make_vif_network')
    @mock.patch('os_vif.objects.vif.VIFBridge')
    @mock.patch('os_vif.objects.vif.VIFPortProfileOpenVSwitch')
    def test_neutron_to_osvif_vif_ovs_hybrid(self,
                                             m_mk_profile,
                                             m_mk_vif,
                                             m_make_vif_network,
                                             m_is_port_active,
                                             m_get_vif_name,
                                             m_get_ovs_hybrid_bridge_name):
        vif_plugin = 'ovs'
        port_id = mock.sentinel.port_id
        mac_address = mock.sentinel.mac_address
        ovs_bridge = mock.sentinel.ovs_bridge
        port_filter = mock.sentinel.port_filter
        subnets = mock.sentinel.subnets
        port_profile = mock.sentinel.port_profile
        network = mock.sentinel.network
        port_active = mock.sentinel.port_active
        vif_name = mock.sentinel.vif_name
        hybrid_bridge = mock.sentinel.hybrid_bridge
        vif = mock.sentinel.vif

        m_mk_profile.return_value = port_profile
        m_make_vif_network.return_value = network
        m_is_port_active.return_value = port_active
        m_get_vif_name.return_value = vif_name
        m_get_ovs_hybrid_bridge_name.return_value = hybrid_bridge
        m_mk_vif.return_value = vif

        port = {'id': port_id,
                'mac_address': mac_address,
                'binding:vif_details': {
                    'ovs_hybrid_plug': True,
                    'bridge_name': ovs_bridge,
                    'port_filter': port_filter},
                }

        self.assertEqual(vif, ovu.neutron_to_osvif_vif_ovs(vif_plugin, port,
                                                           subnets))

        m_mk_profile.assert_called_once_with(interface_id=port_id)
        m_make_vif_network.assert_called_once_with(port, subnets)
        m_is_port_active.assert_called_once_with(port)
        m_get_ovs_hybrid_bridge_name.assert_called_once_with(port)
        m_get_vif_name.assert_called_once_with(port)
        self.assertEqual(ovs_bridge, network.bridge)
        m_mk_vif.assert_called_once_with(
            id=port_id,
            address=mac_address,
            network=network,
            has_traffic_filtering=port_filter,
            preserve_on_delete=False,
            active=port_active,
            port_profile=port_profile,
            plugin=vif_plugin,
            vif_name=vif_name,
            bridge_name=hybrid_bridge)

    @mock.patch('kuryr_kubernetes.os_vif_util._get_vif_name')
    @mock.patch('kuryr_kubernetes.os_vif_util._is_port_active')
    @mock.patch('kuryr_kubernetes.os_vif_util._make_vif_network')
    @mock.patch('os_vif.objects.vif.VIFOpenVSwitch')
    @mock.patch('os_vif.objects.vif.VIFPortProfileOpenVSwitch')
    def test_neutron_to_osvif_vif_ovs_native(self,
                                             m_mk_profile,
                                             m_mk_vif,
                                             m_make_vif_network,
                                             m_is_port_active,
                                             m_get_vif_name):
        vif_plugin = 'ovs'
        port_id = mock.sentinel.port_id
        mac_address = mock.sentinel.mac_address
        ovs_bridge = mock.sentinel.ovs_bridge
        subnets = mock.sentinel.subnets
        port_profile = mock.sentinel.port_profile
        network = mock.sentinel.network
        port_active = mock.sentinel.port_active
        vif_name = mock.sentinel.vif_name
        vif = mock.sentinel.vif

        m_mk_profile.return_value = port_profile
        m_make_vif_network.return_value = network
        m_is_port_active.return_value = port_active
        m_get_vif_name.return_value = vif_name
        m_mk_vif.return_value = vif

        port = {'id': port_id,
                'mac_address': mac_address,
                'binding:vif_details': {
                    'ovs_hybrid_plug': False,
                    'bridge_name': ovs_bridge},
                }

        self.assertEqual(vif, ovu.neutron_to_osvif_vif_ovs(vif_plugin, port,
                                                           subnets))
        m_mk_profile.assert_called_once_with(interface_id=port_id)
        m_make_vif_network.assert_called_once_with(port, subnets)
        m_is_port_active.assert_called_once_with(port)
        m_get_vif_name.assert_called_once_with(port)
        self.assertEqual(ovs_bridge, network.bridge)

    @mock.patch('kuryr_kubernetes.os_vif_util._get_vif_name')
    @mock.patch('kuryr_kubernetes.os_vif_util._is_port_active')
    @mock.patch('kuryr_kubernetes.os_vif_util._make_vif_network')
    @mock.patch('kuryr_kubernetes.objects.vif.VIFVlanNested')
    def test_neutron_to_osvif_nested_vlan(self, m_mk_vif, m_make_vif_network,
                                          m_is_port_active, m_get_vif_name):
        vif_plugin = const.K8S_OS_VIF_NOOP_PLUGIN
        port_id = mock.sentinel.port_id
        mac_address = mock.sentinel.mac_address
        port_filter = mock.sentinel.port_filter
        subnets = mock.sentinel.subnets
        network = mock.sentinel.network
        port_active = mock.sentinel.port_active
        vif_name = mock.sentinel.vif_name
        vif = mock.sentinel.vif
        vlan_id = mock.sentinel.vlan_id

        m_make_vif_network.return_value = network
        m_is_port_active.return_value = port_active
        m_get_vif_name.return_value = vif_name
        m_mk_vif.return_value = vif

        port = {'id': port_id,
                'mac_address': mac_address,
                'binding:vif_details': {
                    'port_filter': port_filter},
                }

        self.assertEqual(vif, ovu.neutron_to_osvif_vif_nested_vlan(port,
                         subnets, vlan_id))

        m_make_vif_network.assert_called_once_with(port, subnets)
        m_is_port_active.assert_called_once_with(port)
        m_get_vif_name.assert_called_once_with(port)
        m_mk_vif.assert_called_once_with(
            id=port_id,
            address=mac_address,
            network=network,
            has_traffic_filtering=port_filter,
            preserve_on_delete=False,
            active=port_active,
            plugin=vif_plugin,
            vif_name=vif_name,
            vlan_id=vlan_id)

    @mock.patch('kuryr_kubernetes.os_vif_util._get_vif_name')
    @mock.patch('kuryr_kubernetes.os_vif_util._is_port_active')
    @mock.patch('kuryr_kubernetes.os_vif_util._make_vif_network')
    @mock.patch('kuryr_kubernetes.objects.vif.VIFMacvlanNested')
    def test_neutron_to_osvif_nested_macvlan(self, m_mk_vif,
                                             m_make_vif_network,
                                             m_is_port_active, m_get_vif_name):
        vif_plugin = const.K8S_OS_VIF_NOOP_PLUGIN
        port_id = mock.sentinel.port_id
        mac_address = mock.sentinel.mac_address
        port_filter = mock.sentinel.port_filter
        subnets = mock.sentinel.subnets
        network = mock.sentinel.network
        port_active = mock.sentinel.port_active
        vif_name = mock.sentinel.vif_name
        vif = mock.sentinel.vif

        m_make_vif_network.return_value = network
        m_is_port_active.return_value = port_active
        m_get_vif_name.return_value = vif_name
        m_mk_vif.return_value = vif

        port = {'id': port_id,
                'mac_address': mac_address,
                'binding:vif_details': {
                    'port_filter': port_filter},
                }

        self.assertEqual(vif, ovu.neutron_to_osvif_vif_nested_macvlan(port,
                                                                      subnets))

        m_make_vif_network.assert_called_once_with(port, subnets)
        m_is_port_active.assert_called_once_with(port)
        m_get_vif_name.assert_called_once_with(port)
        m_mk_vif.assert_called_once_with(
            id=port_id,
            address=mac_address,
            network=network,
            has_traffic_filtering=port_filter,
            preserve_on_delete=False,
            active=port_active,
            plugin=vif_plugin,
            vif_name=vif_name)

    def test_neutron_to_osvif_vif_ovs_no_bridge(self):
        vif_plugin = 'ovs'
        port = {'id': str(uuid.uuid4())}
        subnets = {}

        self.assertRaises(o_cfg.RequiredOptError,
                          ovu.neutron_to_osvif_vif_ovs,
                          vif_plugin, port, subnets)

    def test_get_ovs_hybrid_bridge_name(self):
        port_id = str(uuid.uuid4())
        port = {'id': port_id}

        self.assertEqual("qbr" + port_id[:11],
                         ovu._get_ovs_hybrid_bridge_name(port))

    def test_is_port_active(self):
        port = {'status': 'ACTIVE'}

        self.assertTrue(ovu._is_port_active(port))

    def test_is_port_inactive(self):
        port = {'status': 'DOWN'}

        self.assertFalse(ovu._is_port_active(port))

    @mock.patch('kuryr.lib.binding.drivers.utils.get_veth_pair_names')
    def test_get_vif_name(self, m_get_veth_pair_names):
        port_id = mock.sentinel.port_id
        vif_name = mock.sentinel.vif_name
        port = {'id': port_id}
        m_get_veth_pair_names.return_value = (vif_name, mock.sentinel.any)

        self.assertEqual(vif_name, ovu._get_vif_name(port))
        m_get_veth_pair_names.assert_called_once_with(port_id)

    @mock.patch('kuryr_kubernetes.os_vif_util._make_vif_subnets')
    @mock.patch('os_vif.objects.subnet.SubnetList')
    def test_make_vif_network(self, m_mk_subnet_list, m_make_vif_subnets):
        network_id = mock.sentinel.network_id
        network = mock.Mock()
        orig_network = mock.Mock()
        orig_network.id = network_id
        orig_network.obj_clone.return_value = network
        subnet_id = mock.sentinel.subnet_id
        subnets = {subnet_id: orig_network}
        vif_subnets = mock.sentinel.vif_subnets
        subnet_list = mock.sentinel.subnet_list
        m_make_vif_subnets.return_value = vif_subnets
        m_mk_subnet_list.return_value = subnet_list
        port = {'network_id': network_id}

        self.assertEqual(network, ovu._make_vif_network(port, subnets))
        self.assertEqual(subnet_list, network.subnets)
        m_make_vif_subnets.assert_called_once_with(port, subnets)
        m_mk_subnet_list.assert_called_once_with(objects=vif_subnets)

    def test_make_vif_network_not_found(self):
        network_id = mock.sentinel.network_id
        port = {'network_id': network_id}
        subnets = {}

        self.assertRaises(k_exc.IntegrityError, ovu._make_vif_network,
                          port, subnets)

    @mock.patch('kuryr_kubernetes.os_vif_util._make_vif_subnet')
    @mock.patch('os_vif.objects.fixed_ip.FixedIP')
    def test_make_vif_subnets(self, m_mk_fixed_ip, m_make_vif_subnet):
        subnet_id = mock.sentinel.subnet_id
        ip_address = mock.sentinel.ip_address
        fixed_ip = mock.sentinel.fixed_ip
        subnet = mock.Mock()
        subnets = mock.MagicMock()
        subnets.__contains__.return_value = True
        m_mk_fixed_ip.return_value = fixed_ip
        m_make_vif_subnet.return_value = subnet
        port = {'fixed_ips': [
            {'subnet_id': subnet_id, 'ip_address': ip_address}]}

        self.assertEqual([subnet], ovu._make_vif_subnets(port, subnets))
        m_make_vif_subnet.assert_called_once_with(subnets, subnet_id)
        m_mk_fixed_ip.assert_called_once_with(address=ip_address)
        subnet.ips.objects.append.assert_called_once_with(fixed_ip)

    def test_make_vif_subnets_not_found(self):
        subnet_id = mock.sentinel.subnet_id
        ip_address = mock.sentinel.ip_address
        subnets = mock.MagicMock()
        subnets.__contains__.return_value = False
        port = {'fixed_ips': [
            {'subnet_id': subnet_id, 'ip_address': ip_address}]}

        self.assertRaises(k_exc.IntegrityError, ovu._make_vif_subnets,
                          port, subnets)

    @mock.patch('os_vif.objects.fixed_ip.FixedIPList')
    def test_make_vif_subnet(self, m_mk_fixed_ip_list):
        subnet_id = mock.sentinel.subnet_id
        fixed_ip_list = mock.sentinel.fixed_ip_list
        subnet = mock.Mock()
        orig_subnet = mock.Mock()
        orig_subnet.obj_clone.return_value = subnet
        orig_network = mock.Mock()
        orig_network.subnets.objects = [orig_subnet]
        m_mk_fixed_ip_list.return_value = fixed_ip_list
        subnets = {subnet_id: orig_network}

        self.assertEqual(subnet, ovu._make_vif_subnet(subnets, subnet_id))
        self.assertEqual(fixed_ip_list, subnet.ips)
        m_mk_fixed_ip_list.assert_called_once_with(objects=[])

    def test_make_vif_subnet_invalid(self):
        subnet_id = mock.sentinel.subnet_id
        orig_network = mock.Mock()
        orig_network.subnets.objects = []
        subnets = {subnet_id: orig_network}

        self.assertRaises(k_exc.IntegrityError, ovu._make_vif_subnet,
                          subnets, subnet_id)

    def test_osvif_to_neutron_network_ids(self):
        id_a = mock.sentinel.id_a
        id_b = mock.sentinel.id_b
        net1 = mock.Mock()
        net1.id = id_a
        net2 = mock.Mock()
        net2.id = id_b
        net3 = mock.Mock()
        net3.id = id_a
        subnets = {1: net1, 2: net2, 3: net3}

        ret = ovu.osvif_to_neutron_network_ids(subnets)
        self.assertEqual(2, len(ret))
        self.assertIn(id_a, ret)
        self.assertIn(id_b, ret)

    def test_osvif_to_neutron_fixed_ips(self):
        ip11 = '1.1.1.1'
        ip12 = '2.2.2.2'
        ip3 = '3.3.3.3'
        subnet_id_1 = str(uuid.uuid4())
        subnet_id_2 = str(uuid.uuid4())
        subnet_id_3 = str(uuid.uuid4())

        subnet_1 = osv_subnet.Subnet(ips=osv_fixed_ip.FixedIPList(
            objects=[osv_fixed_ip.FixedIP(address=ip11),
                     osv_fixed_ip.FixedIP(address=ip12)]))
        subnet_2 = osv_subnet.Subnet()
        subnet_3 = osv_subnet.Subnet(ips=osv_fixed_ip.FixedIPList(
            objects=[osv_fixed_ip.FixedIP(address=ip3)]))

        net1 = osv_network.Network(subnets=osv_subnet.SubnetList(
            objects=[subnet_1]))
        net2 = osv_network.Network(subnets=osv_subnet.SubnetList(
            objects=[subnet_2]))
        net3 = osv_network.Network(subnets=osv_subnet.SubnetList(
            objects=[subnet_3]))

        subnets = {subnet_id_1: net1, subnet_id_2: net2, subnet_id_3: net3}

        expected = [{'subnet_id': subnet_id_1, 'ip_address': ip11},
                    {'subnet_id': subnet_id_1, 'ip_address': ip12},
                    {'subnet_id': subnet_id_2},
                    {'subnet_id': subnet_id_3, 'ip_address': ip3}]

        ret = ovu.osvif_to_neutron_fixed_ips(subnets)

        def _sort_key(e):
            return (e.get('subnet_id'), e.get('ip_address'))

        self.assertEqual(sorted(expected, key=_sort_key),
                         sorted(ret, key=_sort_key))

    def test_osvif_to_neutron_fixed_ips_invalid(self):
        subnet_id = str(uuid.uuid4())

        subnet_1 = osv_subnet.Subnet()
        subnet_2 = osv_subnet.Subnet()

        net = osv_network.Network(subnets=osv_subnet.SubnetList(
            objects=[subnet_1, subnet_2]))

        subnets = {subnet_id: net}

        self.assertRaises(k_exc.IntegrityError,
                          ovu.osvif_to_neutron_fixed_ips, subnets)