path: root/tests/unittests/test_datasource/test_opennebula.py

# This file is part of cloud-init. See LICENSE file for license information.

from cloudinit import helpers
from cloudinit.sources import DataSourceOpenNebula as ds
from cloudinit import util
from cloudinit.tests.helpers import mock, populate_dir, CiTestCase

import os
import pwd
import unittest

import pytest


TEST_VARS = {
    'VAR1': 'single',
    'VAR2': 'double word',
    'VAR3': 'multi\nline\n',
    'VAR4': "'single'",
    'VAR5': "'double word'",
    'VAR6': "'multi\nline\n'",
    'VAR7': 'single\\t',
    'VAR8': 'double\\tword',
    'VAR9': 'multi\\t\nline\n',
    'VAR10': '\\',  # expect '\'
    'VAR11': '\'',  # expect '
    'VAR12': '$',   # expect $
}

INVALID_CONTEXT = ';'
USER_DATA = '#cloud-config\napt_upgrade: true'
SSH_KEY = 'ssh-rsa AAAAB3NzaC1....sIkJhq8wdX+4I3A4cYbYP ubuntu@server-460-%i'
HOSTNAME = 'foo.example.com'
PUBLIC_IP = '10.0.0.3'
MACADDR = '02:00:0a:12:01:01'
IP_BY_MACADDR = '10.18.1.1'
IP4_PREFIX = '24'
IP6_GLOBAL = '2001:db8:1:0:400:c0ff:fea8:1ba'
IP6_ULA = 'fd01:dead:beaf:0:400:c0ff:fea8:1ba'
IP6_GW = '2001:db8:1::ffff'
IP6_PREFIX = '48'

DS_PATH = "cloudinit.sources.DataSourceOpenNebula"


class TestOpenNebulaDataSource(CiTestCase):
    parsed_user = None
    allowed_subp = ['bash']

    def setUp(self):
        super(TestOpenNebulaDataSource, self).setUp()
        self.tmp = self.tmp_dir()
        self.paths = helpers.Paths(
            {'cloud_dir': self.tmp, 'run_dir': self.tmp})

        # defaults for few tests
        self.ds = ds.DataSourceOpenNebula
        self.seed_dir = os.path.join(self.paths.seed_dir, "opennebula")
        self.sys_cfg = {'datasource': {'OpenNebula': {'dsmode': 'local'}}}

        # we don't want 'sudo' called in tests. so we patch switch_user_cmd
        def my_switch_user_cmd(user):
            self.parsed_user = user
            return []

        self.switch_user_cmd_real = ds.switch_user_cmd
        ds.switch_user_cmd = my_switch_user_cmd

    def tearDown(self):
        ds.switch_user_cmd = self.switch_user_cmd_real
        super(TestOpenNebulaDataSource, self).tearDown()

    def test_get_data_non_contextdisk(self):
        orig_find_devs_with = util.find_devs_with
        try:
            # dont' try to lookup for CDs
            util.find_devs_with = lambda n: []
            dsrc = self.ds(sys_cfg=self.sys_cfg, distro=None, paths=self.paths)
            ret = dsrc.get_data()
            self.assertFalse(ret)
        finally:
            util.find_devs_with = orig_find_devs_with

    def test_get_data_broken_contextdisk(self):
        orig_find_devs_with = util.find_devs_with
        try:
            # dont' try to lookup for CDs
            util.find_devs_with = lambda n: []
            populate_dir(self.seed_dir, {'context.sh': INVALID_CONTEXT})
            dsrc = self.ds(sys_cfg=self.sys_cfg, distro=None, paths=self.paths)
            self.assertRaises(ds.BrokenContextDiskDir, dsrc.get_data)
        finally:
            util.find_devs_with = orig_find_devs_with

    def test_get_data_invalid_identity(self):
        orig_find_devs_with = util.find_devs_with
        try:
            # generate non-existing system user name
            sys_cfg = self.sys_cfg
            invalid_user = 'invalid'
            while not sys_cfg['datasource']['OpenNebula'].get('parseuser'):
                try:
                    pwd.getpwnam(invalid_user)
                    invalid_user += 'X'
                except KeyError:
                    sys_cfg['datasource']['OpenNebula']['parseuser'] = \
                        invalid_user

            # dont' try to lookup for CDs
            util.find_devs_with = lambda n: []
            populate_context_dir(self.seed_dir, {'KEY1': 'val1'})
            dsrc = self.ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
            self.assertRaises(ds.BrokenContextDiskDir, dsrc.get_data)
        finally:
            util.find_devs_with = orig_find_devs_with

    def test_get_data(self):
        orig_find_devs_with = util.find_devs_with
        try:
            # dont' try to lookup for CDs
            util.find_devs_with = lambda n: []
            populate_context_dir(self.seed_dir, {'KEY1': 'val1'})
            dsrc = self.ds(sys_cfg=self.sys_cfg, distro=None, paths=self.paths)
            ret = dsrc.get_data()
            self.assertTrue(ret)
        finally:
            util.find_devs_with = orig_find_devs_with
        self.assertEqual('opennebula', dsrc.cloud_name)
        self.assertEqual('opennebula', dsrc.platform_type)
        self.assertEqual(
            'seed-dir (%s/seed/opennebula)' % self.tmp, dsrc.subplatform)

    def test_seed_dir_non_contextdisk(self):
        self.assertRaises(ds.NonContextDiskDir, ds.read_context_disk_dir,
                          self.seed_dir, mock.Mock())

    def test_seed_dir_empty1_context(self):
        populate_dir(self.seed_dir, {'context.sh': ''})
        results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

        self.assertIsNone(results['userdata'])
        self.assertEqual(results['metadata'], {})

    def test_seed_dir_empty2_context(self):
        populate_context_dir(self.seed_dir, {})
        results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

        self.assertIsNone(results['userdata'])
        self.assertEqual(results['metadata'], {})

    def test_seed_dir_broken_context(self):
        populate_dir(self.seed_dir, {'context.sh': INVALID_CONTEXT})

        self.assertRaises(ds.BrokenContextDiskDir,
                          ds.read_context_disk_dir,
                          self.seed_dir, mock.Mock())

    def test_context_parser(self):
        populate_context_dir(self.seed_dir, TEST_VARS)
        results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

        self.assertTrue('metadata' in results)
        self.assertEqual(TEST_VARS, results['metadata'])

    def test_ssh_key(self):
        public_keys = ['first key', 'second key']
        for c in range(4):
            for k in ('SSH_KEY', 'SSH_PUBLIC_KEY'):
                my_d = os.path.join(self.tmp, "%s-%i" % (k, c))
                populate_context_dir(my_d, {k: '\n'.join(public_keys)})
                results = ds.read_context_disk_dir(my_d, mock.Mock())

                self.assertTrue('metadata' in results)
                self.assertTrue('public-keys' in results['metadata'])
                self.assertEqual(public_keys,
                                 results['metadata']['public-keys'])

            public_keys.append(SSH_KEY % (c + 1,))

    def test_user_data_plain(self):
        for k in ('USER_DATA', 'USERDATA'):
            my_d = os.path.join(self.tmp, k)
            populate_context_dir(my_d, {k: USER_DATA,
                                        'USERDATA_ENCODING': ''})
            results = ds.read_context_disk_dir(my_d, mock.Mock())

            self.assertTrue('userdata' in results)
            self.assertEqual(USER_DATA, results['userdata'])

    def test_user_data_encoding_required_for_decode(self):
        b64userdata = util.b64e(USER_DATA)
        for k in ('USER_DATA', 'USERDATA'):
            my_d = os.path.join(self.tmp, k)
            populate_context_dir(my_d, {k: b64userdata})
            results = ds.read_context_disk_dir(my_d, mock.Mock())

            self.assertTrue('userdata' in results)
            self.assertEqual(b64userdata, results['userdata'])

    def test_user_data_base64_encoding(self):
        for k in ('USER_DATA', 'USERDATA'):
            my_d = os.path.join(self.tmp, k)
            populate_context_dir(my_d, {k: util.b64e(USER_DATA),
                                        'USERDATA_ENCODING': 'base64'})
            results = ds.read_context_disk_dir(my_d, mock.Mock())

            self.assertTrue('userdata' in results)
            self.assertEqual(USER_DATA, results['userdata'])

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_hostname(self, m_get_phys_by_mac):
        for dev in ('eth0', 'ens3'):
            m_get_phys_by_mac.return_value = {MACADDR: dev}
            for k in ('HOSTNAME', 'PUBLIC_IP', 'IP_PUBLIC', 'ETH0_IP'):
                my_d = os.path.join(self.tmp, k)
                populate_context_dir(my_d, {k: PUBLIC_IP})
                results = ds.read_context_disk_dir(my_d, mock.Mock())

                self.assertTrue('metadata' in results)
                self.assertTrue('local-hostname' in results['metadata'])
                self.assertEqual(
                    PUBLIC_IP, results['metadata']['local-hostname'])

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_network_interfaces(self, m_get_phys_by_mac):
        for dev in ('eth0', 'ens3'):
            m_get_phys_by_mac.return_value = {MACADDR: dev}

            # without ETH0_MAC
            # for Older OpenNebula?
            populate_context_dir(self.seed_dir, {'ETH0_IP': IP_BY_MACADDR})
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP_BY_MACADDR + '/' + IP4_PREFIX in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_IP and ETH0_MAC
            populate_context_dir(
                self.seed_dir, {'ETH0_IP': IP_BY_MACADDR, 'ETH0_MAC': MACADDR})
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP_BY_MACADDR + '/' + IP4_PREFIX in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_IP with empty string and ETH0_MAC
            # in the case of using Virtual Network contains
            # "AR = [ TYPE = ETHER ]"
            populate_context_dir(
                self.seed_dir, {'ETH0_IP': '', 'ETH0_MAC': MACADDR})
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP_BY_MACADDR + '/' + IP4_PREFIX in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_MASK
            populate_context_dir(
                self.seed_dir, {
                    'ETH0_IP': IP_BY_MACADDR,
                    'ETH0_MAC': MACADDR,
                    'ETH0_MASK': '255.255.0.0'
                })
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP_BY_MACADDR + '/16' in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_MASK with empty string
            populate_context_dir(
                self.seed_dir, {
                    'ETH0_IP': IP_BY_MACADDR,
                    'ETH0_MAC': MACADDR,
                    'ETH0_MASK': ''
                })
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP_BY_MACADDR + '/' + IP4_PREFIX in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_IP6
            populate_context_dir(
                self.seed_dir, {
                    'ETH0_IP6': IP6_GLOBAL,
                    'ETH0_MAC': MACADDR,
                })
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP6_GLOBAL + '/64' in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_IP6_ULA
            populate_context_dir(
                self.seed_dir, {
                    'ETH0_IP6_ULA': IP6_ULA,
                    'ETH0_MAC': MACADDR,
                })
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP6_ULA + '/64' in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_IP6 and ETH0_IP6_PREFIX_LENGTH
            populate_context_dir(
                self.seed_dir, {
                    'ETH0_IP6': IP6_GLOBAL,
                    'ETH0_IP6_PREFIX_LENGTH': IP6_PREFIX,
                    'ETH0_MAC': MACADDR,
                })
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP6_GLOBAL + '/' + IP6_PREFIX in
                results['network-interfaces']['ethernets'][dev]['addresses'])

            # ETH0_IP6 and ETH0_IP6_PREFIX_LENGTH with empty string
            populate_context_dir(
                self.seed_dir, {
                    'ETH0_IP6': IP6_GLOBAL,
                    'ETH0_IP6_PREFIX_LENGTH': '',
                    'ETH0_MAC': MACADDR,
                })
            results = ds.read_context_disk_dir(self.seed_dir, mock.Mock())

            self.assertTrue('network-interfaces' in results)
            self.assertTrue(
                IP6_GLOBAL + '/64' in
                results['network-interfaces']['ethernets'][dev]['addresses'])

    def test_find_candidates(self):
        def my_devs_with(criteria):
            return {
                "LABEL=CONTEXT": ["/dev/sdb"],
                "LABEL=CDROM": ["/dev/sr0"],
                "TYPE=iso9660": ["/dev/vdb"],
            }.get(criteria, [])

        orig_find_devs_with = util.find_devs_with
        try:
            util.find_devs_with = my_devs_with
            self.assertEqual(["/dev/sdb", "/dev/sr0", "/dev/vdb"],
                             ds.find_candidate_devs())
        finally:
            util.find_devs_with = orig_find_devs_with


@mock.patch(DS_PATH + '.net.get_interfaces_by_mac', mock.Mock(return_value={}))
class TestOpenNebulaNetwork(unittest.TestCase):

    system_nics = ('eth0', 'ens3')

    def test_context_devname(self):
        """Verify context_devname correctly returns mac and name."""
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH1_MAC': '02:00:0a:12:0f:0f', }
        expected = {
            '02:00:0a:12:01:01': 'ETH0',
            '02:00:0a:12:0f:0f': 'ETH1', }
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        self.assertEqual(expected, net.context_devname)

    def test_get_nameservers(self):
        """
        Verify get_nameservers('device') correctly returns DNS server addresses
        and search domains.
        """
        context = {
            'DNS': '1.2.3.8',
            'ETH0_DNS': '1.2.3.6 1.2.3.7',
            'ETH0_SEARCH_DOMAIN': 'example.com example.org', }
        expected = {
            'addresses': ['1.2.3.6', '1.2.3.7', '1.2.3.8'],
            'search': ['example.com', 'example.org']}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_nameservers('eth0')
        self.assertEqual(expected, val)

    def test_get_mtu(self):
        """Verify get_mtu('device') correctly returns MTU size."""
        context = {'ETH0_MTU': '1280'}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_mtu('eth0')
        self.assertEqual('1280', val)

    def test_get_ip(self):
        """Verify get_ip('device') correctly returns IPv4 address."""
        context = {'ETH0_IP': PUBLIC_IP}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip('eth0', MACADDR)
        self.assertEqual(PUBLIC_IP, val)

    def test_get_ip_emptystring(self):
        """
        Verify get_ip('device') correctly returns IPv4 address.
        It returns IP address created by MAC address if ETH0_IP has empty
        string.
        """
        context = {'ETH0_IP': ''}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip('eth0', MACADDR)
        self.assertEqual(IP_BY_MACADDR, val)

    def test_get_ip6(self):
        """
        Verify get_ip6('device') correctly returns IPv6 address.
        In this case, IPv6 address is Given by ETH0_IP6.
        """
        context = {
            'ETH0_IP6': IP6_GLOBAL,
            'ETH0_IP6_ULA': '', }
        expected = [IP6_GLOBAL]
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip6('eth0')
        self.assertEqual(expected, val)

    def test_get_ip6_ula(self):
        """
        Verify get_ip6('device') correctly returns IPv6 address.
        In this case, IPv6 address is Given by ETH0_IP6_ULA.
        """
        context = {
            'ETH0_IP6': '',
            'ETH0_IP6_ULA': IP6_ULA, }
        expected = [IP6_ULA]
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip6('eth0')
        self.assertEqual(expected, val)

    def test_get_ip6_dual(self):
        """
        Verify get_ip6('device') correctly returns IPv6 address.
        In this case, IPv6 addresses are Given by ETH0_IP6 and ETH0_IP6_ULA.
        """
        context = {
            'ETH0_IP6': IP6_GLOBAL,
            'ETH0_IP6_ULA': IP6_ULA, }
        expected = [IP6_GLOBAL, IP6_ULA]
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip6('eth0')
        self.assertEqual(expected, val)

    def test_get_ip6_prefix(self):
        """
        Verify get_ip6_prefix('device') correctly returns IPv6 prefix.
        """
        context = {'ETH0_IP6_PREFIX_LENGTH': IP6_PREFIX}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip6_prefix('eth0')
        self.assertEqual(IP6_PREFIX, val)

    def test_get_ip6_prefix_emptystring(self):
        """
        Verify get_ip6_prefix('device') correctly returns IPv6 prefix.
        It returns default value '64' if ETH0_IP6_PREFIX_LENGTH has empty
        string.
        """
        context = {'ETH0_IP6_PREFIX_LENGTH': ''}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_ip6_prefix('eth0')
        self.assertEqual('64', val)

    def test_get_gateway(self):
        """
        Verify get_gateway('device') correctly returns IPv4 default gateway
        address.
        """
        context = {'ETH0_GATEWAY': '1.2.3.5'}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_gateway('eth0')
        self.assertEqual('1.2.3.5', val)

    def test_get_gateway6(self):
        """
        Verify get_gateway6('device') correctly returns IPv6 default gateway
        address.
        """
        context = {'ETH0_GATEWAY6': IP6_GW}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_gateway6('eth0')
        self.assertEqual(IP6_GW, val)

    def test_get_mask(self):
        """
        Verify get_mask('device') correctly returns IPv4 subnet mask.
        """
        context = {'ETH0_MASK': '255.255.0.0'}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_mask('eth0')
        self.assertEqual('255.255.0.0', val)

    def test_get_mask_emptystring(self):
        """
        Verify get_mask('device') correctly returns IPv4 subnet mask.
        It returns default value '255.255.255.0' if ETH0_MASK has empty string.
        """
        context = {'ETH0_MASK': ''}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_mask('eth0')
        self.assertEqual('255.255.255.0', val)

    def test_get_network(self):
        """
        Verify get_network('device') correctly returns IPv4 network address.
        """
        context = {'ETH0_NETWORK': '1.2.3.0'}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_network('eth0', MACADDR)
        self.assertEqual('1.2.3.0', val)

    def test_get_network_emptystring(self):
        """
        Verify get_network('device') correctly returns IPv4 network address.
        It returns network address created by MAC address if ETH0_NETWORK has
        empty string.
        """
        context = {'ETH0_NETWORK': ''}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_network('eth0', MACADDR)
        self.assertEqual('10.18.1.0', val)

    def test_get_field(self):
        """
        Verify get_field('device', 'name') returns *context* value.
        """
        context = {'ETH9_DUMMY': 'DUMMY_VALUE'}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_field('eth9', 'dummy')
        self.assertEqual('DUMMY_VALUE', val)

    def test_get_field_withdefaultvalue(self):
        """
        Verify get_field('device', 'name', 'default value') returns *context*
        value.
        """
        context = {'ETH9_DUMMY': 'DUMMY_VALUE'}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_field('eth9', 'dummy', 'DEFAULT_VALUE')
        self.assertEqual('DUMMY_VALUE', val)

    def test_get_field_withdefaultvalue_emptycontext(self):
        """
        Verify get_field('device', 'name', 'default value') returns *default*
        value if context value is empty string.
        """
        context = {'ETH9_DUMMY': ''}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_field('eth9', 'dummy', 'DEFAULT_VALUE')
        self.assertEqual('DEFAULT_VALUE', val)

    def test_get_field_emptycontext(self):
        """
        Verify get_field('device', 'name') returns None if context value is
        empty string.
        """
        context = {'ETH9_DUMMY': ''}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_field('eth9', 'dummy')
        self.assertEqual(None, val)

    def test_get_field_nonecontext(self):
        """
        Verify get_field('device', 'name') returns None if context value is
        None.
        """
        context = {'ETH9_DUMMY': None}
        net = ds.OpenNebulaNetwork(context, mock.Mock())
        val = net.get_field('eth9', 'dummy')
        self.assertEqual(None, val)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_gen_conf_gateway(self, m_get_phys_by_mac):
        """Test rendering with/without IPv4 gateway"""
        self.maxDiff = None
        # empty ETH0_GATEWAY
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_GATEWAY': '', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

        # set ETH0_GATEWAY
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_GATEWAY': '1.2.3.5', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'gateway4': '1.2.3.5',
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_gen_conf_gateway6(self, m_get_phys_by_mac):
        """Test rendering with/without IPv6 gateway"""
        self.maxDiff = None
        # empty ETH0_GATEWAY6
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_GATEWAY6': '', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

        # set ETH0_GATEWAY6
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_GATEWAY6': IP6_GW, }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'gateway6': IP6_GW,
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_gen_conf_ipv6address(self, m_get_phys_by_mac):
        """Test rendering with/without IPv6 address"""
        self.maxDiff = None
        # empty ETH0_IP6, ETH0_IP6_ULA, ETH0_IP6_PREFIX_LENGTH
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_IP6': '',
            'ETH0_IP6_ULA': '',
            'ETH0_IP6_PREFIX_LENGTH': '', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

        # set ETH0_IP6, ETH0_IP6_ULA, ETH0_IP6_PREFIX_LENGTH
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_IP6': IP6_GLOBAL,
            'ETH0_IP6_PREFIX_LENGTH': IP6_PREFIX,
            'ETH0_IP6_ULA': IP6_ULA, }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [
                            IP_BY_MACADDR + '/' + IP4_PREFIX,
                            IP6_GLOBAL + '/' + IP6_PREFIX,
                            IP6_ULA + '/' + IP6_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_gen_conf_dns(self, m_get_phys_by_mac):
        """Test rendering with/without DNS server, search domain"""
        self.maxDiff = None
        # empty DNS, ETH0_DNS, ETH0_SEARCH_DOMAIN
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'DNS': '',
            'ETH0_DNS': '',
            'ETH0_SEARCH_DOMAIN': '', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

        # set DNS, ETH0_DNS, ETH0_SEARCH_DOMAIN
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'DNS': '1.2.3.8',
            'ETH0_DNS': '1.2.3.6 1.2.3.7',
            'ETH0_SEARCH_DOMAIN': 'example.com example.org', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'nameservers': {
                            'addresses': ['1.2.3.6', '1.2.3.7', '1.2.3.8'],
                            'search': ['example.com', 'example.org']},
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_gen_conf_mtu(self, m_get_phys_by_mac):
        """Test rendering with/without MTU"""
        self.maxDiff = None
        # empty ETH0_MTU
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_MTU': '', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

        # set ETH0_MTU
        context = {
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_MTU': '1280', }
        for nic in self.system_nics:
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'mtu': '1280',
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork(context, mock.Mock())
            self.assertEqual(net.gen_conf(), expected)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_eth0(self, m_get_phys_by_mac):
        for nic in self.system_nics:
            m_get_phys_by_mac.return_value = {MACADDR: nic}
            net = ds.OpenNebulaNetwork({}, mock.Mock())
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/' + IP4_PREFIX]}}}

            self.assertEqual(net.gen_conf(), expected)

    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
    def test_distro_passed_through(self, m_get_physical_nics_by_mac):
        ds.OpenNebulaNetwork({}, mock.sentinel.distro)
        self.assertEqual(
            [mock.call(mock.sentinel.distro)],
            m_get_physical_nics_by_mac.call_args_list,
        )

    def test_eth0_override(self):
        self.maxDiff = None
        context = {
            'DNS': '1.2.3.8',
            'ETH0_DNS': '1.2.3.6 1.2.3.7',
            'ETH0_GATEWAY': '1.2.3.5',
            'ETH0_GATEWAY6': '',
            'ETH0_IP': IP_BY_MACADDR,
            'ETH0_IP6': '',
            'ETH0_IP6_PREFIX_LENGTH': '',
            'ETH0_IP6_ULA': '',
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_MASK': '255.255.0.0',
            'ETH0_MTU': '',
            'ETH0_NETWORK': '10.18.0.0',
            'ETH0_SEARCH_DOMAIN': '',
        }
        for nic in self.system_nics:
            net = ds.OpenNebulaNetwork(context, mock.Mock(),
                                       system_nics_by_mac={MACADDR: nic})
            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [IP_BY_MACADDR + '/16'],
                        'gateway4': '1.2.3.5',
                        'nameservers': {
                            'addresses': ['1.2.3.6', '1.2.3.7', '1.2.3.8']}}}}

            self.assertEqual(expected, net.gen_conf())

    def test_eth0_v4v6_override(self):
        self.maxDiff = None
        context = {
            'DNS': '1.2.3.8',
            'ETH0_DNS': '1.2.3.6 1.2.3.7',
            'ETH0_GATEWAY': '1.2.3.5',
            'ETH0_GATEWAY6': IP6_GW,
            'ETH0_IP': IP_BY_MACADDR,
            'ETH0_IP6': IP6_GLOBAL,
            'ETH0_IP6_PREFIX_LENGTH': IP6_PREFIX,
            'ETH0_IP6_ULA': IP6_ULA,
            'ETH0_MAC': '02:00:0a:12:01:01',
            'ETH0_MASK': '255.255.0.0',
            'ETH0_MTU': '1280',
            'ETH0_NETWORK': '10.18.0.0',
            'ETH0_SEARCH_DOMAIN': 'example.com example.org',
        }
        for nic in self.system_nics:
            net = ds.OpenNebulaNetwork(context, mock.Mock(),
                                       system_nics_by_mac={MACADDR: nic})

            expected = {
                'version': 2,
                'ethernets': {
                    nic: {
                        'match': {'macaddress': MACADDR},
                        'addresses': [
                            IP_BY_MACADDR + '/16',
                            IP6_GLOBAL + '/' + IP6_PREFIX,
                            IP6_ULA + '/' + IP6_PREFIX],
                        'gateway4': '1.2.3.5',
                        'gateway6': IP6_GW,
                        'nameservers': {
                            'addresses': ['1.2.3.6', '1.2.3.7', '1.2.3.8'],
                            'search': ['example.com', 'example.org']},
                        'mtu': '1280'}}}

            self.assertEqual(expected, net.gen_conf())

    def test_multiple_nics(self):
        """Test rendering multiple nics with names that differ from context."""
        self.maxDiff = None
        MAC_1 = "02:00:0a:12:01:01"
        MAC_2 = "02:00:0a:12:01:02"
        context = {
            'DNS': '1.2.3.8',
            'ETH0_DNS': '1.2.3.6 1.2.3.7',
            'ETH0_GATEWAY': '1.2.3.5',
            'ETH0_GATEWAY6': IP6_GW,
            'ETH0_IP': '10.18.1.1',
            'ETH0_IP6': IP6_GLOBAL,
            'ETH0_IP6_PREFIX_LENGTH': '',
            'ETH0_IP6_ULA': IP6_ULA,
            'ETH0_MAC': MAC_2,
            'ETH0_MASK': '255.255.0.0',
            'ETH0_MTU': '1280',
            'ETH0_NETWORK': '10.18.0.0',
            'ETH0_SEARCH_DOMAIN': 'example.com',
            'ETH3_DNS': '10.3.1.2',
            'ETH3_GATEWAY': '10.3.0.1',
            'ETH3_GATEWAY6': '',
            'ETH3_IP': '10.3.1.3',
            'ETH3_IP6': '',
            'ETH3_IP6_PREFIX_LENGTH': '',
            'ETH3_IP6_ULA': '',
            'ETH3_MAC': MAC_1,
            'ETH3_MASK': '255.255.0.0',
            'ETH3_MTU': '',
            'ETH3_NETWORK': '10.3.0.0',
            'ETH3_SEARCH_DOMAIN': 'third.example.com third.example.org',
        }
        net = ds.OpenNebulaNetwork(
            context,
            mock.Mock(),
            system_nics_by_mac={MAC_1: 'enp0s25', MAC_2: 'enp1s2'}
        )

        expected = {
            'version': 2,
            'ethernets': {
                'enp1s2': {
                    'match': {'macaddress': MAC_2},
                    'addresses': [
                        '10.18.1.1/16',
                        IP6_GLOBAL + '/64',
                        IP6_ULA + '/64'],
                    'gateway4': '1.2.3.5',
                    'gateway6': IP6_GW,
                    'nameservers': {
                        'addresses': ['1.2.3.6', '1.2.3.7', '1.2.3.8'],
                        'search': ['example.com']},
                    'mtu': '1280'},
                'enp0s25': {
                    'match': {'macaddress': MAC_1},
                    'addresses': ['10.3.1.3/16'],
                    'gateway4': '10.3.0.1',
                    'nameservers': {
                        'addresses': ['10.3.1.2', '1.2.3.8'],
                        'search': [
                            'third.example.com',
                            'third.example.org']}}}}

        self.assertEqual(expected, net.gen_conf())


class TestParseShellConfig:
    @pytest.mark.allow_subp_for("bash")
    def test_no_seconds(self):
        cfg = '\n'.join(["foo=bar", "SECONDS=2", "xx=foo"])
        # we could test 'sleep 2', but that would make the test run slower.
        ret = ds.parse_shell_config(cfg)
        assert ret == {"foo": "bar", "xx": "foo"}


class TestGetPhysicalNicsByMac:
    @pytest.mark.parametrize(
        "interfaces_by_mac,physical_devs,expected_return",
        [
            # No interfaces => empty return
            ({}, [], {}),
            # Only virtual interface => empty return
            ({"mac1": "virtual0"}, [], {}),
            # Only physical interface => it is returned
            ({"mac2": "physical0"}, ["physical0"], {"mac2": "physical0"}),
            # Combination of physical and virtual => only physical returned
            (
                {"mac3": "physical1", "mac4": "virtual1"},
                ["physical1"],
                {"mac3": "physical1"},
            ),
        ],
    )
    def test(self, interfaces_by_mac, physical_devs, expected_return):
        distro = mock.Mock()
        distro.networking.is_physical.side_effect = (
            lambda devname: devname in physical_devs
        )
        with mock.patch(
            DS_PATH + ".net.get_interfaces_by_mac",
            return_value=interfaces_by_mac,
        ):
            assert expected_return == ds.get_physical_nics_by_mac(distro)


def populate_context_dir(path, variables):
    data = "# Context variables generated by OpenNebula\n"
    for k, v in variables.items():
        data += ("%s='%s'\n" % (k.upper(), v.replace(r"'", r"'\''")))
    populate_dir(path, {'context.sh': data})

# vi: ts=4 expandtab