diff --git a/vix/disk_manager.py b/vix/disk_manager.py index d9af2fe..d972a7c 100644 --- a/vix/disk_manager.py +++ b/vix/disk_manager.py @@ -102,7 +102,7 @@ def _resize_disk_vdisk_man(self, disk_path, new_size_mb): self._exec_cmd(args) def _resize_disk_qemu(self, disk_path, new_size_mb, new_disk_type): - tmp_disk_path = "%s.raw" & disk_path + tmp_disk_path = "%s.raw" % disk_path try: args = ["qemu-img", "convert", "-O", DISK_TYPE_RAW, disk_path, tmp_disk_path] diff --git a/vix/tests/compute/__init__.py b/vix/tests/compute/__init__.py new file mode 100644 index 0000000..090fc06 --- /dev/null +++ b/vix/tests/compute/__init__.py @@ -0,0 +1,16 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2013 Cloudbase Solutions Srl +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. diff --git a/vix/tests/compute/test_driver.py b/vix/tests/compute/test_driver.py new file mode 100644 index 0000000..fb63bc8 --- /dev/null +++ b/vix/tests/compute/test_driver.py @@ -0,0 +1,558 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2013 Cloudbase Solutions Srl +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import os +import platform +import unittest + +from nova.compute import task_states +from nova.openstack.common import jsonutils +from oslo.config import cfg +from vix.compute import driver +from vix.compute import image_cache +from vix.compute import pathutils +from vix import utils +from vix import vixlib +from vix import vixutils + + +class VixDriverTestCase(unittest.TestCase): + """Unit tests for Nova VIX driver""" + + def setUp(self): + self.CONF = mock.MagicMock() + cfg.CONF = mock.MagicMock(return_value=self.CONF) + virtapi = mock.MagicMock() + self._conn = mock.MagicMock() + self._pathutils = mock.MagicMock() + self._image_cache = mock.MagicMock() + + vixutils.VixConnection = mock.MagicMock(return_value=self._conn) + pathutils.PathUtils = mock.MagicMock(return_value=self._pathutils) + image_cache.ImageCache = mock.MagicMock(return_value=self._image_cache) + self._driver = driver.VixDriver(virtapi) + + def test_list_instances(self): + self._driver.list_instances() + self._conn.list_running_vms.assert_called_once() + + def test_delete_existing_instance(self): + fake_instance_name = 'fake_name' + fake_path = 'fake/path' + self._pathutils.get_vmx_path.return_value = fake_path + + self._driver._delete_existing_instance(fake_instance_name) + + self._pathutils.get_vmx_path.assert_called_with('fake_name') + self._conn.vm_exists.assert_called_with(fake_path) + self._conn.unregister_vm_and_delete_files.assert_called_with( + fake_path, True) + + def test_clone_vmdk_vm(self): + fake_src_vmdk = 'src/fake.vmdk' + fake_file_name = 'fake.vmdk' + fake_root_vmdk_path = 'root/fake.vmdk' + fake_dest_vmx_path = 'dest/fake.vmdk' + fake_vmdk_path = 'path/fake.vmdk' + fake_split = mock.MagicMock() + fake_base = mock.MagicMock() + + os.path.basename = mock.MagicMock(return_value=fake_base) + os.path.splitext = mock.MagicMock(return_value=fake_split) + os.path.dirname = mock.MagicMock() + os.path.join = mock.MagicMock(return_value=fake_vmdk_path) + vixutils.get_vmx_value = mock.MagicMock() + vixutils.get_vmx_value.return_value = fake_file_name + vixutils.set_vmx_value = mock.MagicMock() + + self._driver._clone_vmdk_vm(fake_src_vmdk, fake_root_vmdk_path, + fake_dest_vmx_path) + + self._conn.create_vm.assert_called_once() + self._conn.clone_vm.assert_called_with(fake_split[0] + ".vmx", + fake_dest_vmx_path, True) + self._pathutils.rename.assert_called_with(fake_vmdk_path, + fake_root_vmdk_path) + vixutils.set_vmx_value.assert_called_with(fake_split[0] + ".vmsd", + "sentinel0", fake_base) + + def test_check_player_compatibility(self): + vixutils.get_vix_host_type = mock.MagicMock( + return_value=vixutils.VIX_VMWARE_PLAYER) + self.assertRaises(NotImplementedError, + self._driver._check_player_compatibility, True) + + def _test_spawn(self, cow): + + fake_admin_password = 'fake password' + fake_instance = mock.MagicMock() + fake_context = mock.MagicMock() + fake_image_meta = mock.MagicMock() + fake_injected_files = mock.MagicMock() + fake_network_info = mock.MagicMock() + fake_block_device_info = mock.MagicMock() + fake_image_info = mock.MagicMock() + fake_iso_image_ids = ['fakeid'] + fake_base_vmdk_path = 'fake/base/vmdk/path' + fake_root_vmdk_path = 'fake/root/vmdk/path' + fake_vmx_path = 'fake/vmx/path' + fake_floppy_path = 'fake/floppy/path' + + self._image_cache.get_image_info.return_value = fake_image_info + self._driver._check_player_compatibility = mock.MagicMock() + self._driver._delete_existing_instance = mock.MagicMock() + self._driver._clone_vmdk_vm = mock.MagicMock() + self._image_cache.get_cached_image.return_value = fake_base_vmdk_path + self._pathutils.get_root_vmdk_path.return_value = fake_root_vmdk_path + self._pathutils.get_vmx_path.return_value = fake_vmx_path + self._pathutils.get_floppy_path.return_value = fake_floppy_path + os.path.join = mock.MagicMock(return_value=fake_base_vmdk_path) + fake_image_info.get().get().lower.return_value = str(cow).lower() + fake_image_info.get().get().split.return_value = fake_iso_image_ids + utils.get_free_port = mock.MagicMock() + utils.get_free_port.return_value = 9999 + + self._driver.spawn(context=fake_context, instance=fake_instance, + image_meta=fake_image_meta, + injected_files=fake_injected_files, + admin_password=fake_admin_password, + network_info=fake_network_info, + block_device_info=fake_block_device_info) + print fake_image_info.get().get.mock_calls + + self._image_cache.get_image_info.assert_called_with( + fake_context, fake_instance['image_ref']) + + self._driver._check_player_compatibility.assert_called_with(cow) + self._driver._delete_existing_instance.assert_called_with( + fake_instance['name']) + self._pathutils.create_instance_dir.assert_called_with( + fake_instance['name']) + self.assertEqual(self._image_cache.get_cached_image.call_count, 3) + self._pathutils.get_root_vmdk_path.assert_called_with( + fake_instance['name']) + self._pathutils.get_vmx_path.assert_called_with(fake_instance['name']) + if cow: + self._driver._clone_vmdk_vm.assert_called_with( + fake_base_vmdk_path, fake_root_vmdk_path, fake_vmx_path) + self.assertEqual(self._pathutils.copy.call_count, 1) + self._conn.update_vm.assert_called_with( + vmx_path=fake_vmx_path, + display_name=fake_instance.get("display_name"), + guest_os=fake_image_info.get().get(), + num_vcpus=fake_instance['vcpus'], + mem_size_mb=fake_instance['memory_mb'], + iso_paths=[fake_base_vmdk_path, fake_base_vmdk_path], + floppy_path=fake_floppy_path, + networks=[], + boot_order=fake_image_info.get().get(), + vnc_enabled=True, + vnc_port=9999, nested_hypervisor=fake_image_info.get().get()) + else: + self.assertEqual(self._pathutils.copy.call_count, 2) + self._conn.create_vm.assert_called_with( + vmx_path=fake_vmx_path, + display_name=fake_instance.get("display_name"), + guest_os=fake_image_info.get().get(), + num_vcpus=fake_instance['vcpus'], + mem_size_mb=fake_instance['memory_mb'], + disk_paths=[fake_root_vmdk_path], + iso_paths=[fake_base_vmdk_path, fake_base_vmdk_path], + floppy_path=fake_floppy_path, + networks=[], + boot_order=fake_image_info.get().get(), + vnc_enabled=True, + vnc_port=9999, nested_hypervisor=fake_image_info.get().get()) + + os.path.join.assert_called_with( + self._conn.get_tools_iso_path(), + "%s.iso" % fake_image_info.get().get()) + self._pathutils.get_floppy_path.assert_called_with( + fake_instance['name']) + utils.get_free_port.assert_called_once() + self._conn.open_vm.assert_called_with(fake_vmx_path) + + def test_spawn_cow(self): + self._test_spawn(cow=True) + + def test_spawn_no_cow(self): + self._test_spawn(cow=False) + + def _test_exec_vm_action(self, vm_exists): + fake_instance = mock.MagicMock() + fake_action = mock.MagicMock() + + fake_path = 'fake/path' + self._pathutils.get_vmx_path.return_value = fake_path + self._conn.vm_exists.return_value = vm_exists + if not vm_exists: + self.assertRaises(Exception, self._driver._exec_vm_action, + fake_instance, fake_action) + else: + response = self._driver._exec_vm_action(fake_instance, + fake_action) + self._conn.open_vm.assert_called_with(fake_path) + self.assertTrue(response is not None) + + def test_exec_vm_action_vm_exists_false(self): + self._test_exec_vm_action(False) + + def test_exec_vm_action_vm_exists_true(self): + self._test_exec_vm_action(True) + + def test_reboot(self): + fake_instance = mock.MagicMock() + fake_context = mock.MagicMock() + fake_network_info = mock.MagicMock() + vixutils.reboot = mock.MagicMock() + self._driver.reboot(fake_context, fake_instance, + fake_network_info, reboot_type=None) + vixutils.reboot.assert_called_once() + + def test_destroy(self): + fake_instance = mock.MagicMock() + fake_network_info = mock.MagicMock() + self._driver._delete_existing_instance = mock.MagicMock() + self._driver.destroy(fake_instance, fake_network_info) + self._driver._delete_existing_instance.assert_called_with( + fake_instance['name'], True) + + def test_get_info(self): + fake_instance = mock.MagicMock() + + vixutils.get_power_state = mock.MagicMock( + return_value=vixlib.VIX_POWERSTATE_POWERED_ON) + + response = self._driver.get_info(fake_instance) + print response + vixutils.get_power_state.assert_called_once() + self.assertTrue(response is not None) + + def test_attach_volume(self): + fake_instance = mock.MagicMock() + fake_context = mock.MagicMock() + fake_connection_info = mock.MagicMock() + fake_mountpoint = mock.MagicMock() + + self.assertRaises(NotImplementedError, + self._driver.attach_volume, fake_context, + fake_connection_info, fake_instance, + fake_mountpoint) + + def test_deattach_volume(self): + fake_instance = mock.MagicMock() + fake_connection_info = mock.MagicMock() + fake_mountpoint = mock.MagicMock() + + self.assertRaises(NotImplementedError, + self._driver.detach_volume, + fake_connection_info, fake_instance, + fake_mountpoint) + + def test_get_volume_connector(self): + fake_instance = mock.MagicMock() + self.assertRaises(NotImplementedError, + self._driver.get_volume_connector, + fake_instance) + + def test_get_host_memory_info(self): + total_mem = 2147483648 + free_mem = 1073741824 + utils.get_host_memory_info = mock.MagicMock() + utils.get_host_memory_info.return_value = (total_mem, free_mem) + response = self._driver._get_host_memory_info() + utils.get_host_memory_info.assert_called_once() + self.assertEqual(response, (2048, 1024, 1024)) + + def test_get_local_hdd_info_gb(self): + total_disk = 2147483648 + free_disk = 1073741824 + fake_dir = 'fake dir' + utils.get_disk_info = mock.MagicMock() + utils.get_disk_info.return_value = (total_disk, free_disk) + self._pathutils.get_instances_dir = mock.MagicMock() + self._pathutils.get_instances_dir.return_value = fake_dir + response = self._driver._get_local_hdd_info_gb() + utils.get_disk_info.assert_called_once_with(fake_dir) + self.assertEqual(response, (2, 1, 1)) + + def test_get_hypervisor_version(self): + self._conn.get_software_version.return_value = 10 + response = self._driver._get_hypervisor_version() + self._conn.get_software_version.assert_called_once() + self.assertEqual(response, 10) + + def test_get_available_resource(self): + fake_nodename = 'fake_name' + total_disk = 2 * 1024 * 1024 * 1024 + free_disk = 1 * 1024 * 1024 * 1024 + total_mem = 2048 * 1024 * 1024 + free_mem = 1024 * 1024 * 1024 + vcpus = 2 + fake_dir = 'fake dir' + compare_dict = {'vcpus': vcpus, + 'memory_mb': 2048, + 'memory_mb_used': 1024, + 'local_gb': 2, + 'local_gb_used': 1, + 'hypervisor_type': "vix", + 'hypervisor_version': 10, + 'hypervisor_hostname': 'fake_hostname', + 'vcpus_used': 0, + 'cpu_info': 0, + 'supported_instances': 0} + + jsonutils.dumps = mock.MagicMock() + jsonutils.dumps.return_value = 0 + platform.node = mock.MagicMock() + platform.node.return_value = 'fake_hostname' + self._conn.get_software_version = mock.MagicMock() + self._conn.get_software_version.return_value = 10 + utils.get_host_memory_info = mock.MagicMock() + utils.get_host_memory_info.return_value = (total_mem, free_mem) + utils.get_disk_info = mock.MagicMock() + utils.get_disk_info.return_value = (total_disk, free_disk) + self._pathutils.get_instances_dir = mock.MagicMock() + self._pathutils.get_instances_dir.return_value = fake_dir + utils.get_cpu_count = mock.MagicMock() + utils.get_cpu_count.return_value = vcpus + + response = self._driver.get_available_resource(fake_nodename) + utils.get_host_memory_info.assert_called_once() + utils.get_disk_info.assert_called_once_with(fake_dir) + self._conn.get_software_version.assert_called_once() + platform.node.assert_called_once() + self.assertEqual(jsonutils.dumps.call_count, 2) + self.assertEqual(response, compare_dict) + + def test_update_stats(self): + total_disk = 2 * 1024 * 1024 * 1024 + free_disk = 1 * 1024 * 1024 * 1024 + total_mem = 2048 * 1024 * 1024 + free_mem = 1024 * 1024 * 1024 + fake_dir = 'fake dir' + compare_dict = {'host_memory_total': 2048, + 'host_memory_overhead': 1024, + 'host_memory_free': 1024, + 'host_memory_free_computed': 1024, + 'disk_total': 2, + 'disk_used': 1, + 'disk_available': 1, + 'hypervisor_hostname': 'fake_hostname', + 'supported_instances': [('i686', 'vix', 'hvm'), + ('x86_64', 'vix', 'hvm')]} + platform.node = mock.MagicMock() + platform.node.return_value = 'fake_hostname' + utils.get_host_memory_info = mock.MagicMock() + utils.get_host_memory_info.return_value = (total_mem, free_mem) + utils.get_disk_info = mock.MagicMock() + utils.get_disk_info.return_value = (total_disk, free_disk) + self._pathutils.get_instances_dir = mock.MagicMock() + self._pathutils.get_instances_dir.return_value = fake_dir + + self._driver._update_stats() + + utils.get_host_memory_info.assert_called_once() + utils.get_disk_info.assert_called_once_with(fake_dir) + platform.node.assert_called_once() + self.assertEqual(self._driver._stats, compare_dict) + + def _test_get_host_stats(self, refresh): + self._driver.get_host_stats(refresh=refresh) + self.assertTrue(self._driver._stats is not None) + + def test_get_host_stats_refresh_true(self): + self._test_get_host_stats(True) + + def test_get_host_stats_refresh_false(self): + self._test_get_host_stats(False) + + def _test_snapshot(self, feature_supported): + fake_name = 'fake name' + fake_instance = mock.MagicMock() + fake_context = mock.MagicMock() + fake_update_task_state = mock.MagicMock() + fake_path = 'fake/path' + fake_root_vmdk_path = 'fake/root/vmdk/path' + fake_vm = mock.MagicMock() + self._conn.open_vm.return_value = fake_vm + self._pathutils.get_root_vmdk_path.return_value = fake_root_vmdk_path + self._pathutils.get_vmx_path.return_value = fake_path + vixutils.VixVM.create_snapshot = mock.MagicMock() + vixutils.VixVM.remove_snapshot = mock.MagicMock() + vixutils.get_vix_host_type = mock.MagicMock() + + if not feature_supported: + host_type = vixutils.VIX_VMWARE_PLAYER + vixutils.get_vix_host_type.return_value = host_type + self.assertRaises(NotImplementedError, self._driver.snapshot, + fake_context, fake_instance, fake_name, + fake_update_task_state) + else: + self._driver.snapshot(fake_context, fake_instance, fake_name, + fake_update_task_state) + + vixutils.get_vix_host_type.assert_called_once() + self._pathutils.get_vmx_path.assert_called_with( + fake_instance['name']) + self._conn.open_vm.assert_called_with(fake_path) + print self._conn.open_vm.mock_calls + self.assertEqual(fake_update_task_state.call_count, 2) + fake_vm.__enter__().create_snapshot.assert_called_with( + name="Nova snapshot") + self._image_cache.save_glance_image.assert_called_with( + fake_context, fake_name, fake_root_vmdk_path) + vixutils.VixVM.remove_snapshot.assert_called_once() + + def test_snapshot_not_implemented(self): + self._test_snapshot(False) + + def test_snapshot(self): + self._test_snapshot(True) + + def test_pause(self): + fake_instance = mock.MagicMock() + vixutils.VixVM.pause = mock.MagicMock() + self._driver.pause(fake_instance) + vixutils.VixVM.pause.assert_called_once() + + def test_unpause(self): + fake_instance = mock.MagicMock() + vixutils.VixVM.unpause = mock.MagicMock() + self._driver.unpause(fake_instance) + vixutils.VixVM.unpause.assert_called_once() + + def test_suspend(self): + fake_instance = mock.MagicMock() + vixutils.VixVM.suspend = mock.MagicMock() + self._driver.suspend(fake_instance) + vixutils.VixVM.suspend.assert_called_once() + + def test_resume(self): + fake_instance = mock.MagicMock() + fake_network_info = mock.MagicMock() + vixutils.VixVM.power_on = mock.MagicMock() + self._driver.resume(fake_instance, fake_network_info) + vixutils.VixVM.power_on.assert_called_once() + + def test_power_off(self): + fake_instance = mock.MagicMock() + vixutils.VixVM.power_off = mock.MagicMock() + self._driver.power_off(fake_instance) + vixutils.VixVM.power_off.assert_called_once() + + def test_power_on(self): + fake_instance = mock.MagicMock() + fake_context = mock.MagicMock() + fake_network_info = mock.MagicMock() + vixutils.VixVM.power_on = mock.MagicMock() + self._driver.power_on(fake_instance, fake_context, + fake_network_info) + vixutils.VixVM.power_on.assert_called_once() + + def test_live_migration(self): + fake_context = mock.MagicMock() + fake_recover_method = mock.MagicMock() + fake_dest = 'fake/dest' + fake_post_method = mock.MagicMock() + fake_instance = mock.MagicMock() + self.assertRaises(NotImplementedError, self._driver.live_migration, + fake_context, fake_instance, fake_dest, + fake_post_method, fake_recover_method) + + def test_pre_live_migration(self): + fake_context = mock.MagicMock() + fake_block_device_info = mock.MagicMock() + fake_disk = 'fake/dest' + fake_network_info = mock.MagicMock() + fake_instance = mock.MagicMock() + self.assertRaises(NotImplementedError, + self._driver.pre_live_migration, fake_context, + fake_instance, fake_block_device_info, + fake_network_info, fake_disk) + + def test_post_live_migration_at_destination(self): + fake_context = mock.MagicMock() + fake_network_info = mock.MagicMock() + fake_instance_ref = mock.MagicMock() + self.assertRaises(NotImplementedError, + self._driver.post_live_migration_at_destination, + fake_context, fake_instance_ref, fake_network_info) + + def test_check_can_live_migrate_destination(self): + fake_context = mock.MagicMock() + fake_src_computer = mock.MagicMock() + fake_dest_computer = mock.MagicMock() + fake_instance_ref = mock.MagicMock() + self.assertRaises(NotImplementedError, + self._driver.check_can_live_migrate_destination, + fake_context, fake_instance_ref, + fake_src_computer, fake_dest_computer) + + def test_check_can_live_migrate_destination_cleanup(self): + fake_context = mock.MagicMock() + fake_dest_data = mock.MagicMock() + self.assertRaises( + NotImplementedError, + self._driver.check_can_live_migrate_destination_cleanup, + fake_context, fake_dest_data) + + def test_check_can_live_migrate_source(self): + fake_context = mock.MagicMock() + fake_instance_ref = mock.MagicMock() + fake_dest_data = mock.MagicMock() + self.assertRaises(NotImplementedError, + self._driver.check_can_live_migrate_source, + fake_context, fake_instance_ref, fake_dest_data) + + def test_get_host_ip_addr(self): + response = self._driver.get_host_ip_addr() + self.assertTrue(response is not None) + + def _test_get_vnc_console(self, vnc_enabled): + fake_instance = mock.MagicMock() + fake_path = 'fake/path' + vnc_port = 9999 + open_vm_enter = mock.MagicMock() + self._conn.open_vm().__enter__.return_value = open_vm_enter + open_vm_enter.get_vnc_settings.return_value = ( + vnc_enabled, vnc_port) + self._pathutils.get_vmx_path.return_value = fake_path + + if vnc_enabled: + response = self._driver.get_vnc_console(fake_instance) + self._pathutils.get_vmx_path.assert_called_with( + fake_instance['name']) + self._conn.open_vm.assert_called_with(fake_path) + open_vm_enter.get_vnc_settings.assert_called_once() + + self.assertTrue(response is not None) + else: + self.assertRaises(utils.VixException, + self._driver.get_vnc_console, fake_instance) + + def test_get_vnc_console(self): + self._test_get_vnc_console(True) + + def test_get_vnc_console_disabled(self): + self._test_get_vnc_console(False) + + def test_get_console_output(self): + fake_instance = mock.MagicMock() + reponse = self._driver.get_console_output(fake_instance) + self.assertEqual(reponse, '') diff --git a/vix/tests/compute/test_image_cache.py b/vix/tests/compute/test_image_cache.py new file mode 100644 index 0000000..7838bc0 --- /dev/null +++ b/vix/tests/compute/test_image_cache.py @@ -0,0 +1,149 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2013 Cloudbase Solutions Srl +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import os +import unittest +import sys +if sys.platform == 'win32': + import _winreg + import win32api +from nova.image import glance +from nova.openstack.common import excutils +from nova.virt import images +from vix.compute import image_cache +from vix.compute import pathutils + + +class VixUtilsTestCase(unittest.TestCase): + """Unit tests for utility class""" + + def setUp(self): + self._image_cache = image_cache.ImageCache() + self._pathutils = pathutils.PathUtils() + + def test_get_image_info(self): + fake_context = mock.MagicMock() + fake_image_service = mock.MagicMock() + fake_image_id = "1" + glance.get_remote_image_service = mock.MagicMock() + glance.get_remote_image_service.return_value = \ + (fake_image_service, fake_image_id) + + response = self._image_cache.get_image_info(fake_context, + fake_image_id) + + glance.get_remote_image_service.assert_called_with(fake_context, + fake_image_id) + self.assertEqual(response, fake_image_service.show(fake_context, + fake_image_id)) + fake_image_service.show.assert_called_with(fake_context, fake_image_id) + + def test_save_glance_image(self): + fake_context = mock.MagicMock() + fake_name = mock.MagicMock() + fake_image_vmdk_path = mock.MagicMock() + fake_image_metadata = {"is_public": False, + "disk_format": "vmdk", + "container_format": "bare", + "properties": {}} + fake_glance_image_service = mock.MagicMock() + fake_image_id = mock.MagicMock() + + glance.get_remote_image_service = mock.MagicMock( + return_value=(fake_glance_image_service, fake_image_id)) + fake_glance_image_service.update = mock.MagicMock() + + with mock.patch('vix.compute.image_cache.open', + mock.mock_open(read_data='fake data'), + create=True) as m: + self._image_cache.save_glance_image(fake_context, fake_name, + fake_image_vmdk_path) + fake_glance_image_service.update.assert_called_with( + fake_context, fake_image_id, fake_image_metadata, m()) + + glance.get_remote_image_service.assert_called_with(fake_context, + fake_name) + + def _test_get_cached_image(self, image_exists, exception=False, + image_path_exists=False): + fake_context = mock.MagicMock() + fake_image_id = mock.MagicMock() + fake_user_id = mock.MagicMock() + fake_project_id = mock.MagicMock() + fake_disk_format = mock.MagicMock() + fake_image_info = mock.MagicMock() + fake_image_path = mock.MagicMock() + fake_base_vmdk_dir = mock.MagicMock() + self._image_cache.get_image_info = mock.MagicMock() + self._image_cache.get_image_info.return_value = fake_image_info + fake_image_info.get = mock.MagicMock() + fake_image_info.get.return_value = fake_disk_format + + self._image_cache._pathutils.exists = mock.MagicMock() + self._image_cache._pathutils.exists.return_value = image_exists + self._image_cache._pathutils.get_base_vmdk_dir = mock.MagicMock( + return_value=fake_base_vmdk_dir) + + os.path.join = mock.MagicMock() + os.path.join.return_value = fake_image_path + + images.fetch = mock.MagicMock() + if not image_exists: + if exception: + excutils.save_and_reraise_exception = mock.MagicMock() + excutils.save_and_reraise_exception.side_effect = Exception + images.fetch.side_effect = Exception + self.assertRaises(Exception, + self._image_cache.get_cached_image, + fake_context, fake_image_id, fake_user_id, + fake_project_id) + else: + response = self._image_cache.get_cached_image( + fake_context, fake_image_id, fake_user_id, fake_project_id) + self.assertEqual(response, fake_image_path) + #It cannot go here unless it s a race - this fails + images.fetch.assert_called_with(fake_context, fake_image_id, + fake_image_path, fake_user_id, + fake_project_id) + else: + response = self._image_cache.get_cached_image(fake_context, + fake_image_id, + fake_user_id, + fake_project_id) + self.assertEqual(response, fake_image_path) + + self._image_cache._pathutils.exists.assert_called_with( + fake_image_path) + self._image_cache.get_image_info.assert_called_with(fake_context, + fake_image_id) + fake_image_info.get.assert_called_with("disk_format") + self._image_cache._pathutils.get_base_vmdk_dir.assert_called_once() + os.path.join.assert_called_with(fake_base_vmdk_dir, + fake_image_id + "." + fake_disk_format) + + def test_get_cached_image_existent(self): + self._test_get_cached_image(True) + + def test_get_cached_image_not_existent(self): + self._test_get_cached_image(False) + + def test_get_cached_image_not_existent_and_exception(self): + self._test_get_cached_image(False, True) + + def test_get_cached_image_not_existent_and_path_exists(self): + self._test_get_cached_image(False, True, True) diff --git a/vix/tests/test_disk_manager.py b/vix/tests/test_disk_manager.py new file mode 100644 index 0000000..1fe06bc --- /dev/null +++ b/vix/tests/test_disk_manager.py @@ -0,0 +1,274 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2013 Cloudbase Solutions Srl +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import os +import unittest +import subprocess +import sys + +if sys.platform == 'win32': + import _winreg + import win32api + +from vix import disk_manager +from vix import vixutils +from vix import utils + +DISK_TYPE_VMDK = "vmdk" +DISK_TYPE_VHD = "vpc" +DISK_TYPE_QCOW2 = "qcow2" +DISK_TYPE_RAW = "raw" + + +class VixUtilsTestCase(unittest.TestCase): + """Unit tests for utility class""" + + def setUp(self): + self._disk_manager = disk_manager.DiskManager() + + def _test_get_vdisk_man_path(self, platform_win32=False): + fake_vdisk_man_path = 'fake/path' + fake_vix_result = mock.MagicMock() + os.path.join = mock.MagicMock() + vixutils.get_vix_bin_path = mock.MagicMock() + os.path.join.return_value = fake_vdisk_man_path + vixutils.get_vix_bin_path.return_value = fake_vix_result + sys.platform = 'not_win32' + if platform_win32: + sys.platform = 'win32' + fake_vdisk_man_path += ".exe" + + response = self._disk_manager._get_vdisk_man_path() + vixutils.get_vix_bin_path.assert_called_once() + os.path.join.assert_called_with(fake_vix_result, 'vmware-vdiskmanager') + self.assertEqual(response, fake_vdisk_man_path) + + def test_get_vdisk_man_path(self): + self._test_get_vdisk_man_path() + + def test_get_vdisk_man_path_win32(self): + self._test_get_vdisk_man_path(True) + + def test_check_vdisk_man_exists(self): + os.path.join = mock.MagicMock() + vixutils.get_vix_bin_path = mock.MagicMock() + fake_check_vdisk_man_exists = mock.MagicMock() + self._disk_manager._get_vdisk_man_path = mock.MagicMock() + fake_vdisk_man = mock.MagicMock() + self._disk_manager._get_vdisk_man_path.return_value = fake_vdisk_man + os.path.exists = mock.MagicMock() + os.path.exists.return_value = fake_check_vdisk_man_exists + + response = self._disk_manager._check_vdisk_man_exists() + self.assertEqual(response, fake_check_vdisk_man_exists) + self._disk_manager._get_vdisk_man_path.assert_called_once() + os.path.exists.assert_called_with(fake_vdisk_man) + + def test_create_disk_vdisk_man(self): + self._disk_manager._get_vdisk_man_path = mock.MagicMock() + fake_vdisk_man = 'fake/path' + self._disk_manager._get_vdisk_man_path.return_value = fake_vdisk_man + fake_size_mb = "1" + fake_disk_path = "fake/disk/path" + fake_args = [fake_vdisk_man, "-c", "-s", "%sMB" % fake_size_mb, + "-a", "lsilogic", "-t", "0", fake_disk_path] + self._disk_manager._exec_cmd = mock.MagicMock() + + self._disk_manager._create_disk_vdisk_man(fake_disk_path, fake_size_mb) + self._disk_manager._get_vdisk_man_path.assert_called_once() + self._disk_manager._exec_cmd.assert_called_with(fake_args) + + def _test_exec_cmd(self, exception=False): + fake_args = 'fake args' + fake_process = mock.MagicMock() + subprocess.Popen = mock.MagicMock() + subprocess.Popen.return_value = fake_process + fake_process.communicate = mock.MagicMock() + fake_out = mock.MagicMock() + fake_error = mock.MagicMock() + fake_process.communicate = mock.MagicMock() + fake_process.communicate.return_value = (fake_out, fake_error) + fake_process.returncode = exception + + if exception: + self.assertRaises(utils.VixException, self._disk_manager._exec_cmd, + fake_args) + else: + response = self._disk_manager._exec_cmd(fake_args) + self.assertEqual(response, (fake_out, fake_error)) + + fake_process.Popen.assert_called_once() + fake_process.communicate.assert_called_once() + + def test_exec_cmd(self): + self._test_exec_cmd() + + def test_exec_cmd_exception(self): + self._test_exec_cmd(True) + + def _test_get_disk_info(self, exception=False): + fake_disk_path = "disk\path" + fake_args = ["qemu-img", "info", fake_disk_path] + fake_format = 'qcow2' + fake_internal_size = 21474836480 + fake_out = mock.MagicMock() + fake_err = mock.MagicMock() + self._disk_manager._exec_cmd = mock.MagicMock() + self._disk_manager._exec_cmd.return_value = fake_out, fake_err + fake_file_size = mock.MagicMock() + fake_out.split = mock.MagicMock() + #is it ok if I put return values or should I mock re.match? + fake_out.split.return_value = [ + 'file format: qcow2', 'virtual size: 20G (21474836480 bytes)'] + os.path.getsize = mock.MagicMock() + os.path.getsize.return_value = fake_file_size + + if exception: + fake_out.split.return_value = ['file format: qcow2'] + self.assertRaises(utils.VixException, + self._disk_manager.get_disk_info, + fake_disk_path) + else: + response = self._disk_manager.get_disk_info(fake_disk_path) + self.assertEqual(response, (fake_format, fake_internal_size, + fake_file_size)) + os.path.getsize.assert_called_with(fake_disk_path) + + self._disk_manager._exec_cmd.assert_called_with(fake_args) + fake_out.split.assert_called_with(os.linesep) + + def test_get_disk_info(self): + self._test_get_disk_info() + + def test_get_disk_info_exception(self): + self._test_get_disk_info(exception=True) + + def test_create_disk_qemu(self): + fake_disk_type = "disk type" + fake_disk_path = "disk_path" + fake_size_mb = "1" + fake_args = ["qemu-img", "create", "-f", fake_disk_type, + fake_disk_path, "%sM" % fake_size_mb] + self._disk_manager._exec_cmd = mock.MagicMock() + + self._disk_manager._create_disk_qemu(fake_disk_path, fake_size_mb, + fake_disk_type) + self._disk_manager._exec_cmd.assert_called_with(fake_args) + + def _test_create_disk(self, disk_exists=True): + fake_disk_type = DISK_TYPE_VMDK + fake_disk_path = "disk_path" + fake_size_mb = "1" + self._disk_manager._check_vdisk_man_exists = mock.MagicMock() + self._disk_manager._check_vdisk_man_exists.return_value = disk_exists + if disk_exists: + self._disk_manager._create_disk_vdisk_man = mock.MagicMock() + self._disk_manager.create_disk(fake_disk_path, fake_size_mb, + fake_disk_type) + self._disk_manager._create_disk_vdisk_man.assert_called_with( + fake_disk_path, fake_size_mb) + else: + self._disk_manager._create_disk_qemu = mock.MagicMock() + self._disk_manager.create_disk(fake_disk_path, fake_size_mb, + fake_disk_type) + self._disk_manager._create_disk_qemu.assert_called_with( + fake_disk_path, fake_size_mb, fake_disk_type) + + def test_create_disk_with_vdisk_man(self): + self._test_create_disk() + + def test_create_disk_with_qemu(self): + self._test_create_disk() + + def test_resize_disk_vdisk_man(self): + fake_disk_path = "disk_path" + fake_new_size_mb = "1" + fake_vdisk_man = "disk_path_man" + self._disk_manager._get_vdisk_man_path = mock.MagicMock() + self._disk_manager._get_vdisk_man_path.return_value = fake_vdisk_man + fake_args = [fake_vdisk_man, "-x", "%sMB" % fake_new_size_mb, + fake_disk_path] + self._disk_manager._exec_cmd = mock.MagicMock() + + self._disk_manager._resize_disk_vdisk_man(fake_disk_path, + fake_new_size_mb) + self._disk_manager._exec_cmd.assert_called_with(fake_args) + self._disk_manager._get_vdisk_man_path.assert_called_once() + + def _test_resize_disk_vdisk_qemu(self, path_exists=True, exception=False): + fake_new_disk_type = DISK_TYPE_VMDK + fake_disk_path = "disk_path" + fake_new_size_mb = "1" + fake_tmp_disk_path = "%s.raw" % fake_disk_path + os.path.exists = mock.MagicMock() + os.path.exists.return_value = path_exists + os.remove = mock.MagicMock() + + fake_args_3 = ["qemu-img", "convert", "-f", DISK_TYPE_RAW, "-O", + fake_new_disk_type, fake_tmp_disk_path, fake_disk_path] + self._disk_manager._exec_cmd = mock.MagicMock() + + if exception: + self._disk_manager._exec_cmd.side_effect = Exception + self.assertRaises(Exception, self._disk_manager._resize_disk_qemu, + fake_disk_path, fake_new_size_mb, + fake_new_disk_type) + else: + self._disk_manager._resize_disk_qemu(fake_disk_path, + fake_new_size_mb, + fake_new_disk_type) + #How can I test more than one call to the same method with params? + self._disk_manager._exec_cmd.assert_called_with(fake_args_3) + + os.path.exists.assert_called_with(fake_tmp_disk_path) + if path_exists: + os.remove.assert_called_with(fake_tmp_disk_path) + + def test_resize_disk_vdisk_qemu_path_exists(self): + self._test_resize_disk_vdisk_qemu() + + def test_resize_disk_vdisk_qemu_path_does_not_exist(self): + self._test_resize_disk_vdisk_qemu(False) + + def test_resize_disk_vdisk_qemu_path_exists_with_exception(self): + self._test_resize_disk_vdisk_qemu(exception=True) + + def _test_resize_disk(self, disk_exists=True): + fake_new_disk_type = DISK_TYPE_VMDK + fake_disk_path = "disk_path" + fake_new_size_mb = "1" + self._disk_manager._check_vdisk_man_exists = mock.MagicMock() + self._disk_manager._check_vdisk_man_exists.return_value = disk_exists + if disk_exists: + self._disk_manager._resize_disk_vdisk_man = mock.MagicMock() + self._disk_manager.resize_disk(fake_disk_path, fake_new_size_mb, + fake_new_disk_type) + self._disk_manager._resize_disk_vdisk_man.assert_called_with( + fake_disk_path, fake_new_size_mb) + else: + self._disk_manager._resize_disk_qemu = mock.MagicMock() + self._disk_manager.resize_disk(fake_disk_path, fake_new_size_mb, + fake_new_disk_type) + self._disk_manager._resize_disk_qemu.assert_called_with( + fake_disk_path, fake_new_size_mb, fake_new_disk_type) + + def test_resize_disk(self): + self._test_resize_disk() + + def test_resize_disku(self): + self._test_resize_disk() diff --git a/vix/tests/test_vixutils.py b/vix/tests/test_vixutils.py new file mode 100644 index 0000000..9c58640 --- /dev/null +++ b/vix/tests/test_vixutils.py @@ -0,0 +1,230 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2013 Cloudbase Solutions Srl +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import os +import re +import unittest +import sys + +if sys.platform == 'win32': + import _winreg + import win32api + +from vix import utils +from vix import vixutils +from vix import vixlib + + +class VixUtilsTestCase(unittest.TestCase): + """Unit tests for utility class""" + + def test_check_job_err_code(self): + fake_err = 1 + vixlib.Vix_GetErrorText = mock.MagicMock() + + self.assertRaises(utils.VixException, vixutils._check_job_err_code, + fake_err) + + def test_load_config_file_values(self): + fake_path = 'fake/path' + match_mock = mock.MagicMock() + + re.match = mock.MagicMock() + re.match.return_value = match_mock + + with mock.patch('vix.vixutils.open', + mock.mock_open(read_data='fake data'), + create=True) as m: + response = vixutils.load_config_file_values(fake_path) + print m.mock_calls + m.assert_called_with('fake/path', 'rb') + m().readlines.assert_called_once() + + self.assertTrue(response is not None) + + def _test_get_player_preferences_file_path(self, platform): + sys.platform = platform + os.getenv = mock.MagicMock() + os.getenv.return_value = 'APPDATA' + os.path.join = mock.MagicMock() + os.path.join.return_value = 'APPDATA/VMWare/preferences.ini' + os.path.expanduser = mock.MagicMock() + os.path.expanduser.return_value = "~/.vmware/preferences" + + response = vixutils._get_player_preferences_file_path() + + if platform == "win32": + os.getenv.assert_called_with('APPDATA') + os.path.join.assert_called_with('APPDATA', "VMWare", + "preferences.ini") + self.assertEqual(response, 'APPDATA/VMWare/preferences.ini') + else: + os.path.expanduser.assert_called_with("~/.vmware/preferences") + self.assertEqual(response, "~/.vmware/preferences") + + def test_get_player_preferences_file_path_win(self): + self._test_get_player_preferences_file_path('win32') + + def test_get_player_preferences_file_path_darwin(self): + self._test_get_player_preferences_file_path('linux') + + def _test_get_install_dir(self, platform): + fake_key = mock.MagicMock() + fake_query_response = mock.MagicMock() + + sys.platform = platform + _winreg.OpenKey = mock.MagicMock() + _winreg.OpenKey.return_value = fake_key + _winreg.QueryValueEx = mock.MagicMock() + _winreg.QueryValueEx.return_value = fake_query_response + + if platform == "darwin": + response = vixutils._get_install_dir() + self.assertEqual(response, "/Applications/VMware Fusion.app") + elif platform == "win32": + response = vixutils._get_install_dir() + _winreg.OpenKey.assert_called_with( + _winreg.HKEY_LOCAL_MACHINE, "SOFTWARE\VMware, Inc.\VMware " + "Workstation") + _winreg.QueryValueEx.assert_called_with(fake_key.__enter__(), + "InstallPath") + self.assertEqual(response, fake_query_response[0]) + else: + self.assertRaises(NotImplementedError, vixutils._get_install_dir) + + def test_get_install_dir_win(self): + self._test_get_install_dir(platform="win32") + + def test_get_install_dir_darwin(self): + self._test_get_install_dir(platform="darwin") + + def test_get_install_dir_other(self): + self._test_get_install_dir(platform=None) + + def _test_get_vix_bin_path(self, platform): + sys.platform = platform + + vixutils._get_install_dir = mock.MagicMock() + vixutils._get_install_dir.return_value = 'fake/path/' + os.path.join = mock.MagicMock() + os.path.join.return_value = 'fake/path/' + 'Contents/Library' + + if platform == "darwin": + response = vixutils.get_vix_bin_path() + self.assertEqual(response, 'fake/path/' + 'Contents/Library') + elif platform == "win32": + response = vixutils.get_vix_bin_path() + os.path.join.assert_called_once() + self.assertEqual(response, 'fake/path/') + else: + response = vixutils.get_vix_bin_path() + self.assertEqual(response, "/usr/bin") + + def test_get_vix_bin_path_win(self): + self._test_get_vix_bin_path('win32') + + def test_get_vix_bin_path_darwin(self): + self._test_get_vix_bin_path('darwin') + + def test_get_vix_bin_path_other(self): + self._test_get_vix_bin_path('linux') + + def test_remove_vmx_value(self): + fake_path = 'fake/path' + fake_name = 'fake_name' + utils.remove_lines = mock.MagicMock() + vixutils.remove_vmx_value(fake_path, fake_name) + utils.remove_lines.assert_called_with(fake_path, + r"^%s\s*=\s*.*$" % fake_name) + + def test_set_vmx_value(self): + fake_path = 'fake/path' + fake_name = 'fake_name' + fake_value = 'fake_value' + utils.replace_text = mock.MagicMock() + utils.replace_text.return_value = False + with mock.patch('vix.vixutils.open', mock.mock_open(), + create=True) as m: + + vixutils.set_vmx_value(fake_path, fake_name, fake_value) + + m.assert_called_with(fake_path, "ab") + m().write.assert_called_with( + "%(name)s = \"%(value)s\"" % + {'name': fake_name, 'value': fake_value} + os.linesep) + utils.replace_text.assert_called_with( + fake_path, r"^(%s\s*=\s*)(.*)$" % fake_name, + "\\1\"%s\"" % fake_value) + + def test_get_vmx_value(self): + fake_path = 'fake/path' + fake_name = 'fake_name' + pattern = r"^%s\s*=\s*\"(.*)\"$" % fake_name + fake_value = mock.MagicMock() + + utils.get_text = mock.MagicMock() + utils.get_text.return_value = fake_value + response = vixutils.get_vmx_value(fake_path, fake_name) + utils.get_text.assert_called_with(fake_path, + pattern) + self.assertEqual(response, fake_value[0]) + + def _test_get_vix_host_type(self, platform, path_exists=False, + fake_key=None, product_name=None): + fake_value = mock.MagicMock() + sys.platform = platform + + os.path.exists = mock.MagicMock() + os.path.exists.return_value = path_exists + + _winreg.EnumValue = mock.MagicMock() + _winreg.EnumValue.return_value = fake_value + _winreg.QueryValueEx = mock.MagicMock() + _winreg.QueryValueEx.return_value = product_name + + if platform == 'darwin' and path_exists: + response = vixutils.get_vix_host_type() + os.path.exists.assert_called_with( + "/Applications/VMware Fusion.app") + self.assertEqual(response, + vixlib.VIX_SERVICEPROVIDER_VMWARE_WORKSTATION) + + elif platform == 'win32 'and fake_key is not None: + response = vixutils.get_vix_host_type() + _winreg.OpenKey = mock.MagicMock(return_value=fake_key) + print response + self.assertEqual(_winreg.OpenKey.call_count, 2) + _winreg.EnumValue.assert_called_with(fake_key, 0) + _winreg.QueryValueEx.assert_called_with(fake_key, "ProductName") + self.assertEqual(response, product_name) + + elif platform == 'linux' and product_name == "VMware Player": + response = vixutils.get_vix_host_type() + print response + self.assertEqual(response, 3) + + def test_get_vix_host_darwin_path_exists(self): + self._test_get_vix_host_type('darwin', path_exists=True) + + def test_get_vix_host_win(self): + fake_key = mock.MagicMock() + self._test_get_vix_host_type('win32', fake_key=fake_key, + product_name="VMware Player") + + def test_get_vix_host_other_VMware_player(self): + self._test_get_vix_host_type('linux', product_name="VMware Player") diff --git a/vix/tests/test_vixutils_classes.py b/vix/tests/test_vixutils_classes.py new file mode 100644 index 0000000..4a4d85f --- /dev/null +++ b/vix/tests/test_vixutils_classes.py @@ -0,0 +1,979 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2013 Cloudbase Solutions Srl +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import ctypes +import mock +import os +import re +import unittest +import shutil +import sys +import time + +if sys.platform == 'win32': + import _winreg + import win32api + +from vix import vixutils +from vix import vixlib + + +class VixUtilsTestCase(unittest.TestCase): + """Unit tests for utility class""" + + def setUp(self): + ctypes = mock.MagicMock() + self.ctypes_handle = mock.MagicMock() + ctypes.c_int = mock.MagicMock(return_value=self.ctypes_handle) + self._VixVM = vixutils.VixVM(self.ctypes_handle) + self._VixSnapshot = vixutils.VixSnapshot(self.ctypes_handle) + self._VixConnection = vixutils.VixConnection() + + ########### TESTING VixVM CLASS ########### + def test_close_VixVM(self): + vixlib.Vix_ReleaseHandle = mock.MagicMock() + + self._VixVM.close() + + vixlib.Vix_ReleaseHandle.assert_called_once() + self.assertIsNone(self._VixVM._vm_handle) + + def test_get_power_state(self): + fake_power_state = mock.MagicMock() + ctypes_mock = mock.Mock() + ctypes.c_int.return_value = fake_power_state + vixlib.Vix_GetProperties = mock.MagicMock() + vixlib.Vix_GetProperties.return_value = None + vixutils._check_job_err_code = mock.MagicMock() + ctypes.byref.return_value = ctypes_mock + + self._VixVM.get_power_state() + + vixlib.Vix_GetProperties.assert_called_with( + self._VixVM._vm_handle, vixlib.VIX_PROPERTY_VM_POWER_STATE, + ctypes_mock, vixlib.VIX_PROPERTY_NONE) + ctypes.byref.assert_called_once() + vixutils._check_job_err_code.assert_called_once_with(None) + + def _test_power_on(self, show_gui): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_PowerOn = mock.MagicMock() + vixlib.VixVM_PowerOn.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.power_on(show_gui) + if show_gui: + options = vixlib.VIX_VMPOWEROP_LAUNCH_GUI + else: + options = vixlib.VIX_VMPOWEROP_NORMAL + vixlib.VixVM_PowerOn.assert_called_with(self._VixVM._vm_handle, + options, + vixlib.VIX_INVALID_HANDLE, + None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def test_power_on_with_gui(self): + self._test_power_on(True) + + def test_power_on_without_gui(self): + self._test_power_on(False) + + def test_pause(self): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_Pause = mock.MagicMock() + vixlib.VixVM_Pause.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.pause() + + vixlib.VixVM_Pause.assert_called_with(self._VixVM._vm_handle, 0, + vixlib.VIX_INVALID_HANDLE, + None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def test_unpause(self): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_Unpause = mock.MagicMock() + vixlib.VixVM_Unpause.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.unpause() + + vixlib.VixVM_Unpause.assert_called_with(self._VixVM._vm_handle, 0, + vixlib.VIX_INVALID_HANDLE, + None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def test_suspend(self): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_Suspend = mock.MagicMock() + vixlib.VixVM_Suspend.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.suspend() + + vixlib.VixVM_Suspend.assert_called_with(self._VixVM._vm_handle, 0, + None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def _test_reboot(self, soft): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_Reset = mock.MagicMock() + vixlib.VixVM_Reset.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.reboot(soft) + + if soft: + power_op = vixlib.VIX_VMPOWEROP_FROM_GUEST + else: + power_op = vixlib.VIX_VMPOWEROP_NORMAL + + vixlib.VixVM_Reset.assert_called_with(self._VixVM._vm_handle, + power_op, + None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def test_reboot_soft(self): + self._test_reboot(True) + + def test_reboot_hard(self): + self._test_reboot(False) + + def _test_power_off(self, soft): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_PowerOff = mock.MagicMock() + vixlib.VixVM_PowerOff.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.power_off(soft) + + if soft: + power_op = vixlib.VIX_VMPOWEROP_FROM_GUEST + else: + power_op = vixlib.VIX_VMPOWEROP_NORMAL + + vixlib.VixVM_PowerOff.assert_called_with(self._VixVM._vm_handle, + power_op, None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def test_power_off_soft(self): + self._test_reboot(True) + + def test_power_off_hard(self): + self._test_reboot(False) + + def test_wait_for_tools_in_guest(self): + timeout_seconds = 99999 + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_WaitForToolsInGuest = mock.MagicMock() + vixlib.VixVM_WaitForToolsInGuest.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.wait_for_tools_in_guest(timeout_seconds) + + vixlib.VixVM_WaitForToolsInGuest.assert_called_with( + self._VixVM._vm_handle, timeout_seconds, None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + + def _test_get_guest_ip_address(self): + #1)ALWAYS time.sleep(3) + + #2)cannot mock time.time() + #need something like testfixtures or another repository + + fake_job_handle = mock.MagicMock() + read_value = mock.MagicMock() + ctypes_byref_mock = mock.MagicMock() + fake_ip = '10.10.10.10' + + vixlib.VixVM_WaitForToolsInGuest = mock.MagicMock() + vixlib.VixVM_ReadVariable = mock.MagicMock() + vixlib.VixVM_ReadVariable.return_value = fake_job_handle + ctypes.c_char_p = mock.MagicMock() + ctypes.c_char_p.return_value = read_value + read_value.value = mock.MagicMock() + read_value.value.return_value = fake_ip + ctypes.byref = mock.MagicMock() + ctypes.byref.return_value = ctypes_byref_mock + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + vixlib.Vix_FreeBuffer = mock.MagicMock() + + response = self._VixVM.get_guest_ip_address() + + vixlib.Vix_FreeBuffer.assert_called_with(read_value) + vixlib.VixVM_WaitForToolsInGuest.assert_called_with( + self._VixVM._vm_handle, 600, None, None) + time.time.assert_called_once() + vixlib.VixVM_ReadVariable.assert_called_with( + self._VixVM._vm_handle, vixlib.VIX_VM_GUEST_VARIABLE, "ip", 0, + None, None) + vixlib.VixJob_Wait.assert_called_with( + fake_job_handle, + vixlib.VIX_PROPERTY_JOB_RESULT_VM_VARIABLE_STRING, + ctypes_byref_mock, vixlib.VIX_PROPERTY_NONE) + ctypes.byref.assert_called_with(read_value) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_with(None) + self.assertEqual(response, read_value.value) + + def _test_delete(self, delete_disk_files): + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_Delete = mock.MagicMock() + vixlib.VixVM_Delete.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.delete(delete_disk_files) + + if delete_disk_files: + delete_options = vixlib.VIX_VMDELETE_DISK_FILES + else: + delete_options = 0 + + vixlib.VixVM_Delete.assert_called_with(self.ctypes_handle, + delete_options, None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + self.assertEqual(vixlib.Vix_ReleaseHandle.call_count, 2) + vixutils._check_job_err_code.assert_called_once_with(None) + + def test_delete_disk_files_True(self): + self._test_delete(delete_disk_files=True) + + def test_delete_disk_files_False(self): + self._test_delete(delete_disk_files=False) + + def _test_create_snapshot(self, include_memory): + fake_job_handle = mock.MagicMock() + fake_name = 'fake name' + fake_description = 'fake description' + fake_snapshot_handle = mock.MagicMock() + ctypes_byref_mock = mock.MagicMock() + + vixlib.VixVM_CreateSnapshot = mock.MagicMock() + vixlib.VixVM_CreateSnapshot.return_value = fake_job_handle + vixlib.VixHandle = mock.MagicMock() + vixlib.VixHandle.return_value = fake_snapshot_handle + ctypes.byref = mock.MagicMock() + ctypes.byref.return_value = ctypes_byref_mock + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.create_snapshot(include_memory=include_memory, + name=fake_name, + description=fake_description) + if include_memory: + options = vixlib.VIX_SNAPSHOT_INCLUDE_MEMORY + else: + options = 0 + + vixlib.VixVM_CreateSnapshot.assert_called_with( + self._VixVM._vm_handle, fake_name, fake_description, options, + vixlib.VIX_INVALID_HANDLE, None, None) + vixlib.VixHandle.assert_called_once() + ctypes.byref.assert_called_with(fake_snapshot_handle) + vixlib.VixJob_Wait.assert_called_with( + fake_job_handle, vixlib.VIX_PROPERTY_JOB_RESULT_HANDLE, + ctypes_byref_mock, vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_with(None) + + def test_create_snapshot_include_memory_true(self): + self._test_create_snapshot(include_memory=True) + + def test_create_snapshot_include_memory_false(self): + self._test_create_snapshot(include_memory=True) + + def test_remove_snapshot(self): + fake_snapshot = mock.MagicMock() + fake_job_handle = mock.MagicMock() + + vixlib.VixVM_RemoveSnapshot = mock.MagicMock() + vixlib.VixVM_RemoveSnapshot.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixutils._check_job_err_code = mock.MagicMock() + + self._VixVM.remove_snapshot(fake_snapshot) + + vixlib.VixVM_RemoveSnapshot.assert_called_with( + self._VixVM._vm_handle, fake_snapshot._snapshot_handle, 0, None, + None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixutils._check_job_err_code.assert_called_once_with(None) + fake_snapshot.close.assert_called_once() + + def test_get_vmx_path(self): + fake_vmx_path = mock.MagicMock() + mock_ctypes_byref = mock.MagicMock() + + ctypes.c_char_p = mock.MagicMock() + ctypes.c_char_p.return_value = fake_vmx_path + ctypes.byref = mock.MagicMock() + ctypes.byref.return_value = mock_ctypes_byref + vixlib.Vix_GetProperties = mock.MagicMock() + vixlib.Vix_GetProperties.return_value = None + vixutils._check_job_err_code = mock.MagicMock() + vixlib.Vix_FreeBuffer = mock.MagicMock() + + response = self._VixVM.get_vmx_path() + + ctypes.c_char_p.assert_called_once() + vixlib.Vix_GetProperties.assert_called_with( + self._VixVM._vm_handle, vixlib.VIX_PROPERTY_VM_VMX_PATHNAME, + mock_ctypes_byref, vixlib.VIX_PROPERTY_NONE) + vixutils._check_job_err_code.assert_called_with(None) + vixlib.Vix_FreeBuffer.assert_called_with(fake_vmx_path) + self.assertTrue(response is not None) + + def test_get_vnc_settings(self): + vixlib.Vix_GetProperties = mock.MagicMock() + vixutils.get_vmx_value = mock.MagicMock() + vixutils.get_vmx_value.side_effect = ['True', '9999'] + + response = self._VixVM.get_vnc_settings() + + self.assertEqual(response, (True, 9999)) + + ########### TESTING VixSnapshot CLASS ########### + def test_close_VixSnapshot(self): + vixlib.Vix_ReleaseHandle = mock.MagicMock() + self._VixSnapshot.close() + vixlib.Vix_ReleaseHandle.assert_called_once() + self.assertIsNone(self._VixSnapshot._snapshot_handle) + + ########### TESTING VixConnection CLASS ########### + def _test_unregister_vm_and_delete_files(self, destroy_disks): + fake_path = 'fake/path' + + mock_vm = mock.MagicMock() + self._VixConnection.open_vm = mock.MagicMock() + self._VixConnection.open_vm.return_value = mock_vm + + mock_vm.get_power_state = mock.MagicMock( + return_value=vixlib.VIX_POWERSTATE_POWERED_OFF) + vixutils.VixConnection.unregister_vm = mock.MagicMock() + vixutils.VixConnection.delete_vm_files = mock.MagicMock() + + self._VixConnection.unregister_vm_and_delete_files( + fake_path, destroy_disks=destroy_disks) + + self._VixConnection.unregister_vm.assert_called_with(fake_path) + if destroy_disks: + self._VixConnection.delete_vm_files.assert_called_with( + fake_path) + + def test_unregister_vm_and_delete_files_destroy_disks(self): + self._test_unregister_vm_and_delete_files(destroy_disks=True) + + def test_unregister_vm_and_delete_files_no_destroy_disks(self): + self._test_unregister_vm_and_delete_files(destroy_disks=False) + + def test_connect(self): + job_handle = mock.MagicMock() + host_handle = mock.MagicMock() + + vixlib.VixHost_Connect = mock.MagicMock() + vixlib.VixHost_Connect.return_value = job_handle + vixlib.VixHandle = mock.MagicMock() + vixlib.VixHandle.return_value = host_handle + + vixutils.get_vix_host_type = mock.MagicMock() + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixConnection.connect() + vixlib.VixHost_Connect.assert_called_with(vixlib.VIX_API_VERSION, + vixutils.get_vix_host_type(), + None, 0, None, None, 0, + vixlib.VIX_INVALID_HANDLE, + None, None) + vixlib.Vix_ReleaseHandle.assert_called_with(job_handle) + vixutils._check_job_err_code.assert_called_with(None) + + self.assertEqual(self._VixConnection._host_handle, host_handle) + + def test_open_vm(self): + fake_path = 'fake/path' + mock_job_handle = mock.MagicMock() + mock_vm_handle = mock.MagicMock() + + vixlib.VixVM_Open = mock.MagicMock() + vixlib.VixVM_Open.return_value = mock_job_handle + vixlib.VixHandle = mock.MagicMock() + vixlib.VixHandle.return_value = mock_vm_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + #TODO: with side effect for getting and error than continue + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + response = self._VixConnection.open_vm(fake_path) + + vixlib.VixVM_Open.assert_called_with( + self._VixConnection._host_handle, fake_path, None, None) + vixlib.VixJob_Wait.assert_called_with( + mock_job_handle, vixlib.VIX_PROPERTY_JOB_RESULT_HANDLE, + ctypes.byref(mock_vm_handle), vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(mock_job_handle) + vixutils._check_job_err_code.assert_called_with(None) + self.assertIsInstance(response, vixutils.VixVM) + + def test_create_vm(self): + fake_path = 'fake/path' + display_name = 'fake_name' + guest_os = 'guest_os' + disk_paths = ['fake/disk/path'] + iso_paths = ['fake/iso/path'] + floppy_path = 'fake/floppy/path' + networks = [('eth', 'mac')] + nested_hypervisor = True + vnc_enabled = True + vnc_port = 9999 + + self._VixConnection._get_scsi_config = mock.MagicMock() + self._VixConnection._get_ide_config = mock.MagicMock() + self._VixConnection._get_networks_config = mock.MagicMock() + self._VixConnection._get_nested_hypervisor_config = mock.MagicMock() + self._VixConnection._get_vnc_config = mock.MagicMock() + os.path.dirname = mock.MagicMock() + os.path.dirname.return_value = 'fake_dir' + os.path.exists = mock.MagicMock() + os.path.exists.return_value = False + os.makedirs = mock.MagicMock() + os.linesep = mock.MagicMock() + + with mock.patch('vix.vixutils.open', mock.mock_open(), + create=True) as m: + self._VixConnection.create_vm(vmx_path=fake_path, + display_name=display_name, + guest_os=guest_os, + disk_paths=disk_paths, + iso_paths=iso_paths, + floppy_path=floppy_path, + networks=networks, + nested_hypervisor=nested_hypervisor, + vnc_enabled=vnc_enabled, + vnc_port=vnc_port) + m.assert_called_with('fake/path', 'wb') + + self._VixConnection._get_scsi_config.assert_called_with(disk_paths) + self._VixConnection._get_ide_config.assert_called_with(iso_paths) + self._VixConnection._get_networks_config.assert_called_with(networks) + self._VixConnection._get_nested_hypervisor_config.assert_called_once() + self._VixConnection._get_vnc_config.assert_called_with(vnc_enabled, + vnc_port) + os.path.dirname.assert_called_with(fake_path) + os.path.exists.assert_called_with('fake_dir') + os.makedirs.assert_called_with('fake_dir') + + def test_update_vm(self): + fake_path = 'fake/path' + display_name = 'fake_name' + guest_os = 'guest_os' + virtual_hw_version = 10 + num_vcpus = 1 + cores_per_socket = 1 + mem_size_mb = 1024 + disk_paths = ['fake/disk/path'] + iso_paths = ['fake/iso/path'] + floppy_path = 'fake/floppy/path' + networks = [('eth', 'mac')] + boot_order = "hdd,cdrom,floppy" + nested_hypervisor = True + vnc_enabled = True + vnc_port = mock.MagicMock() + additional_config = {"fake_config": "fake_value"} + + self._VixConnection._get_scsi_config = mock.MagicMock() + self._VixConnection._get_scsi_config.return_value = { + "fake disk": "fake path"} + self._VixConnection._get_ide_config = mock.MagicMock() + self._VixConnection._get_ide_config.return_value = { + "fake iso": "fake path"} + self._VixConnection._get_floppy_config = mock.MagicMock() + self._VixConnection._get_floppy_config.return_value = { + "fake floppy": "fake path"} + self._VixConnection._get_nested_hypervisor_config = mock.MagicMock() + self._VixConnection._get_nested_hypervisor_config.return_value = { + "fake hypervisor": "fake value"} + self._VixConnection._get_networks_config = mock.MagicMock() + self._VixConnection._get_networks_config.return_value = { + "fake network": "fake mac"} + self._VixConnection._get_vnc_config = mock.MagicMock() + self._VixConnection._get_vnc_config.return_value = {"enabled": True, + "port": 9999, + } + + vixutils.remove_vmx_value = mock.MagicMock() + vixutils.set_vmx_value = mock.MagicMock() + + self._VixConnection.update_vm( + vmx_path=fake_path, display_name=display_name, + guest_os=guest_os, virtual_hw_version=virtual_hw_version, + num_vcpus=num_vcpus, cores_per_socket=cores_per_socket, + mem_size_mb=mem_size_mb, disk_paths=disk_paths, + iso_paths=iso_paths, floppy_path=floppy_path, networks=networks, + boot_order=boot_order, nested_hypervisor=nested_hypervisor, + vnc_enabled=vnc_enabled, vnc_port=vnc_port, + additional_config=additional_config) + + self._VixConnection._get_scsi_config.assert_called_with(disk_paths) + self._VixConnection._get_floppy_config.assert_called_with(floppy_path) + self._VixConnection._get_ide_config.assert_called_with(iso_paths) + self._VixConnection._get_networks_config.assert_called_with(networks) + self._VixConnection._get_nested_hypervisor_config.assert_called_once() + self._VixConnection._get_vnc_config.assert_called_with(vnc_enabled, + vnc_port) + + vixutils.remove_vmx_value.assert_called_with( + fake_path, r"ethernet[\d]+\.[a-zA-Z]+") + self.assertEqual(vixutils.set_vmx_value.call_count, 15) + + def test_get_vnc_config(self): + vnc_enabled = True + vnc_port = 9999 + + response = self._VixConnection._get_vnc_config(vnc_enabled, vnc_port) + + self.assertEqual(response, {'RemoteDisplay.vnc.enabled': True, + 'RemoteDisplay.vnc.port': 9999}) + + def test_get_scsi_config(self): + disk_paths = ['fake/disk/path'] + + response = self._VixConnection._get_scsi_config(disk_paths) + + self.assertEqual(response, {'scsi0:0.present': 'TRUE', + 'scsi0.sharedBus': 'none', + 'scsi0:0.fileName': 'fake/disk/path', + 'scsi0.present': 'TRUE', + 'scsi0:0.deviceType': 'scsi-hardDisk', + 'scsi0.virtualDev': 'lsisas1068'}) + + def test_get_scsi_disk_config(self): + ctrl_idx = 9999 + disk_idx = 9999 + path = 'fake/path' + + response = self._VixConnection._get_scsi_disk_config(ctrl_idx, + disk_idx, path) + + self.assertEqual(response, + {'scsi9999:9999.present': 'TRUE', + 'scsi9999:9999.fileName': 'fake/path', + 'scsi9999:9999.deviceType': 'scsi-hardDisk'}) + + def test_get_ide_config(self): + iso_paths = ['fake/iso/path'] + + response = self._VixConnection._get_ide_config(iso_paths) + + self.assertEqual(response, {'ide1:0.deviceType': 'cdrom-image', + 'ide1:0.present': 'TRUE', + 'ide1:0.clientDevice': 'FALSE', + 'ide1:0.startConnected': True, + 'ide1:0.fileName': 'fake/iso/path'}) + + def test_get_ide_iso_config(self): + ctrl_idx = 9999 + disk_idx = 9999 + fake_path = 'fake/path' + + response = self._VixConnection._get_ide_iso_config(ctrl_idx, + disk_idx, + fake_path) + + self.assertEqual(response, {'ide9999:9999.fileName': 'fake/path', + 'ide9999:9999.deviceType': 'cdrom-image', + 'ide9999:9999.startConnected': True, + 'ide9999:9999.present': 'TRUE', + 'ide9999:9999.clientDevice': 'FALSE'}) + + def test_get_floppy_config(self): + floppy_path = 'fake/floppy/path' + + response = self._VixConnection._get_floppy_config(floppy_path) + + self.assertEqual(response, {'floppy0.fileType': 'file', + 'floppy0.clientDevice': 'FALSE', + 'floppy0.present': 'TRUE', + 'floppy0.fileName': 'fake/floppy/path'}) + + def test_get_nested_hypervisor_config(self): + response = self._VixConnection._get_nested_hypervisor_config() + self.assertEqual(response, {'vhv.enable': 'TRUE', + 'vcpu.hotadd': 'FALSE', + 'featMask.vm.hv.capable': 'Min:1'}) + + def test_get_networks_config(self): + networks = [('eth', 'mac')] + + response = self._VixConnection._get_networks_config(networks) + + self.assertEqual(response['ethernet0.networkName'], 'eth') + self.assertEqual(response['ethernet0.address'], 'mac') + + def test_register_vm(self): + fake_path = 'fake/path' + fake_job_handle = mock.MagicMock() + + vixlib.VixHost_RegisterVM = mock.MagicMock() + vixlib.VixHost_RegisterVM.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixConnection.register_vm(fake_path) + + vixlib.VixHost_RegisterVM.assert_called_with( + self._VixConnection._host_handle, fake_path, None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_with(None) + + def _test_unregister_vm_local(self, platform): + fake_path = 'fake/path' + pref_file_path = 'fake_file' + other_fake_path = 'other/fake/path' + fake_vmx_path_norm = 'fake/vmx/path' + some_object = mock.Mock() + + sys.platform = mock.MagicMock() + sys.platform.return_value = platform + vixutils._get_player_preferences_file_path = mock.MagicMock() + vixutils._get_player_preferences_file_path.return_value = \ + pref_file_path + os.path.exists = mock.MagicMock() + os.path.exists.return_value = True + win32api.GetLongPathName = mock.MagicMock() + win32api.GetLongPathName.return_value = fake_path + os.path.abspath = mock.MagicMock() + os.path.abspath.return_value = other_fake_path + os.path.normcase = mock.MagicMock() + os.path.normcase.return_value = fake_vmx_path_norm + re.match = mock.MagicMock() + re.match.return_value = some_object + + with mock.patch('vix.vixutils.open', mock.mock_open(read_data=''), + create=True) as m: + print m.mock_calls + self._VixConnection._unregister_vm_local(fake_path) + m.assert_called_with(pref_file_path, 'r') + + if platform == 'win32': + win32api.GetLongPathName.assert_called_with(fake_path) + self.assertEqual(os.path.exists.call_count, 2) + else: + os.path.exists.assert_called_once() + os.path.abspath.assert_called_with(fake_path) + os.path.normcase.assert_called_with(other_fake_path) + + def _test_unregister_vm_local_platform_windows(self): + self._test_unregister_vm_local(platform='win32') + + def test_unregister_vm_local_other_platform(self): + self._test_unregister_vm_local(platform='linux') + + def test_unregister_vm_server(self): + fake_path = 'fake/path' + fake_job_handle = mock.MagicMock() + vixlib.VixHost_UnregisterVM = mock.MagicMock() + vixlib.VixHost_UnregisterVM.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + self._VixConnection._unregister_vm_server(fake_path) + + vixlib.VixHost_UnregisterVM.assert_called_with( + self._VixConnection._host_handle, fake_path, None, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_with(None) + + def _test_unregister_vm(self, vmware): + fake_path = 'fake/path' + + self._VixConnection._unregister_vm_local = mock.MagicMock() + self._VixConnection._unregister_vm_server = mock.MagicMock() + vixutils.get_vix_host_type = mock.MagicMock() + vixutils.get_vix_host_type.return_value = vmware + + if vmware == 1: + self.assertRaises(Exception, self._VixConnection.unregister_vm) + else: + self._VixConnection.unregister_vm(fake_path) + + vixutils.get_vix_host_type.assert_called_once() + if vmware == 3 or 4: + self._VixConnection._unregister_vm_local.assert_called_with( + fake_path) + else: + self._VixConnection._unregister_vm_server.assert_called_with( + fake_path) + + def test_unregister_vm_VMWARE_WORKSTATION(self): + self._test_unregister_vm(vmware=3) + + def test_unregister_vm_VMWARE_PLAYER(self): + self._test_unregister_vm(vmware=4) + + def _test_unregister_vm_default(self): + #not called ??get_vix_host_type raises exception?? + self._test_unregister_vm(vmware=1) + + def test_disconnect(self): + vixlib.VixHost_Disconnect = mock.MagicMock() + self._VixConnection.disconnect() + vixlib.VixHost_Disconnect.assert_called_once() + self.assertIsNone(self._VixConnection._host_handle) + + def test_vm_exists(self): + fake_path = 'fake/path' + os.path.exists = mock.MagicMock() + os.path.exists.return_value = True + + response = self._VixConnection.vm_exists(fake_path) + + os.path.exists.assert_called_with(fake_path) + self.assertEqual(response, True) + + def test_delete_vm_files(self): + fake_path = 'fake/path' + fake_name = 'fake_name' + os.path.dirname = mock.MagicMock() + os.path.dirname.return_value = fake_name + os.path.exists = mock.MagicMock() + os.path.exists.return_value = True + shutil.rmtree = mock.MagicMock() + + self._VixConnection.delete_vm_files(fake_path) + + os.path.dirname.assert_called_with(fake_path) + os.path.exists.assert_called_with(fake_name) + shutil.rmtree.assert_called_with(fake_name) + + def test_list_running_vms(self): + fake_job_handle = mock.MagicMock() + cb = mock.MagicMock() + + #cannot get inside nested callback method + vixlib.VixEventProc = mock.MagicMock() + vixlib.VixEventProc.return_value = cb + vixlib.VixHost_FindItems = mock.MagicMock() + vixlib.VixHost_FindItems.return_value = fake_job_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + response = self._VixConnection.list_running_vms() + + vixlib.VixEventProc.assert_called_once() + vixlib.VixHost_FindItems.assert_called_with( + self._VixConnection._host_handle, vixlib.VIX_FIND_RUNNING_VMS, + vixlib.VIX_INVALID_HANDLE, -1, cb, None) + vixlib.VixJob_Wait.assert_called_with(fake_job_handle, + vixlib.VIX_PROPERTY_NONE) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_once_with(None) + self.assertTrue(response is not None) + + def _test_get_tools_iso_path(self, platform): + fake_dir = 'fake_dir' + + sys.platform = platform + os.path.join = mock.MagicMock() + os.path.join.return_value = 'fake_dir/Contents/Library/isoimages' + vixutils._get_install_dir = mock.MagicMock() + vixutils._get_install_dir.return_value = fake_dir + + response = self._VixConnection.get_tools_iso_path() + + if platform == "darwin": + self.assertEqual(response, 'fake_dir/Contents/Library/isoimages') + elif platform == "win32": + self.assertEqual(response, fake_dir) + else: + self.assertEqual(response, "/usr/lib/vmware/isoimages") + + def test_get_tools_iso_path_darwin(self): + self._test_get_tools_iso_path(platform="darwin") + + def test_get_tools_iso_path_win32(self): + self._test_get_tools_iso_path(platform="win32") + + def test_get_tools_iso_path_linux(self): + self._test_get_tools_iso_path(platform="linux") + + def _test_clone_vm(self, linked_clone): + fake_src_vmx_path = 'fake_src_path' + fake_dest_vmx_path = 'fake_dest_path' + fake_job_handle = mock.MagicMock() + cloned_vm_handle = mock.MagicMock() + fake_vm = mock.MagicMock() + byref_mock = mock.MagicMock() + if linked_clone: + clone_type = vixlib.VIX_CLONETYPE_LINKED + else: + clone_type = vixlib.VIX_CLONETYPE_FULL + + self._VixConnection.open_vm = mock.MagicMock() + self._VixConnection.open_vm.return_value = fake_vm + vixlib.VixVM_Clone = mock.MagicMock() + vixlib.VixVM_Clone.return_value = fake_job_handle + vixlib.VixHandle = mock.MagicMock() + vixlib.VixHandle.return_value = cloned_vm_handle + vixlib.VixJob_Wait = mock.MagicMock() + vixlib.VixJob_Wait.return_value = None + ctypes.byref = mock.MagicMock() + ctypes.byref.return_value = byref_mock + vixlib.Vix_ReleaseHandle = mock.MagicMock() + vixutils._check_job_err_code = mock.MagicMock() + + response = self._VixConnection.clone_vm(fake_src_vmx_path, + fake_dest_vmx_path, + linked_clone) + + self._VixConnection.open_vm.assert_called_with(fake_src_vmx_path) + vixlib.VixVM_Clone.assert_called_with(fake_vm.__enter__()._vm_handle, + vixlib.VIX_INVALID_HANDLE, + clone_type, + fake_dest_vmx_path, 0, + vixlib.VIX_INVALID_HANDLE, + None, None) + vixlib.VixHandle.assert_called_once() + vixlib.VixJob_Wait.assert_called_with( + fake_job_handle, vixlib.VIX_PROPERTY_JOB_RESULT_HANDLE, + byref_mock, vixlib.VIX_PROPERTY_NONE) + ctypes.byref.assert_called_with(cloned_vm_handle) + vixlib.Vix_ReleaseHandle.assert_called_with(fake_job_handle) + vixutils._check_job_err_code.assert_called_with(None) + self.assertIsInstance(response, vixutils.VixVM) + + def test_clone_vm_linked_clone_true(self): + self._test_clone_vm(linked_clone=True) + + def test_clone_vm_linked_clone_false(self): + self._test_clone_vm(linked_clone=False) + + def test_get_software_version(self): + version = mock.MagicMock() + byref_mock = mock.MagicMock() + + ctypes.c_char_p = mock.MagicMock() + ctypes.c_char_p.return_value = version + vixlib.Vix_GetProperties = mock.MagicMock() + vixlib.Vix_GetProperties.return_value = None + vixutils._check_job_err_code = mock.MagicMock() + vixlib.Vix_FreeBuffer = mock.MagicMock() + ctypes.byref.return_value = byref_mock + + response = self._VixConnection.get_software_version() + + ctypes.c_char_p.assert_called_once() + vixlib.Vix_GetProperties.assert_called_with( + self._VixConnection._host_handle, + vixlib.VIX_PROPERTY_HOST_SOFTWARE_VERSION, byref_mock, + vixlib.VIX_PROPERTY_NONE) + vixutils._check_job_err_code.assert_called_with(None) + vixlib.Vix_FreeBuffer.assert_called_with(version) + self.assertTrue(response is not None) + + def test_get_host_type(self): + host_type = mock.MagicMock() + byref_mock = mock.MagicMock() + ctypes.c_int.return_value = host_type + vixlib.Vix_GetProperties = mock.MagicMock() + vixlib.Vix_GetProperties.return_value = None + vixutils._check_job_err_code = mock.MagicMock() + ctypes.byref.return_value = byref_mock + + response = self._VixConnection.get_host_type() + + vixlib.Vix_GetProperties.assert_called_with( + self._VixConnection._host_handle, + vixlib.VIX_PROPERTY_HOST_HOSTTYPE, byref_mock, + vixlib.VIX_PROPERTY_NONE) + vixutils._check_job_err_code.assert_called_with(None) + self.assertTrue(response is not None)