test_networkconfig.py

# Copyright 2013 Cloudbase Solutions Srl
#
#    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 functools
import unittest
import unittest.mock as mock

from cloudbaseinit import exception
from cloudbaseinit.models import network as network_model
from cloudbaseinit.plugins.common import base as plugin_base
from cloudbaseinit.plugins.common import networkconfig
from cloudbaseinit.tests import testutils


class TestNetworkConfigPlugin(unittest.TestCase):

    def setUp(self):
        self._setup_network_details_v1()

    @mock.patch("cloudbaseinit.osutils.factory.get_os_utils")
    def _test_execute_network_details_v1(self, mock_get_os_utils,
                                         network_adapters=None,
                                         network_details=None,
                                         invalid_details=False,
                                         missed_adapters=[],
                                         extra_network_details=[]):
        # Prepare mock environment.
        mock_service = mock.MagicMock()
        mock_shared_data = mock.Mock()
        mock_osutils = mock.MagicMock()
        mock_service.get_network_details.return_value = network_details
        mock_service.get_network_details_v2.return_value = None
        mock_get_os_utils.return_value = mock_osutils
        mock_osutils.get_network_adapters.return_value = network_adapters
        mock_osutils.set_static_network_config.return_value = True

        mock_osutils.get_network_adapter_name_by_mac_address = (
            lambda mac: [n[0] for n in network_adapters if n[1] == mac][0])

        network_execute = functools.partial(
            self._network_plugin.execute,
            mock_service, mock_shared_data
        )
        # Actual tests.
        if not network_details:
            ret = network_execute()
            self.assertEqual((plugin_base.PLUGIN_EXECUTION_DONE, False), ret)
            return
        if invalid_details or not network_adapters:
            with self.assertRaises(exception.CloudbaseInitException):
                network_execute()
            return
        # Good to go for the configuration process.
        with testutils.LogSnatcher('cloudbaseinit.plugins.'
                                   'common.networkconfig'):
            ret = network_execute()

        calls = []
        for adapter in set(network_adapters) - set(missed_adapters):
            nics = [nic for nic in (network_details +
                                    extra_network_details)
                    if nic.mac == adapter[1]]
            self.assertTrue(nics)    # missed_adapters should do the job
            nic = nics[0]
            call = mock.call(
                adapter[0],
                nic.address,
                nic.netmask,
                nic.gateway,
                nic.dnsnameservers
            )
            call6 = mock.call(
                adapter[0],
                nic.address6,
                nic.netmask6,
                nic.gateway6,
                []
            )
            if nic.address and nic.netmask:
                calls.append(call)
            if nic.address6 and nic.netmask6:
                calls.append(call6)
        self.assertEqual(
            len(calls),
            mock_osutils.set_static_network_config.call_count)
        mock_osutils.set_static_network_config.assert_has_calls(
            calls, any_order=True)
        reboot = len(missed_adapters) != self._count
        self.assertEqual((plugin_base.PLUGIN_EXECUTION_DONE, reboot), ret)

    def _setup_network_details_v1(self, same_names=True, wrong_names=False,
                                  no_macs=False):
        # Generate fake pairs of NetworkDetails objects and
        # local ethernet network adapters.
        iface_name = "Ethernet" if wrong_names else "eth"
        self._count = 3
        details_names = ["{}{}".format(iface_name, idx)
                         for idx in range(self._count)]
        if same_names:
            adapters_names = details_names[:]
        else:
            adapters_names = ["vm " + name for name in details_names]
        macs = [
            "54:EE:75:19:F4:61",
            "54:EE:75:19:F4:62",
            "54:EE:75:19:F4:63"
        ]
        addresses = [
            "192.168.122.101",
            "192.168.103.104",
            "192.168.122.105",
        ]
        addresses6 = [
            "::ffff:c0a8:7a65",
            "::ffff:c0a8:6768",
            "::ffff:c0a8:7a69"
        ]
        netmasks = [
            "255.255.255.0",
            "255.255.0.0",
            "255.255.255.128",
        ]
        netmasks6 = [
            "96",
            "64",
            "100"
        ]
        broadcasts = [
            "192.168.122.255",
            "192.168.255.255",
            "192.168.122.127",
        ]
        gateways = [
            "192.168.122.1",
            "192.168.122.16",
            "192.168.122.32",
        ]
        gateways6 = [
            "::ffff:c0a8:7a01",
            "::ffff:c0a8:7a10",
            "::ffff:c0a8:7a20"
        ]
        dnsnses = [
            "8.8.8.8",
            "8.8.8.8 8.8.4.4",
            "8.8.8.8 0.0.0.0",
        ]
        self._network_adapters = []
        self._network_details = []
        for ind in range(self._count):
            adapter = (adapters_names[ind], macs[ind])
            nic = network_model.NetworkDetails(
                details_names[ind],
                None if no_macs else macs[ind],
                addresses[ind],
                addresses6[ind],
                netmasks[ind],
                netmasks6[ind],
                broadcasts[ind],
                gateways[ind],
                gateways6[ind],
                dnsnses[ind].split()
            )
            self._network_adapters.append(adapter)
            self._network_details.append(nic)
        # Get the network config plugin.
        self._network_plugin = networkconfig.NetworkConfigPlugin()
        # Execution wrapper.
        self._partial_test_execute_network_details_v1 = functools.partial(
            self._test_execute_network_details_v1,
            network_adapters=self._network_adapters,
            network_details=self._network_details
        )

    def test_execute_no_network_details(self):
        self._network_details[:] = []
        self._partial_test_execute_network_details_v1()

    def test_execute_no_network_adapters(self):
        self._network_adapters[:] = []
        self._partial_test_execute_network_details_v1()

    def test_execute_invalid_network_details(self):
        self._network_details.append([None] * 6)
        self._partial_test_execute_network_details_v1(invalid_details=True)

    def test_execute_invalid_network_details_name(self):
        self._setup_network_details_v1(wrong_names=True, no_macs=True)
        self._partial_test_execute_network_details_v1(invalid_details=True)

    def test_execute_single(self):
        for _ in range(self._count - 1):
            self._network_adapters.pop()
            self._network_details.pop()
        self._partial_test_execute_network_details_v1()

    def test_execute_multiple(self):
        self._partial_test_execute_network_details_v1()

    def test_execute_missing_one(self):
        self.assertGreater(self._count, 1)
        self._network_details.pop(0)
        adapter = self._network_adapters[0]
        self._partial_test_execute_network_details_v1(
            missed_adapters=[adapter])

    def test_execute_missing_all(self):
        nic = self._network_details[0]
        nic = network_model.NetworkDetails(
            nic.name,
            "00" + nic.mac[2:],
            nic.address,
            nic.address6,
            nic.netmask,
            nic.netmask6,
            nic.broadcast,
            nic.gateway,
            nic.gateway6,
            nic.dnsnameservers
        )
        self._network_details[:] = [nic]
        self._partial_test_execute_network_details_v1(
            missed_adapters=self._network_adapters)

    def _test_execute_missing_smth(self, name=False, mac=False,
                                   address=False, address6=False,
                                   netmask=False, netmask6=False,
                                   gateway=False, fail=False):
        ind = self._count - 1
        nic = self._network_details[ind]
        nic2 = network_model.NetworkDetails(
            None if name else nic.name,
            None if mac else nic.mac,
            None if address else nic.address,
            None if address6 else nic.address6,
            None if netmask else nic.netmask,
            None if netmask6 else nic.netmask6,
            nic.broadcast,
            None if gateway else nic.gateway,
            None if gateway else nic.gateway6,
            nic.dnsnameservers
        )
        self._network_details[ind] = nic2
        # Excluding address and gateway switches...
        if not fail:
            # Even this way, all adapters should be configured.
            missed_adapters = []
            extra_network_details = [nic]
        else:
            # Both name and MAC are missing, so we can't make the match.
            # Or other vital details.
            missed_adapters = [self._network_adapters[ind]]
            extra_network_details = []
        self._partial_test_execute_network_details_v1(
            missed_adapters=missed_adapters,
            extra_network_details=extra_network_details
        )

    def test_execute_missing_mac(self):
        self._test_execute_missing_smth(mac=True)

    def test_execute_missing_mac2(self):
        self._setup_network_details_v1(same_names=False)
        self._test_execute_missing_smth(mac=True)

    def test_execute_missing_name_mac(self):
        self._test_execute_missing_smth(name=True, mac=True, fail=True)

    def test_execute_missing_address(self):
        self._test_execute_missing_smth(address=True)

    def test_execute_missing_netmask(self):
        self._test_execute_missing_smth(netmask=True)

    def test_execute_missing_address6(self):
        self._test_execute_missing_smth(address6=True)

    def test_execute_missing_netmask6(self):
        self._test_execute_missing_smth(netmask6=True)

    def test_execute_missing_address_netmask6(self):
        self._test_execute_missing_smth(address=True, netmask6=True,
                                        fail=True)

    def test_execute_missing_gateway(self):
        self._test_execute_missing_smth(gateway=True)

    def _get_network_details_v2(self, dns1, dns3):
        links = []
        link1 = network_model.Link(
            id=mock.sentinel.link_id1,
            name=mock.sentinel.link_name1,
            type=network_model.LINK_TYPE_PHYSICAL,
            enabled=mock.sentinel.link_enabled1,
            mac_address=mock.sentinel.link_mac1,
            mtu=mock.sentinel.link_mtu1,
            bond=None,
            vlan_link=None,
            vlan_id=None)
        links.append(link1)

        bond1 = network_model.Bond(
            members=[mock.sentinel.link_id1],
            type=mock.sentinel.bond_type1,
            lb_algorithm=mock.sentinel.bond_lb_algo1,
            lacp_rate=mock.sentinel.lacp_rate1)

        bond_link1 = network_model.Link(
            id=mock.sentinel.bond_link_id1,
            name=mock.sentinel.bond_link_name1,
            type=network_model.LINK_TYPE_BOND,
            enabled=mock.sentinel.bond_link_enabled1,
            mac_address=mock.sentinel.bond_link_mac1,
            mtu=mock.sentinel.bond_link_mtu1,
            bond=bond1,
            vlan_link=None,
            vlan_id=None)
        links.append(bond_link1)

        vlan_link1 = network_model.Link(
            id=mock.sentinel.vlan_link_id1,
            name=mock.sentinel.vlan_link_name1,
            type=network_model.LINK_TYPE_VLAN,
            enabled=mock.sentinel.vlan_link_enabled1,
            mac_address=mock.sentinel.vlan_link_mac1,
            mtu=mock.sentinel.vlan_link_mtu1,
            bond=None,
            vlan_link=mock.sentinel.bond_link_id1,
            vlan_id=mock.sentinel.vlan_id1)
        links.append(vlan_link1)

        networks = []
        route1 = network_model.Route(
            network_cidr=mock.sentinel.network_cidr1,
            gateway=mock.sentinel.gateway1)

        route2 = network_model.Route(
            network_cidr=mock.sentinel.network_cidr2,
            gateway=mock.sentinel.gateway2)

        network1 = network_model.Network(
            link=mock.sentinel.link_id1,
            address_cidr=mock.sentinel.address_cidr1,
            dns_nameservers=mock.sentinel.network_dns_list1,
            routes=[route1, route2])
        networks.append(network1)

        services = []
        service1 = network_model.NameServerService(
            addresses=[dns1, dns3],
            search=mock.sentinel.dns_search1)
        services.append(service1)

        return network_model.NetworkDetailsV2(
            links=links, networks=networks, services=services)

    @mock.patch("cloudbaseinit.osutils.factory.get_os_utils")
    def _test_execute_network_details_v2(self, mock_get_os_utils,
                                         empty_network_dns_list=False,
                                         both_ipv4_dns_list=False,
                                         both_ipv6_dns_list=False):
        mock.sentinel.link_mac1 = u"00:00:00:00:00:01"
        mock.sentinel.network_cidr1 = u"0.0.0.0/0"
        mock.sentinel.gateway1 = u"10.0.0.254"
        mock.sentinel.network_cidr2 = u"172.16.0.0/16"
        mock.sentinel.gateway2 = u"172.16.1.1"
        mock.sentinel.address_cidr1 = u"10.0.0.1/24"
        dns1 = "10.0.0.1"
        dns3 = "10.0.0.3"
        dns3_ipv6 = "2001:db8::3"
        dns3_ipv6_v2 = "2001:db8::4"
        mock.sentinel.dns1 = dns1
        mock.sentinel.dns2 = "10.0.0.2"
        mock.sentinel.network_dns_list1 = []

        if empty_network_dns_list:
            mock.sentinel.dns3 = dns3
            expected_dns_list = [mock.sentinel.dns1, mock.sentinel.dns3]
        elif both_ipv4_dns_list:
            mock.sentinel.dns3 = dns3_ipv6
            expected_dns_list = [mock.sentinel.dns1]
            dns3 = dns3_ipv6
        elif both_ipv6_dns_list:
            mock.sentinel.address_cidr1 = u"2001:db8::3/24"
            mock.sentinel.dns3 = dns3_ipv6_v2
            dns3 = dns3_ipv6_v2
            expected_dns_list = [mock.sentinel.dns3]
        else:
            mock.sentinel.network_dns_list1 = [
                mock.sentinel.dns1, mock.sentinel.dns2]
            expected_dns_list = mock.sentinel.network_dns_list1

        service = mock.Mock()
        network_details = self._get_network_details_v2(dns1, dns3)
        service.get_network_details_v2.return_value = network_details

        mock_os_utils = mock.Mock()
        mock_get_os_utils.return_value = mock_os_utils

        m = mock_os_utils.get_network_adapter_name_by_mac_address
        m.return_value = mock.sentinel.adapter_old_name1

        plugin = networkconfig.NetworkConfigPlugin()
        plugin.execute(service, {})

        service.get_network_details_v2.assert_called_once_with()
        service.get_network_details.assert_not_called()

        m.assert_called_once_with(mock.sentinel.link_mac1)

        mock_os_utils.rename_network_adapter.assert_called_once_with(
            mock.sentinel.adapter_old_name1, mock.sentinel.link_name1)

        bond_name = (networkconfig.BOND_FORMAT_STR %
                     mock.sentinel.bond_link_id1)
        mock_os_utils.create_network_team.assert_called_once_with(
            bond_name, mock.sentinel.bond_type1,
            mock.sentinel.bond_lb_algo1,
            [mock.sentinel.link_id1],
            mock.sentinel.bond_link_mac1,
            mock.sentinel.bond_link_name1,
            None,
            mock.sentinel.lacp_rate1)

        mock_os_utils.add_network_team_nic.assert_called_once_with(
            bond_name,
            mock.sentinel.vlan_link_name1,
            mock.sentinel.vlan_id1)

        mock_os_utils.set_network_adapter_mtu.assert_has_calls(
            [mock.call(mock.sentinel.link_name1, mock.sentinel.link_mtu1),
             mock.call(
                mock.sentinel.bond_link_name1, mock.sentinel.bond_link_mtu1),
             mock.call(
                mock.sentinel.vlan_link_name1, mock.sentinel.vlan_link_mtu1)],
            any_order=False)

        mock_os_utils.enable_network_adapter.assert_has_calls(
            [mock.call(mock.sentinel.link_name1, mock.sentinel.link_enabled1),
             mock.call(
                mock.sentinel.bond_link_name1,
                mock.sentinel.bond_link_enabled1),
             mock.call(
                mock.sentinel.vlan_link_name1,
                mock.sentinel.vlan_link_enabled1)],
            any_order=False)

        ip_address, prefix_len = mock.sentinel.address_cidr1.split("/")
        # When address is IPv6 but gateway is IPv4-only, gateway should
        # be None because the address family filter excludes it.
        expected_gateway = mock.sentinel.gateway1
        if ":" in ip_address:
            expected_gateway = None
        mock_os_utils.set_static_network_config.assert_called_once_with(
            mock.sentinel.link_id1, ip_address, prefix_len,
            expected_gateway, expected_dns_list)

    def test_execute_network_details_v2(self):
        self._test_execute_network_details_v2()

    def test_execute_network_details_v2_empty_network_dns_list(self):
        self._test_execute_network_details_v2(empty_network_dns_list=True)

    def test_execute_network_details_v2_ipv4_dns_list(self):
        self._test_execute_network_details_v2(both_ipv4_dns_list=True)

    def test_execute_network_details_v2_ipv6_dns_list(self):
        self._test_execute_network_details_v2(both_ipv6_dns_list=True)

    @mock.patch("cloudbaseinit.osutils.factory.get_os_utils")
    def test_execute_network_details_v2_dual_stack_gateways(
            self, mock_get_os_utils):
        """IPv4 address gets IPv4 gateway, IPv6 address gets IPv6 gateway."""
        link1 = network_model.Link(
            id="eth0",
            name="eth0",
            type=network_model.LINK_TYPE_PHYSICAL,
            enabled=True,
            mac_address=u"00:00:00:00:00:01",
            mtu=None,
            bond=None,
            vlan_link=None,
            vlan_id=None)

        route_v4 = network_model.Route(
            network_cidr=u"0.0.0.0/0",
            gateway=u"169.254.0.1")
        route_v6 = network_model.Route(
            network_cidr=u"::/0",
            gateway=u"fe80::1")

        network_v4 = network_model.Network(
            link="eth0",
            address_cidr=u"10.0.0.1/32",
            dns_nameservers=["1.1.1.1"],
            routes=[route_v4, route_v6])
        network_v6 = network_model.Network(
            link="eth0",
            address_cidr=u"2001:db8::1/96",
            dns_nameservers=["1.1.1.1"],
            routes=[route_v4, route_v6])

        network_details = network_model.NetworkDetailsV2(
            links=[link1],
            networks=[network_v4, network_v6],
            services=[])

        service = mock.Mock()
        service.get_network_details_v2.return_value = network_details

        mock_os_utils = mock.Mock()
        mock_get_os_utils.return_value = mock_os_utils
        mock_os_utils.get_network_adapter_name_by_mac_address.return_value = \
            "eth0"

        plugin = networkconfig.NetworkConfigPlugin()
        plugin.execute(service, {})

        calls = mock_os_utils.set_static_network_config.call_args_list
        self.assertEqual(len(calls), 2)
        # IPv4 address should get IPv4 gateway
        self.assertEqual(calls[0], mock.call(
            "eth0", "10.0.0.1", "32", "169.254.0.1", ["1.1.1.1"]))
        # IPv6 address should get IPv6 gateway
        self.assertEqual(calls[1], mock.call(
            "eth0", "2001:db8::1", "96", "fe80::1", ["1.1.1.1"]))