salt/tests/unit/modules/virt_test.py

# -*- coding: utf-8 -*-

# Import python libs
import sys
import re

# Import Salt Testing libs
from salttesting import skipIf, TestCase
from salttesting.helpers import ensure_in_syspath
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, MagicMock, patch
ensure_in_syspath('../../')

# Import salt libs
from salt.modules import virt
from salt.modules import config
from salt._compat import ElementTree as ET
import salt.utils

# Import third party libs
import yaml

config.__grains__ = {}
config.__opts__ = {}
config.__pillar__ = {}
virt.__salt__ = {
    'config.get': config.get,
    'config.option': config.option,
}


@skipIf(NO_MOCK, NO_MOCK_REASON)
class VirtTestCase(TestCase):

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_boot_default_dev(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm'
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('os/boot').attrib['dev'], 'hd')

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_boot_custom_dev(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            boot_dev='cdrom'
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('os/boot').attrib['dev'], 'cdrom')

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_boot_multiple_devs(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            boot_dev='cdrom network'
            )
        root = ET.fromstring(xml_data)
        devs = root.findall('.//boot')
        self.assertTrue(len(devs) == 2)

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_serial_console(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            serial_type='pty',
            console=True
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('devices/serial').attrib['type'], 'pty')
        self.assertEqual(root.find('devices/console').attrib['type'], 'pty')

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_telnet_console(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            serial_type='tcp',
            console=True,
            telnet_port=22223
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('devices/serial').attrib['type'], 'tcp')
        self.assertEqual(root.find('devices/console').attrib['type'], 'tcp')
        self.assertEqual(root.find('devices/console/source').attrib['service'], '22223')

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_telnet_console_unspecified_port(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            serial_type='tcp',
            console=True
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('devices/serial').attrib['type'], 'tcp')
        self.assertEqual(root.find('devices/console').attrib['type'], 'tcp')
        self.assertIsInstance(int(root.find('devices/console/source').attrib['service']), int)

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_serial_no_console(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            serial_type='pty',
            console=False
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('devices/serial').attrib['type'], 'pty')
        self.assertEqual(root.find('devices/console'), None)

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_telnet_no_console(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            serial_type='tcp',
            console=False,
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('devices/serial').attrib['type'], 'tcp')
        self.assertEqual(root.find('devices/console'), None)

    def test_default_disk_profile_hypervisor_esxi(self):
        mock = MagicMock(return_value={})
        with patch.dict(virt.__salt__, {'config.get': mock}):
            ret = virt._disk_profile('nonexistant', 'esxi')
            self.assertTrue(len(ret) == 1)
            self.assertIn('system', ret[0])
            system = ret[0]['system']
            self.assertEqual(system['format'], 'vmdk')
            self.assertEqual(system['model'], 'scsi')
            self.assertTrue(system['size'] >= 1)

    def test_default_disk_profile_hypervisor_kvm(self):
        mock = MagicMock(return_value={})
        with patch.dict(virt.__salt__, {'config.get': mock}):
            ret = virt._disk_profile('nonexistant', 'kvm')
            self.assertTrue(len(ret) == 1)
            self.assertIn('system', ret[0])
            system = ret[0]['system']
            self.assertEqual(system['format'], 'qcow2')
            self.assertEqual(system['model'], 'virtio')
            self.assertTrue(system['size'] >= 1)

    def test_default_nic_profile_hypervisor_esxi(self):
        mock = MagicMock(return_value={})
        with patch.dict(virt.__salt__, {'config.get': mock}):
            ret = virt._nic_profile('nonexistant', 'esxi')
            self.assertTrue(len(ret) == 1)
            eth0 = ret[0]
            self.assertEqual(eth0['name'], 'eth0')
            self.assertEqual(eth0['type'], 'bridge')
            self.assertEqual(eth0['source'], 'DEFAULT')
            self.assertEqual(eth0['model'], 'e1000')

    def test_default_nic_profile_hypervisor_kvm(self):
        mock = MagicMock(return_value={})
        with patch.dict(virt.__salt__, {'config.get': mock}):
            ret = virt._nic_profile('nonexistant', 'kvm')
            self.assertTrue(len(ret) == 1)
            eth0 = ret[0]
            self.assertEqual(eth0['name'], 'eth0')
            self.assertEqual(eth0['type'], 'bridge')
            self.assertEqual(eth0['source'], 'br0')
            self.assertEqual(eth0['model'], 'virtio')

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_vol_xml_for_kvm(self):
        xml_data = virt._gen_vol_xml('vmname', 'system', 8192, 'kvm')
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('name').text, 'vmname/system.qcow2')
        self.assertEqual(root.find('key').text, 'vmname/system')
        self.assertEqual(root.find('capacity').attrib['unit'], 'KiB')
        self.assertEqual(root.find('capacity').text, str(8192 * 1024))

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_vol_xml_for_esxi(self):
        xml_data = virt._gen_vol_xml('vmname', 'system', 8192, 'esxi')
        root = ET.fromstring(xml_data)
        self.assertEqual(root.find('name').text, 'vmname/system.vmdk')
        self.assertEqual(root.find('key').text, 'vmname/system')
        self.assertEqual(root.find('capacity').attrib['unit'], 'KiB')
        self.assertEqual(root.find('capacity').text, str(8192 * 1024))

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_kvm_default_profile(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.attrib['type'], 'kvm')
        self.assertEqual(root.find('vcpu').text, '1')
        self.assertEqual(root.find('memory').text, str(512 * 1024))
        self.assertEqual(root.find('memory').attrib['unit'], 'KiB')

        disks = root.findall('.//disk')
        self.assertEqual(len(disks), 1)
        disk = disks[0]
        self.assertTrue(disk.find('source').attrib['file'].startswith('/'))
        self.assertTrue('hello/system' in disk.find('source').attrib['file'])
        self.assertEqual(disk.find('target').attrib['dev'], 'vda')
        self.assertEqual(disk.find('target').attrib['bus'], 'virtio')
        self.assertEqual(disk.find('driver').attrib['name'], 'qemu')
        self.assertEqual(disk.find('driver').attrib['type'], 'qcow2')

        interfaces = root.findall('.//interface')
        self.assertEqual(len(interfaces), 1)
        iface = interfaces[0]
        self.assertEqual(iface.attrib['type'], 'bridge')
        self.assertEqual(iface.find('source').attrib['bridge'], 'br0')
        self.assertEqual(iface.find('model').attrib['type'], 'virtio')

        mac = iface.find('mac').attrib['address']
        self.assertTrue(
              re.match('^([0-9A-F]{2}[:-]){5}([0-9A-F]{2})$', mac, re.I))

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_gen_xml_for_esxi_default_profile(self):
        diskp = virt._disk_profile('default', 'esxi')
        nicp = virt._nic_profile('default', 'esxi')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'esxi',
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.attrib['type'], 'vmware')
        self.assertEqual(root.find('vcpu').text, '1')
        self.assertEqual(root.find('memory').text, str(512 * 1024))
        self.assertEqual(root.find('memory').attrib['unit'], 'KiB')

        disks = root.findall('.//disk')
        self.assertEqual(len(disks), 1)
        disk = disks[0]
        self.assertTrue('[0]' in disk.find('source').attrib['file'])
        self.assertTrue('hello/system' in disk.find('source').attrib['file'])
        self.assertEqual(disk.find('target').attrib['dev'], 'sda')
        self.assertEqual(disk.find('target').attrib['bus'], 'scsi')
        self.assertEqual(disk.find('address').attrib['unit'], '0')

        interfaces = root.findall('.//interface')
        self.assertEqual(len(interfaces), 1)
        iface = interfaces[0]
        self.assertEqual(iface.attrib['type'], 'bridge')
        self.assertEqual(iface.find('source').attrib['bridge'], 'DEFAULT')
        self.assertEqual(iface.find('model').attrib['type'], 'e1000')

        mac = iface.find('mac').attrib['address']
        self.assertTrue(
              re.match('^([0-9A-F]{2}[:-]){5}([0-9A-F]{2})$', mac, re.I))

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    @patch('salt.modules.virt._nic_profile')
    @patch('salt.modules.virt._disk_profile')
    def test_gen_xml_for_esxi_custom_profile(self, disk_profile, nic_profile):
        diskp_yaml = '''
- first:
    size: 8192
    format: vmdk
    model: scsi
    pool: datastore1
- second:
    size: 4096
    format: vmdk  # FIX remove line, currently test fails
    model: scsi   # FIX remove line, currently test fails
    pool: datastore2
'''
        nicp_yaml = '''
- type: bridge
  name: eth1
  source: ONENET
  model: e1000
  mac: '00:00:00:00:00:00'
- name: eth2
  type: bridge
  source: TWONET
  model: e1000
  mac: '00:00:00:00:00:00'
'''
        disk_profile.return_value = yaml.load(diskp_yaml)
        nic_profile.return_value = yaml.load(nicp_yaml)
        diskp = virt._disk_profile('noeffect', 'esxi')
        nicp = virt._nic_profile('noeffect', 'esxi')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'esxi',
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.attrib['type'], 'vmware')
        self.assertEqual(root.find('vcpu').text, '1')
        self.assertEqual(root.find('memory').text, str(512 * 1024))
        self.assertEqual(root.find('memory').attrib['unit'], 'KiB')
        self.assertTrue(len(root.findall('.//disk')) == 2)
        self.assertTrue(len(root.findall('.//interface')) == 2)

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    @patch('salt.modules.virt._nic_profile')
    @patch('salt.modules.virt._disk_profile')
    def test_gen_xml_for_kvm_custom_profile(self, disk_profile, nic_profile):
        diskp_yaml = '''
- first:
    size: 8192
    format: qcow2
    model: virtio
    pool: /var/lib/images
- second:
    size: 4096
    format: qcow2   # FIX remove line, currently test fails
    model: virtio   # FIX remove line, currently test fails
    pool: /var/lib/images
'''
        nicp_yaml = '''
- type: bridge
  name: eth1
  source: b2
  model: virtio
  mac: '00:00:00:00:00:00'
- name: eth2
  type: bridge
  source: b2
  model: virtio
  mac: '00:00:00:00:00:00'
'''
        disk_profile.return_value = yaml.load(diskp_yaml)
        nic_profile.return_value = yaml.load(nicp_yaml)
        diskp = virt._disk_profile('noeffect', 'kvm')
        nicp = virt._nic_profile('noeffect', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm',
            )
        root = ET.fromstring(xml_data)
        self.assertEqual(root.attrib['type'], 'kvm')
        self.assertEqual(root.find('vcpu').text, '1')
        self.assertEqual(root.find('memory').text, str(512 * 1024))
        self.assertEqual(root.find('memory').attrib['unit'], 'KiB')
        self.assertTrue(len(root.findall('.//disk')) == 2)
        self.assertTrue(len(root.findall('.//interface')) == 2)

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_controller_for_esxi(self):
        diskp = virt._disk_profile('default', 'esxi')
        nicp = virt._nic_profile('default', 'esxi')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'esxi'
            )
        root = ET.fromstring(xml_data)
        controllers = root.findall('.//devices/controller')
        self.assertTrue(len(controllers) == 1)
        controller = controllers[0]
        self.assertEqual(controller.attrib['model'], 'lsilogic')

    @skipIf(sys.version_info < (2, 7), 'ElementTree version 1.3 required'
            ' which comes with Python 2.7')
    def test_controller_for_kvm(self):
        diskp = virt._disk_profile('default', 'kvm')
        nicp = virt._nic_profile('default', 'kvm')
        xml_data = virt._gen_xml(
            'hello',
            1,
            512,
            diskp,
            nicp,
            'kvm'
            )
        root = ET.fromstring(xml_data)
        controllers = root.findall('.//devices/controller')
        # There should be no controller
        self.assertTrue(len(controllers) == 0)

    def test_mixed_dict_and_list_as_profile_objects(self):

        yaml_config = '''
          virt.nic:
             new-listonly-profile:
                - bridge: br0
                  name: eth0
                - model: virtio
                  name: eth1
                  source: test_network
                  type: network
             new-list-with-legacy-names:
                - eth0:
                     bridge: br0
                - eth1:
                     bridge: br1
                     model: virtio
             non-default-legacy-profile:
                eth0:
                   bridge: br0
                eth1:
                   bridge: br1
                   model: virtio
        '''
        mock_config = yaml.load(yaml_config)
        salt.modules.config.__opts__ = mock_config

        for name in mock_config['virt.nic'].keys():
            profile = salt.modules.virt._nic_profile(name, 'kvm')
            self.assertEqual(len(profile), 2)

            interface_attrs = profile[0]
            self.assertIn('source', interface_attrs)
            self.assertIn('type', interface_attrs)
            self.assertIn('name', interface_attrs)
            self.assertIn('model', interface_attrs)
            self.assertEqual(interface_attrs['model'], 'virtio')
            self.assertIn('mac', interface_attrs)
            self.assertTrue(
                re.match('^([0-9A-F]{2}[:-]){5}([0-9A-F]{2})$',
                interface_attrs['mac'] , re.I))


if __name__ == '__main__':
    from integration import run_tests
    run_tests(VirtTestCase, needs_daemon=False)