* code cleanup

* revised unittest approach
* added stubout and a number of tests

1 files changed, 126 insertions, 10 deletions

diff --git a/nova/tests/xenapi_unittest.py b/nova/tests/xenapi_unittest.py
index c9be17327..b9955a946 100644
--- a/nova/tests/xenapi_unittest.py
+++ b/nova/tests/xenapi_unittest.py
@@ -31,7 +31,7 @@
 #    under the License.
 
 
-import mox
+import stubout
 import uuid
 
 from twisted.internet import defer
@@ -51,20 +51,34 @@ from nova.virt.xenapi import fake
 from nova.virt.xenapi import volume_utils
 from nova.virt.xenapi import vm_utils
 from nova.virt.xenapi import volumeops
+from boto.ec2.volume import Volume
 
 FLAGS = flags.FLAGS
 
 
+def stubout_session(stubs, cls):
+    def fake_import(self):
+        fake_module = 'nova.virt.xenapi.fake'
+        from_list = ['fake']
+        return __import__(fake_module, globals(), locals(), from_list, -1)
+
+    stubs.Set(xenapi_conn.XenAPISession, '_create_session',
+                       lambda s, url: cls(url))
+    stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi',
+                       fake_import)
+
+
 class XenAPIVolumeTestCase(test.TrialTestCase):
     """
-    This uses Ewan's fake session approach
+    Unit tests for VM operations
     """
     def setUp(self):
         super(XenAPIVolumeTestCase, self).setUp()
-        FLAGS.xenapi_use_fake_session = True
+        self.stubs = stubout.StubOutForTesting()
         FLAGS.target_host = '127.0.0.1'
         FLAGS.xenapi_connection_url = 'test_url'
         FLAGS.xenapi_connection_password = 'test_pass'
+        fake.reset()
 
     def _create_volume(self, size='0'):
         """Create a volume object."""
@@ -78,11 +92,12 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
         vol['attach_status'] = "detached"
         return db.volume_create(context.get_admin_context(), vol)
 
-    def test_create_iscsi_storage_raise_no_exception(self):
-        fake.reset()
+    def test_create_iscsi_storage(self):
+        """ This shows how to test helper classes' methods """
+        stubout_session(self.stubs, FakeSessionForVolumeTests)
         session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
         helper = volume_utils.VolumeHelper
-        helper.late_import(FLAGS)
+        helper.XenAPI = session.get_imported_xenapi()
         vol = self._create_volume()
         info = yield helper.parse_volume_info(vol['ec2_id'], '/dev/sdc')
         label = 'SR-%s' % vol['ec2_id']
@@ -93,8 +108,25 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
                                                       description)
         db.volume_destroy(context.get_admin_context(), vol['id'])
 
+    def test_parse_volume_info_raise_exception(self):
+        """ This shows how to test helper classes' methods """
+        stubout_session(self.stubs, FakeSessionForVolumeTests)
+        session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
+        helper = volume_utils.VolumeHelper
+        helper.XenAPI = session.get_imported_xenapi()
+        vol = self._create_volume()
+        # oops, wrong mount point!
+        info = helper.parse_volume_info(vol['ec2_id'], '/dev/sd')
+
+        def check(exc):
+            self.assertIsInstance(exc.value, volume_utils.StorageError)
+
+        info.addErrback(check)
+        db.volume_destroy(context.get_admin_context(), vol['id'])
+
     def test_attach_volume(self):
-        fake.reset()
+        """ This shows how to test Ops classes' methods """
+        stubout_session(self.stubs, FakeSessionForVolumeTests)
         conn = xenapi_conn.get_connection(False)
         volume = self._create_volume()
         instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3,
@@ -116,13 +148,34 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
         result.addCallback(check)
         return result
 
+    def test_attach_volume_raise_exception(self):
+        """ This shows how to test when exceptions are raised """
+        stubout_session(self.stubs, FakeSessionForVolumeFailedTests)
+        conn = xenapi_conn.get_connection(False)
+        volume = self._create_volume()
+        instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3,
+                                'm1.large', 'aa:bb:cc:dd:ee:ff')
+        fake.create_vm(instance.name, 'Running')
+        result = conn.attach_volume(instance.name, volume['ec2_id'],
+                                    '/dev/sdc')
+
+        def check(exc):
+            if exc:
+                pass
+            else:
+                self.fail('Oops, no exception has been raised!')
+
+        result.addErrback(check)
+        return result
+
     def tearDown(self):
         super(XenAPIVolumeTestCase, self).tearDown()
+        self.stubs.UnsetAll()
 
 
 class XenAPIVMTestCase(test.TrialTestCase):
     """
-    This uses Ewan's fake session approach
+    Unit tests for VM operations
     """
     def setUp(self):
         super(XenAPIVMTestCase, self).setUp()
@@ -131,19 +184,20 @@ class XenAPIVMTestCase(test.TrialTestCase):
                                              admin=True)
         self.project = self.manager.create_project('fake', 'fake', 'fake')
         self.network = utils.import_object(FLAGS.network_manager)
-        FLAGS.xenapi_use_fake_session = True
+        self.stubs = stubout.StubOutForTesting()
         FLAGS.xenapi_connection_url = 'test_url'
         FLAGS.xenapi_connection_password = 'test_pass'
         fake.reset()
         fake.create_network('fake', FLAGS.flat_network_bridge)
 
     def test_list_instances_0(self):
+        stubout_session(self.stubs, FakeSessionForVMTests)
         conn = xenapi_conn.get_connection(False)
         instances = conn.list_instances()
         self.assertEquals(instances, [])
-    #test_list_instances_0.skip = "E"
 
     def test_spawn(self):
+        stubout_session(self.stubs, FakeSessionForVMTests)
         conn = xenapi_conn.get_connection(False)
         instance = FakeInstance(1, self.project.id, self.user.id, 1, 2, 3,
                                 'm1.large', 'aa:bb:cc:dd:ee:ff')
@@ -186,6 +240,7 @@ class XenAPIVMTestCase(test.TrialTestCase):
         super(XenAPIVMTestCase, self).tearDown()
         self.manager.delete_project(self.project)
         self.manager.delete_user(self.user)
+        self.stubs.UnsetAll()
 
 
 class FakeInstance():
@@ -199,3 +254,64 @@ class FakeInstance():
         self.ramdisk_id = ramdisk_id
         self.instance_type = instance_type
         self.mac_address = mac_address
+
+
+class FakeSessionForVMTests(fake.SessionBase):
+    def __init__(self, uri):
+        super(FakeSessionForVMTests, self).__init__(uri)
+
+    def network_get_all_records_where(self, _1, _2):
+        return self.xenapi.network.get_all_records()
+
+    def host_call_plugin(self, _1, _2, _3, _4, _5):
+        return ''
+
+    def VM_start(self, _1, ref, _2, _3):
+        vm = fake.get_record('VM', ref)
+        if vm['power_state'] != 'Halted':
+            raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
+                                  vm['power_state']])
+        vm['power_state'] = 'Running'
+
+
+class FakeSessionForVolumeTests(fake.SessionBase):
+    def __init__(self, uri):
+        super(FakeSessionForVolumeTests, self).__init__(uri)
+
+    def VBD_plug(self, _1, _2):
+        #FIXME(armando):make proper plug
+        pass
+
+    def PBD_unplug(self, _1, _2):
+        #FIXME(armando):make proper unplug
+        pass
+
+    def SR_forget(self, _1, _2):
+        #FIXME(armando):make proper forget
+        pass
+
+    def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
+                      _6, _7, _8, _9, _10, _11):
+        #FIXME(armando):make proper introduce
+        valid_vdi = False
+        refs = fake.get_all('VDI')
+        for ref in refs:
+            rec = fake.get_record('VDI', ref)
+            if rec['uuid'] == uuid:
+                valid_vdi = True
+        if not valid_vdi:
+            raise fake.Failure([['INVALID_VDI', 'session', self._session]])
+
+
+class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
+    def __init__(self, uri):
+        super(FakeSessionForVolumeFailedTests, self).__init__(uri)
+
+    def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
+                      _6, _7, _8, _9, _10, _11):
+        # test failure
+        raise fake.Failure([['INVALID_VDI', 'session', self._session]])
+
+    def VBD_plug(self, _1, _2):
+        # test failure
+        raise fake.Failure([['INVALID_VBD', 'session', self._session]])