import copy
import os
-import mock
-import unittest
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
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_ssh = mock.Mock()
+ self.mock_ssh = self._mock_ssh.start()
+ self.addCleanup(self._cleanup)
+
+ def _cleanup(self):
+ 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(
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)
+ 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',
def setUp(self):
self._mock_ssh = mock.patch.object(ssh, 'SSH')
- self.mock_ssh = self._mock_ssh .start()
+ self.mock_ssh = self._mock_ssh.start()
self.ovs_deploy = model.OvsDeploy(self.mock_ssh,
'/tmp/dpdk-devbind.py',
self.OVS_DETAILS)