path: root/tests/vmtests/test_network.py

from . import CloudInitVMBaseClass, logger
from .releases import base_vm_classes as relbase

import ipaddress
import os
import re
from subprocess import check_output
import textwrap
import yaml


def cidr2mask(cidr):
    mask = [0, 0, 0, 0]
    for i in list(range(0, cidr)):
        idx = int(i / 8)
        mask[idx] = mask[idx] + (1 << (7 - i % 8))
    return ".".join([str(x) for x in mask])

def ipv4mask2cidr(mask):
    if '.' not in mask:
        return mask
    return sum([bin(int(x)).count('1') for x in mask.split('.')])


def ipv6mask2cidr(mask):
    if ':' not in mask:
        return mask

    bitCount = [0, 0x8000, 0xc000, 0xe000, 0xf000, 0xf800, 0xfc00, 0xfe00,
                0xff00, 0xff80, 0xffc0, 0xffe0, 0xfff0, 0xfff8, 0xfffc,
                0xfffe, 0xffff]
    cidr = 0
    for word in mask.split(':'):
        if not word or int(word, 16) == 0:
            break
        cidr += bitCount.index(int(word, 16))

    return cidr


def mask2cidr(mask):
    if ':' in mask:
        return ipv6mask2cidr(mask)
    elif '.' in mask:
        return ipv4mask2cidr(mask)
    else:
        return mask


def iface_extract(input):
    mo = re.search(r'^(?P<interface>\w+|\w+:\d+)\s+' +
                   r'Link encap:(?P<link_encap>\S+)\s+' +
                   r'(HWaddr\s+(?P<mac_address>\S+))?' +
                   r'(\s+inet addr:(?P<address>\S+))?' +
                   r'(\s+Bcast:(?P<broadcast>\S+)\s+)?' +
                   r'(Mask:(?P<netmask>\S+)\s+)?',
                   input, re.MULTILINE)

    mtu = re.search(r'(\s+MTU:(?P<mtu>\d+)\s+)\s+', input, re.MULTILINE)
    mtu_info = mtu.groupdict('')
    mtu_info['mtu'] = int(mtu_info['mtu'])

    if mo:
        info = mo.groupdict('')
        info['running'] = False
        info['up'] = False
        info['multicast'] = False
        if 'RUNNING' in input:
            info['running'] = True
        if 'UP' in input:
            info['up'] = True
        if 'MULTICAST' in input:
            info['multicast'] = True
        info.update(mtu_info)
        return info
    return {}


def ifconfig_to_dict(ifconfig):
    interfaces = {}
    for iface in [iface_extract(iface) for iface in ifconfig.split('\n\n')
                  if iface.strip()]:
        interfaces[iface['interface']] = iface

    return interfaces


class TestNetworkAbs(CloudInitVMBaseClass):
    interactive = False
    conf_file = "configs/basic_network.yaml"
    install_timeout = 600
    boot_timeout = 600
    extra_disks = []
    extra_nics = []
    collect_scripts = [textwrap.dedent("""
        cd OUTPUT_COLLECT_D
        ls -al /dev/disk/by-id/*
        cp -a /var/log/cloud-init* .
        cp -a /var/lib/cloud .
        ifconfig -a > ifconfig_a
        cp -a /etc/network/interfaces .
        cp -r /etc/network/interfaces.d .
        cp /etc/resolv.conf .
        cp -a /etc/udev/rules.d/70-persistent-net.rules .
        ip -o route show > ip_route_show
        route -n > route_n
        ls -al /mnt/output
        cat /proc/partitions > proc_partitions
        cat /proc/cmdline > proc_cmdline
        dpkg-query --show cloud-init > dpkg_cloud_init
        journalctl -o short-precise > journalctl_short_precise
        cp /run/cloud-init/* . || :
        cp -r /etc/systemd/network systemd-network
        """)]

    def test_output_files_exist(self):
        self.output_files_exist(["ifconfig_a",
                                 "interfaces",
                                 "interfaces.d/50-cloud-init.cfg",
                                 "resolv.conf",
                                 "ip_route_show",
                                 "route_n",
                                 "dpkg_cloud_init"])

    def _read_collect_file(self, collect_file):
        with open(os.path.join(self.td.collect, collect_file)) as fp:
            data = fp.read()
        return data

    def test_cloud_init_version(self):
        installed_version = self._read_collect_file("dpkg_cloud_init")

        dpkg_info = check_output(["dpkg", "--info",
                                 self.cloud_init_deb]).decode('utf-8')
        match = re.search(r'Version:.*', dpkg_info)
        self.assertIsNotNone(match)
        self.assertIsNotNone(match.group(0))
        injected_version = match.group(0).split()[-1]
        self.assertIn(injected_version, installed_version)

    def test_cloud_init_interface_rename(self):
        cloud_init_log = self._read_collect_file("cloud-init.log")
        match = re.search(r'Running command.*ip.*link.*set.*name.*\]',
                          cloud_init_log)
        self.assertIsNotNone(match)
        self.assertIsNotNone(match.group(0))

    def test_etc_network_interfaces(self):
        ci_eni = "interfaces.d/50-cloud-init.cfg"
        eni = self._read_collect_file(ci_eni)
        logger.debug('etc/network/{}\n{}'.format(ci_eni, eni))

        expected_eni = self.__class__.get_expected_etc_network_interfaces()
        eni_lines = eni.split('\n')
        print("expected:\n%s" % expected_eni)
        print("found:\n%s" % eni)
        for line in expected_eni.split('\n'):
            # in cloud-init, we don't emit the source line since the
            # output from cloud-init is expected to be sourced.
            if line in ["source /etc/network/interfaces.d/*.cfg"]:
                continue
            elif "netmask" in line:
                if "netmask" not in eni:
                    # if the found data includes netmask we can skip
                    # this conversion
                    cidr = mask2cidr(line.split()[-1])
                    self.assertIn("/%s" % cidr, eni)
            elif "/" in line:
                if "netmask" in eni:
                    # if we find cidr notation in expected, then
                    address, cidr = line.split("/")
                    netmask = cidr2mask(int(cidr))
                    self.assertIn("    address %s" % address, eni)
                    self.assertIn("    netmask %s" % netmask, eni)
            else:
                self.assertIn(line, eni)

    def test_etc_resolvconf(self):
        resolvconf = self._read_collect_file("resolv.conf")
        logger.debug('etc/resolv.conf:\n{}'.format(resolvconf))

        resolv_lines = resolvconf.split('\n')
        logger.debug('resolv.conf lines:\n{}'.format(resolv_lines))
        # resolv.conf
        '''
        nameserver X.Y.Z.A
        nameserver 1.2.3.4
        search foo.bar
        '''

        # eni
        ''''
        auto eth1:1
        iface eth1:1 inet static
            dns-nameserver X.Y.Z.A
            dns-search foo.bar
        '''

        # iface dict
        ''''
        eth1:1:
          dns:
            nameserver: X.Y.Z.A
            search: foo.bar
        '''
        expected_ifaces = self.get_expected_etc_resolvconf()
        logger.debug('parsed eni ifaces:\n{}'.format(expected_ifaces))
        for ifname in expected_ifaces.keys():
            iface = expected_ifaces.get(ifname)
            for k, v in iface.get('dns', {}).items():
                dns_line = '{} {}'.format(
                    k.replace('nameservers', 'nameserver'), " ".join(v))
                logger.debug('dns_line:{}'.format(dns_line))
                self.assertTrue(dns_line in resolv_lines)

    def test_ifconfig_output(self):
        '''check ifconfig output with test input'''
        network_state = self.get_network_state()
        logger.debug('expected_network_state:\n{}'.format(
            yaml.dump(network_state, default_flow_style=False, indent=4)))

        ifconfig_a = self._read_collect_file("ifconfig_a")
        logger.debug('ifconfig -a:\n{}'.format(ifconfig_a))

        ifconfig_dict = ifconfig_to_dict(ifconfig_a)
        logger.debug('parsed ifcfg dict:\n{}'.format(
            yaml.dump(ifconfig_dict, default_flow_style=False, indent=4)))

        ip_route_show = self._read_collect_file("ip_route_show")
        logger.debug("ip route show:\n{}".format(ip_route_show))
        for line in [line for line in ip_route_show.split('\n')
                     if 'src' in line]:
            m = re.search(r'^(?P<network>\S+)\sdev\s' +
                          r'(?P<devname>\S+)\s+' +
                          r'proto kernel\s+scope link' +
                          r'\s+src\s(?P<src_ip>\S+)',
                          line)
            route_info = m.groupdict('')
            logger.debug(route_info)

        route_n = self._read_collect_file("route_n")
        logger.debug("route -n:\n{}".format(route_n))

        interfaces = network_state.get('interfaces')
        for iface in interfaces.values():
            subnets = iface.get('subnets', {})
            if subnets:
                for index, subnet in zip(range(0, len(subnets)), subnets):
                    iface['index'] = index
                    if index == 0:
                        ifname = "{name}".format(**iface)
                    else:
                        ifname = "{name}:{index}".format(**iface)

                    self.check_interface(iface,
                                         ifconfig_dict.get(ifname),
                                         route_n)
            else:
                iface['index'] = 0
                self.check_interface(iface,
                                     ifconfig_dict.get(iface['name']),
                                     route_n)

    def check_interface(self, iface, ifconfig, route_n):
        logger.debug(
            'testing iface:\n{}\n\nifconfig:\n{}'.format(iface, ifconfig))
        subnets = iface.get('subnets', {})
        if subnets and iface['index'] != 0:
            ifname = "{name}:{index}".format(**iface)
        else:
            ifname = "{name}".format(**iface)

        # initial check, do we have the correct iface ?
        logger.debug('ifname={}'.format(ifname))
        logger.debug("ifconfig['interface']={}".format(ifconfig['interface']))
        self.assertEqual(ifname, ifconfig['interface'])

        # check physical interface attributes
        for key in ['mac_address', 'mtu']:
            if key in iface and iface[key]:
                self.assertEqual(iface[key],
                                 ifconfig[key])

        def __get_subnet(subnets, subidx):
            for index, subnet in zip(range(0, len(subnets)), subnets):
                if index == subidx:
                    break
            return subnet

        # check subnet related attributes, and specifically only
        # the subnet specified by iface['index']
        subnets = iface.get('subnets', {})
        if subnets:
            subnet = __get_subnet(subnets, iface['index'])
            if 'address' in subnet and subnet['address']:
                if ':' in subnet['address']:
                    inet_iface = ipaddress.IPv6Interface(
                        subnet['address'])
                else:
                    inet_iface = ipaddress.IPv4Interface(
                        subnet['address'])

                # check ip addr
                self.assertEqual(str(inet_iface.ip),
                                 ifconfig['address'])

                self.assertEqual(str(inet_iface.netmask),
                                 ifconfig['netmask'])

                self.assertEqual(
                    str(inet_iface.network.broadcast_address),
                    ifconfig['broadcast'])

            # handle gateway by looking at routing table
            if 'gateway' in subnet and subnet['gateway']:
                gw_ip = subnet['gateway']
                gateways = [line for line in route_n.split('\n')
                            if 'UG' in line and gw_ip in line]
                logger.debug('matching gateways:\n{}'.format(gateways))
                self.assertEqual(len(gateways), 1)
                [gateways] = gateways
                (dest, gw, genmask, flags, metric, ref, use, iface) = \
                    gateways.split()
                logger.debug('expected gw:{} found gw:{}'.format(gw_ip, gw))
                self.assertEqual(gw_ip, gw)


class TestNetworkStaticAbs(TestNetworkAbs):
    conf_file = "configs/basic_network_static.yaml"


class PreciseHWETTestNetwork(relbase.precise_hwe_t, TestNetworkAbs):
    # FIXME: off due to hang at test: Starting execute cloud user/final scripts
    __test__ = False
    pyver = 'py2'
    cloud_init_deb = 'cloud-init-%s.deb' % pyver


# Precise and Trusty tests are omitted due to not being able
# to run older cloud-init

class VividTestNetwork(relbase.vivid, TestNetworkAbs):
    __test__ = True


class VividTestNetworkStatic(relbase.vivid, TestNetworkStaticAbs):
    __test__ = True


class VividTestNetworkStaticConfigDrive(relbase.vivid, TestNetworkStaticAbs):
    __test__ = True
    localds = "cloud-config-drive"
    network_data = "configs/basic_network_static.json"

class WilyTestNetwork(relbase.wily, TestNetworkAbs):
    __test__ = True


class WilyTestNetworkStatic(relbase.wily, TestNetworkStaticAbs):
    __test__ = True


class WilyTestNetworkStaticConfigDrive(relbase.wily, TestNetworkStaticAbs):
    __test__ = True
    localds = "cloud-config-drive"
    network_data = "configs/basic_network_static.json"


class XenialTestNetwork(relbase.xenial, TestNetworkAbs):
    __test__ = True


class XenialTestNetworkStatic(relbase.xenial, TestNetworkStaticAbs):
    __test__ = True


class XenialTestNetworkStaticConfigDrive(relbase.xenial, TestNetworkStaticAbs):
    __test__ = True
    localds = "cloud-config-drive"
    network_data = "configs/basic_network_static.json"