import copy
import os
-import unittest
-import mock
+import uuid
+import mock
+import unittest
from xml.etree import ElementTree
+
+from yardstick import ssh
from yardstick.benchmark.contexts.standalone import model
+from yardstick.common import exceptions
+from yardstick import constants
from yardstick.network_services import utils
</domain>
"""
+
class ModelLibvirtTestCase(unittest.TestCase):
def setUp(self):
- self.xml = ElementTree.ElementTree(
- element=ElementTree.fromstring(XML_SAMPLE))
self.pci_address_str = '0001:04:03.2'
self.pci_address = utils.PciAddress(self.pci_address_str)
self.mac = '00:00:00:00:00:01'
- self._mock_write_xml = mock.patch.object(ElementTree.ElementTree,
- 'write')
- self.mock_write_xml = self._mock_write_xml.start()
-
+ self._mock_ssh = mock.Mock()
+ self.mock_ssh = self._mock_ssh.start()
self.addCleanup(self._cleanup)
def _cleanup(self):
- self._mock_write_xml.stop()
+ self._mock_ssh.stop()
# TODO: Remove mocking of yardstick.ssh.SSH (here and elsewhere)
# In this case, we are mocking a param to be passed into other methods
model.Libvirt.check_if_vm_exists_and_delete("vm_0", ssh_mock)
def test_virsh_create_vm(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = mock.Mock(return_value=(0, "a", ""))
- ssh.return_value = ssh_mock
- # NOTE(ralonsoh): this test doesn't cover function execution.
- model.Libvirt.virsh_create_vm(ssh_mock, "vm_0")
+ self.mock_ssh.execute = mock.Mock(return_value=(0, 0, 0))
+ model.Libvirt.virsh_create_vm(self.mock_ssh, 'vm_0')
+ self.mock_ssh.execute.assert_called_once_with('virsh create vm_0')
+
+ def test_virsh_create_vm_error(self):
+ self.mock_ssh.execute = mock.Mock(return_value=(1, 0, 'error_create'))
+ with self.assertRaises(exceptions.LibvirtCreateError) as exc:
+ model.Libvirt.virsh_create_vm(self.mock_ssh, 'vm_0')
+ self.assertEqual('Error creating the virtual machine. Error: '
+ 'error_create.', str(exc.exception))
+ self.mock_ssh.execute.assert_called_once_with('virsh create vm_0')
def test_virsh_destroy_vm(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = mock.Mock(return_value=(0, "a", ""))
- ssh.return_value = ssh_mock
- # NOTE(ralonsoh): this test doesn't cover function execution.
- model.Libvirt.virsh_destroy_vm("vm_0", ssh_mock)
+ self.mock_ssh.execute = mock.Mock(return_value=(0, 0, 0))
+ model.Libvirt.virsh_destroy_vm('vm_0', self.mock_ssh)
+ self.mock_ssh.execute.assert_called_once_with('virsh destroy vm_0')
+
+ @mock.patch.object(model, 'LOG')
+ def test_virsh_destroy_vm_error(self, mock_logger):
+ self.mock_ssh.execute = mock.Mock(return_value=(1, 0, 'error_destroy'))
+ mock_logger.warning = mock.Mock()
+ model.Libvirt.virsh_destroy_vm('vm_0', self.mock_ssh)
+ mock_logger.warning.assert_called_once_with(
+ 'Error destroying VM %s. Error: %s', 'vm_0', 'error_destroy')
+ self.mock_ssh.execute.assert_called_once_with('virsh destroy vm_0')
def test_add_interface_address(self):
xml = ElementTree.ElementTree(
result.get('function'))
def test_add_ovs_interfaces(self):
- xml_input = mock.Mock()
- with mock.patch.object(ElementTree, 'parse', return_value=self.xml) \
- as mock_parse:
- xml = copy.deepcopy(self.xml)
- mock_parse.return_value = xml
- model.Libvirt.add_ovs_interface(
- '/usr/local', 0, self.pci_address_str, self.mac, xml_input)
- mock_parse.assert_called_once_with(xml_input)
- self.mock_write_xml.assert_called_once_with(xml_input)
- interface = xml.find('devices').find('interface')
- self.assertEqual('vhostuser', interface.get('type'))
- mac = interface.find('mac')
- self.assertEqual(self.mac, mac.get('address'))
- source = interface.find('source')
- self.assertEqual('unix', source.get('type'))
- self.assertEqual('/usr/local/var/run/openvswitch/dpdkvhostuser0',
- source.get('path'))
- self.assertEqual('client', source.get('mode'))
- _model = interface.find('model')
- self.assertEqual('virtio', _model.get('type'))
- driver = interface.find('driver')
- self.assertEqual('4', driver.get('queues'))
- host = driver.find('host')
- self.assertEqual('off', host.get('mrg_rxbuf'))
- self.assertIsNotNone(interface.find('address'))
+ xml_input = copy.deepcopy(XML_SAMPLE)
+ xml_output = model.Libvirt.add_ovs_interface(
+ '/usr/local', 0, self.pci_address_str, self.mac, xml_input)
+
+ root = ElementTree.fromstring(xml_output)
+ et_out = ElementTree.ElementTree(element=root)
+ interface = et_out.find('devices').find('interface')
+ self.assertEqual('vhostuser', interface.get('type'))
+ mac = interface.find('mac')
+ self.assertEqual(self.mac, mac.get('address'))
+ source = interface.find('source')
+ self.assertEqual('unix', source.get('type'))
+ self.assertEqual('/usr/local/var/run/openvswitch/dpdkvhostuser0',
+ source.get('path'))
+ self.assertEqual('client', source.get('mode'))
+ _model = interface.find('model')
+ self.assertEqual('virtio', _model.get('type'))
+ driver = interface.find('driver')
+ self.assertEqual('4', driver.get('queues'))
+ host = driver.find('host')
+ self.assertEqual('off', host.get('mrg_rxbuf'))
+ self.assertIsNotNone(interface.find('address'))
def test_add_sriov_interfaces(self):
- xml_input = mock.Mock()
- with mock.patch.object(ElementTree, 'parse', return_value=self.xml) \
- as mock_parse:
- xml = copy.deepcopy(self.xml)
- mock_parse.return_value = xml
- vm_pci = '0001:05:04.2'
- model.Libvirt.add_sriov_interfaces(
- vm_pci, self.pci_address_str, self.mac, xml_input)
- mock_parse.assert_called_once_with(xml_input)
- self.mock_write_xml.assert_called_once_with(xml_input)
- interface = xml.find('devices').find('interface')
- self.assertEqual('yes', interface.get('managed'))
- self.assertEqual('hostdev', interface.get('type'))
- mac = interface.find('mac')
- self.assertEqual(self.mac, mac.get('address'))
- source = interface.find('source')
- source_address = source.find('address')
- self.assertIsNotNone(source.find('address'))
-
- self.assertEqual('pci', source_address.get('type'))
- self.assertEqual('0x' + self.pci_address_str.split(':')[0],
- source_address.get('domain'))
- self.assertEqual('0x' + self.pci_address_str.split(':')[1],
- source_address.get('bus'))
- self.assertEqual('0x' + self.pci_address_str.split(':')[2].split('.')[0],
- source_address.get('slot'))
- self.assertEqual('0x' + self.pci_address_str.split(':')[2].split('.')[1],
- source_address.get('function'))
-
- interface_address = interface.find('address')
- self.assertEqual('pci', interface_address.get('type'))
- self.assertEqual('0x' + vm_pci.split(':')[0],
- interface_address.get('domain'))
- self.assertEqual('0x' + vm_pci.split(':')[1],
- interface_address.get('bus'))
- self.assertEqual('0x' + vm_pci.split(':')[2].split('.')[0],
- interface_address.get('slot'))
- self.assertEqual('0x' + vm_pci.split(':')[2].split('.')[1],
- interface_address.get('function'))
+ xml_input = copy.deepcopy(XML_SAMPLE)
+ vm_pci = '0001:05:04.2'
+ xml_output = model.Libvirt.add_sriov_interfaces(
+ vm_pci, self.pci_address_str, self.mac, xml_input)
+ root = ElementTree.fromstring(xml_output)
+ et_out = ElementTree.ElementTree(element=root)
+ interface = et_out.find('devices').find('interface')
+ self.assertEqual('yes', interface.get('managed'))
+ self.assertEqual('hostdev', interface.get('type'))
+ mac = interface.find('mac')
+ self.assertEqual(self.mac, mac.get('address'))
+ source = interface.find('source')
+ source_address = source.find('address')
+ self.assertIsNotNone(source.find('address'))
+
+ self.assertEqual('pci', source_address.get('type'))
+ self.assertEqual('0x' + self.pci_address_str.split(':')[0],
+ source_address.get('domain'))
+ self.assertEqual('0x' + self.pci_address_str.split(':')[1],
+ source_address.get('bus'))
+ self.assertEqual('0x' + self.pci_address_str.split(':')[2].split('.')[0],
+ source_address.get('slot'))
+ self.assertEqual('0x' + self.pci_address_str.split(':')[2].split('.')[1],
+ source_address.get('function'))
+
+ interface_address = interface.find('address')
+ self.assertEqual('pci', interface_address.get('type'))
+ self.assertEqual('0x' + vm_pci.split(':')[0],
+ interface_address.get('domain'))
+ self.assertEqual('0x' + vm_pci.split(':')[1],
+ interface_address.get('bus'))
+ self.assertEqual('0x' + vm_pci.split(':')[2].split('.')[0],
+ interface_address.get('slot'))
+ self.assertEqual('0x' + vm_pci.split(':')[2].split('.')[1],
+ interface_address.get('function'))
def test_create_snapshot_qemu(self):
- result = "/var/lib/libvirt/images/0.qcow2"
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "a", ""))
- ssh.return_value = ssh_mock
- image = model.Libvirt.create_snapshot_qemu(ssh_mock, "0", "ubuntu.img")
- self.assertEqual(image, result)
-
- @mock.patch.object(model.Libvirt, 'pin_vcpu_for_perf')
- @mock.patch.object(model.Libvirt, 'create_snapshot_qemu')
+ self.mock_ssh.execute = mock.Mock(return_value=(0, 0, 0))
+ index = 1
+ vm_image = '/var/lib/libvirt/images/%s.qcow2' % index
+ base_image = '/tmp/base_image'
+
+ model.Libvirt.create_snapshot_qemu(self.mock_ssh, index, base_image)
+ self.mock_ssh.execute.assert_has_calls([
+ mock.call('rm -- "%s"' % vm_image),
+ mock.call('test -r %s' % base_image),
+ mock.call('qemu-img create -f qcow2 -o backing_file=%s %s' %
+ (base_image, vm_image))
+ ])
+
+ @mock.patch.object(os.path, 'basename', return_value='base_image')
+ @mock.patch.object(os.path, 'normpath')
+ @mock.patch.object(os, 'access', return_value=True)
+ def test_create_snapshot_qemu_no_image_remote(self,
+ mock_os_access, mock_normpath, mock_basename):
+ self.mock_ssh.execute = mock.Mock(
+ side_effect=[(0, 0, 0), (1, 0, 0), (0, 0, 0), (0, 0, 0)])
+ index = 1
+ vm_image = '/var/lib/libvirt/images/%s.qcow2' % index
+ base_image = '/tmp/base_image'
+ mock_normpath.return_value = base_image
+
+ model.Libvirt.create_snapshot_qemu(self.mock_ssh, index, base_image)
+ self.mock_ssh.execute.assert_has_calls([
+ mock.call('rm -- "%s"' % vm_image),
+ mock.call('test -r %s' % base_image),
+ mock.call('mv -- "/tmp/%s" "%s"' % ('base_image', base_image)),
+ mock.call('qemu-img create -f qcow2 -o backing_file=%s %s' %
+ (base_image, vm_image))
+ ])
+ mock_os_access.assert_called_once_with(base_image, os.R_OK)
+ mock_normpath.assert_called_once_with(base_image)
+ mock_basename.assert_has_calls([mock.call(base_image)])
+ self.mock_ssh.put_file.assert_called_once_with(base_image,
+ '/tmp/base_image')
+
+ @mock.patch.object(os, 'access', return_value=False)
+ def test_create_snapshot_qemu_no_image_local(self, mock_os_access):
+ self.mock_ssh.execute = mock.Mock(side_effect=[(0, 0, 0), (1, 0, 0)])
+ base_image = '/tmp/base_image'
+
+ with self.assertRaises(exceptions.LibvirtQemuImageBaseImageNotPresent):
+ model.Libvirt.create_snapshot_qemu(self.mock_ssh, 3, base_image)
+ mock_os_access.assert_called_once_with(base_image, os.R_OK)
+
+ def test_create_snapshot_qemu_error_qemuimg_command(self):
+ self.mock_ssh.execute = mock.Mock(
+ side_effect=[(0, 0, 0), (0, 0, 0), (1, 0, 0)])
+ index = 1
+ vm_image = '/var/lib/libvirt/images/%s.qcow2' % index
+ base_image = '/tmp/base_image'
+
+ with self.assertRaises(exceptions.LibvirtQemuImageCreateError):
+ model.Libvirt.create_snapshot_qemu(self.mock_ssh, index,
+ base_image)
+ self.mock_ssh.execute.assert_has_calls([
+ mock.call('rm -- "%s"' % vm_image),
+ mock.call('test -r %s' % base_image),
+ mock.call('qemu-img create -f qcow2 -o backing_file=%s %s' %
+ (base_image, vm_image))
+ ])
+
+ @mock.patch.object(model.Libvirt, 'pin_vcpu_for_perf', return_value='4,5')
+ @mock.patch.object(model.Libvirt, 'create_snapshot_qemu',
+ return_value='qemu_image')
def test_build_vm_xml(self, mock_create_snapshot_qemu,
- *args):
- # NOTE(ralonsoh): this test doesn't cover function execution. This test
- # should also check mocked function calls.
- cfg_file = 'test_config_file.cfg'
- self.addCleanup(os.remove, cfg_file)
- result = [4]
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, "a", ""))
- ssh.return_value = ssh_mock
- mock_create_snapshot_qemu.return_value = "0.img"
-
- status = model.Libvirt.build_vm_xml(ssh_mock, {}, cfg_file, 'vm_0', 0)
- self.assertEqual(status[0], result[0])
+ mock_pin_vcpu_for_perf):
+ extra_specs = {'hw:cpu_cores': '4',
+ 'hw:cpu_sockets': '3',
+ 'hw:cpu_threads': '2',
+ 'cputune': 'cool'}
+ flavor = {'ram': '1024',
+ 'extra_specs': extra_specs,
+ 'hw_socket': '1',
+ 'images': 'images'}
+ mac = model.StandaloneContextHelper.get_mac_address(0x00)
+ _uuid = uuid.uuid4()
+ connection = mock.Mock()
+ with mock.patch.object(model.StandaloneContextHelper,
+ 'get_mac_address', return_value=mac) as \
+ mock_get_mac_address, \
+ mock.patch.object(uuid, 'uuid4', return_value=_uuid):
+ xml_out, mac = model.Libvirt.build_vm_xml(
+ connection, flavor, 'vm_name', 100)
+
+ xml_ref = model.VM_TEMPLATE.format(vm_name='vm_name',
+ random_uuid=_uuid, mac_addr=mac, memory='1024', vcpu='8', cpu='4',
+ numa_cpus='0-7', socket='3', threads='2',
+ vm_image='qemu_image', cpuset='4,5', cputune='cool')
+ self.assertEqual(xml_ref, xml_out)
+ mock_get_mac_address.assert_called_once_with(0x00)
+ mock_create_snapshot_qemu.assert_called_once_with(
+ connection, 100, 'images')
+ mock_pin_vcpu_for_perf.assert_called_once_with(connection, '1')
# TODO: Edit this test to test state instead of output
# update_interrupts_hugepages_perf does not return anything
class OvsDeployTestCase(unittest.TestCase):
- NETWORKS = {
- 'mgmt': {'cidr': '152.16.100.10/24'},
- 'private_0': {
- 'phy_port': "0000:05:00.0",
- 'vpci': "0000:00:07.0",
- 'driver': 'i40e',
- 'mac': '',
- 'cidr': '152.16.100.10/24',
- 'gateway_ip': '152.16.100.20'},
- 'public_0': {
- 'phy_port': "0000:05:00.1",
- 'vpci': "0000:00:08.0",
- 'driver': 'i40e',
- 'mac': '',
- 'cidr': '152.16.40.10/24',
- 'gateway_ip': '152.16.100.20'}
- }
-
- @mock.patch('yardstick.ssh.SSH')
- def setUp(self, mock_ssh):
- self.ovs_deploy = model.OvsDeploy(mock_ssh, '/tmp/dpdk-devbind.py', {})
+ OVS_DETAILS = {'version': {'ovs': 'ovs_version', 'dpdk': 'dpdk_version'}}
- def test___init__(self):
- self.assertIsNotNone(self.ovs_deploy.connection)
-
- @mock.patch('yardstick.benchmark.contexts.standalone.model.os')
- def test_prerequisite(self, *args):
- # NOTE(ralonsoh): this test should check mocked function calls.
- self.ovs_deploy.helper = mock.Mock()
- self.assertIsNone(self.ovs_deploy.prerequisite())
-
- @mock.patch('yardstick.benchmark.contexts.standalone.model.os')
- def test_prerequisite_2(self, *args):
- # NOTE(ralonsoh): this test should check mocked function calls. Rename
- # this test properly.
- self.ovs_deploy.helper = mock.Mock()
- self.ovs_deploy.connection.execute = mock.Mock(
- return_value=(1, '1.2.3.4 00:00:00:00:00:01', ''))
- self.ovs_deploy.prerequisite = mock.Mock()
- self.assertIsNone(self.ovs_deploy.ovs_deploy())
+ def setUp(self):
+ self._mock_ssh = mock.patch.object(ssh, 'SSH')
+ self.mock_ssh = self._mock_ssh.start()
+ self.ovs_deploy = model.OvsDeploy(self.mock_ssh,
+ '/tmp/dpdk-devbind.py',
+ self.OVS_DETAILS)
+ self._mock_path_isfile = mock.patch.object(os.path, 'isfile')
+ self._mock_path_join = mock.patch.object(os.path, 'join')
+ self.mock_path_isfile = self._mock_path_isfile.start()
+ self.mock_path_join = self._mock_path_join.start()
+
+ self.addCleanup(self._stop_mock)
+
+ def _stop_mock(self):
+ self._mock_ssh.stop()
+ self._mock_path_isfile.stop()
+ self._mock_path_join.stop()
+
+ @mock.patch.object(model.StandaloneContextHelper, 'install_req_libs')
+ def test_prerequisite(self, mock_install_req_libs):
+ pkgs = ["git", "build-essential", "pkg-config", "automake",
+ "autotools-dev", "libltdl-dev", "cmake", "libnuma-dev",
+ "libpcap-dev"]
+ self.ovs_deploy.prerequisite()
+ mock_install_req_libs.assert_called_once_with(
+ self.ovs_deploy.connection, pkgs)
+
+ def test_ovs_deploy_no_file(self):
+ self.mock_path_isfile.return_value = False
+ mock_file = mock.Mock()
+ self.mock_path_join.return_value = mock_file
+
+ self.ovs_deploy.ovs_deploy()
+ self.mock_path_isfile.assert_called_once_with(mock_file)
+ self.mock_path_join.assert_called_once_with(
+ constants.YARDSTICK_ROOT_PATH,
+ 'yardstick/resources/scripts/install/',
+ self.ovs_deploy.OVS_DEPLOY_SCRIPT)
+
+ @mock.patch.object(os.environ, 'get', return_value='test_proxy')
+ def test_ovs_deploy(self, mock_env_get):
+ self.mock_path_isfile.return_value = True
+ mock_deploy_file = mock.Mock()
+ mock_remove_ovs_deploy = mock.Mock()
+ self.mock_path_join.side_effect = [mock_deploy_file,
+ mock_remove_ovs_deploy]
+ dpdk_version = self.OVS_DETAILS['version']['dpdk']
+ ovs_version = self.OVS_DETAILS['version']['ovs']
+
+ with mock.patch.object(self.ovs_deploy.connection, 'put') as \
+ mock_put, \
+ mock.patch.object(self.ovs_deploy.connection, 'execute') as \
+ mock_execute, \
+ mock.patch.object(self.ovs_deploy, 'prerequisite'):
+ mock_execute.return_value = (0, 0, 0)
+ self.ovs_deploy.ovs_deploy()
+
+ self.mock_path_isfile.assert_called_once_with(mock_deploy_file)
+ self.mock_path_join.assert_has_calls([
+ mock.call(constants.YARDSTICK_ROOT_PATH,
+ 'yardstick/resources/scripts/install/',
+ self.ovs_deploy.OVS_DEPLOY_SCRIPT),
+ mock.call(self.ovs_deploy.bin_path,
+ self.ovs_deploy.OVS_DEPLOY_SCRIPT)
+ ])
+ mock_put.assert_called_once_with(mock_deploy_file,
+ mock_remove_ovs_deploy)
+ cmd = ("sudo -E %(remote_ovs_deploy)s --ovs='%(ovs_version)s' "
+ "--dpdk='%(dpdk_version)s' -p='%(proxy)s'" %
+ {'remote_ovs_deploy': mock_remove_ovs_deploy,
+ 'ovs_version': ovs_version,
+ 'dpdk_version': dpdk_version,
+ 'proxy': 'test_proxy'})
+ mock_execute.assert_called_once_with(cmd)
+ mock_env_get.assert_called_once_with('http_proxy', '')
Code Review / yardstick.git / blobdiff