worked on unittests

occi_os_api/backends/compute.py

@@ -75,6 +75,7 @@ class ComputeBackend(KindBackend, ActionBackend):
                          infrastructure.SUSPEND,
                          infrastructure.RESTART]
 
+        # Tell the world that is is an VM in OpenStack...
         entity.mixins.append(os_addon.OS_VM)
 
     def retrieve(self, entity, extras):
@@ -111,6 +112,10 @@ class ComputeBackend(KindBackend, ActionBackend):
         LOG.debug('Updating an Virtual machine: ' + repr(uid))
 
         # for now we will only handle one mixin change per request
+        if len(new.mixins) != 1:
+            raise AttributeError('Only updates with one mixin in request are'
+                                 ' currently supported')
+
         mixin = new.mixins[0]
         if isinstance(mixin, os_mixins.ResourceTemplate):
             flavor_name = mixin.term

run_tests.sh

@@ -26,11 +26,12 @@ fi
 
 echo '\n Code style        \n****************************************\n'
 
-pep8 --repeat --statistics --count occi_os_api
+pep8 --repeat --statistics --count occi_os_api tests
 
 echo '\n Issues report     \n****************************************\n'
 
 pyflakes occi_os_api
+vulture occi_os_api
 
 echo '\n Pychecker report  \n****************************************\n'
 

tests/system_test.py

@@ -29,8 +29,8 @@ import unittest
 import random
 
 
-HEADS = {'Content-Type':'text/occi',
+HEADS = {'Content-Type': 'text/occi',
-         'Accept':'text/occi'
+         'Accept': 'text/occi'
 }
 
 KEYSTONE_HOST = '127.0.0.1:5000'
@@ -74,9 +74,9 @@ def get_os_token(username, password):
     """
     Get a security token from Keystone.
     """
-    body = '{"auth": {"tenantName": "'+username+'", ' \
+    body = '{"auth": {"tenantName": "' + username + '", ' \
-           '"passwordCredentials":{"username": "'+username+'", ' \
+           '"passwordCredentials":{"username": "' + username + '", ' \
-           '"password": "'+password+'"}}}'
+           '"password": "' + password + '"}}}'
 
     heads = {'Content-Type': 'application/json'}
     conn = httplib.HTTPConnection(KEYSTONE_HOST)
@@ -154,7 +154,7 @@ def destroy_node(token, location):
     return heads
 
 
-def trigger_action(token, url, action_cat, action_param =None):
+def trigger_action(token, url, action_cat, action_param=None):
     """
     Trigger an OCCI action.
     """
@@ -172,7 +172,6 @@ class SystemTest(unittest.TestCase):
     Do a simple set of test.
     """
 
-
     def setUp(self):
         """
         Setup the test.
@@ -181,7 +180,6 @@ class SystemTest(unittest.TestCase):
         self.token = get_os_token('admin', 'os4all')
         LOG.info('security token is: ' + self.token)
 
-
     def test_compute_node(self):
         """
         Test ops on a compute node!
@@ -242,7 +240,6 @@ class SystemTest(unittest.TestCase):
         # delete
         destroy_node(self.token, vm_location)
 
-
     def test_security_grouping(self):
         """
         Test some security and accessibility stuff!
@@ -298,9 +295,9 @@ class SystemTest(unittest.TestCase):
                 '.org/occi/infrastructure#"', 'ipnetworkinterface; '
                                               'scheme="http://schemas.ogf'
                 '.org/occi/infrastructure/networkinterface#"']
-        attrs = ['occi.core.source=http://"' + OCCI_HOST + vm_location +  '"',
+        attrs = ['occi.core.source=http://"' + OCCI_HOST + vm_location + '"',
                  'occi.core.target=http://"' + OCCI_HOST +
-                 '/network/public"',]
+                 '/network/public"']
         float_ip_location = create_node(self.token, cats, attrs)
 
         time.sleep(15)
@@ -327,21 +324,21 @@ class SystemTest(unittest.TestCase):
         heads['Category'] = name + '; scheme="http://www.mystuff.org/sec#"'
         do_request('DELETE', '/-/', heads)
 
-
     def test_storage_stuff(self):
         """
         Test attaching and detaching storage volumes + snapshotting etc.
         """
         # create new VM
         cats = ['m1.tiny; scheme="http://schemas.openstack'
-                '.org/template/resource#"',
+            '.org/template/resource#"',
-                'cirros-0.3.0-x86_64-uec; scheme="http://schemas.openstack'
+            'cirros-0.3.0-x86_64-uec; scheme="http://schemas.openstack'
-                '.org/template/os#"',
+            '.org/template/os#"',
-                'compute; scheme="http://schemas.ogf.org/occi/infrastructure#"']
+            'compute; scheme="http://schemas.ogf.org/occi/infrastructure#"']
         vm_location = create_node(self.token, cats)
 
         # create volume
-        cats = ['storage; scheme="http://schemas.ogf.org/occi/infrastructure#"']
+        cats = ['storage; scheme="http://schemas.ogf'
+                '.org/occi/infrastructure#"']
         attrs = ['occi.storage.size = 1.0']
         vol_location = create_node(self.token, cats, attrs)
 
@@ -353,7 +350,7 @@ class SystemTest(unittest.TestCase):
         # link volume and compute
         cats = ['storagelink; scheme="http://schemas.ogf'
                 '.org/occi/infrastructure#"']
-        attrs = ['occi.core.source=http://"' + OCCI_HOST + vm_location +  '"',
+        attrs = ['occi.core.source=http://"' + OCCI_HOST + vm_location + '"',
                  'occi.core.target=http://"' + OCCI_HOST + vol_location + '"',
                  'occi.storagelink.deviceid="/dev/vdc"']
         link_location = create_node(self.token, cats, attrs)
@@ -388,17 +385,16 @@ class SystemTest(unittest.TestCase):
 
         destroy_node(self.token, vm_location)
 
-
     def test_scaling(self):
         """
         Test the scaling operations
         """
         # create new VM
         cats = ['itsy; scheme="http://schemas.openstack'
-                '.org/template/resource#"',
+            '.org/template/resource#"',
-                'cirros-0.3.0-x86_64-uec; scheme="http://schemas.openstack'
+            'cirros-0.3.0-x86_64-uec; scheme="http://schemas.openstack'
-                '.org/template/os#"',
+            '.org/template/os#"',
-                'compute; scheme="http://schemas.ogf.org/occi/infrastructure#"']
+            'compute; scheme="http://schemas.ogf.org/occi/infrastructure#"']
         vm_location = create_node(self.token, cats)
 
         # wait
@@ -427,4 +423,3 @@ class SystemTest(unittest.TestCase):
                 time.sleep(5)
 
         destroy_node(self.token, vm_location)
-

tests/test_backends_compute.py

@@ -23,20 +23,26 @@ import unittest
 
 # depenency from nova :-)
 import mox
+from nova.compute import vm_states
 
 from occi import core_model
 from occi.extensions import infrastructure
-from occi_os_api import nova_glue
 
+from occi_os_api import nova_glue
 from occi_os_api.backends import compute
 from occi_os_api.extensions import os_mixins
 
+
 class TestComputeBackend(unittest.TestCase):
     """
     Tests the compute backend.
     """
 
-    os_template = os_mixins.OsTemplate('', '')
+    os_template = os_mixins.OsTemplate('http://example.com', 'unix')
+    os_template2 = os_mixins.OsTemplate('http://example.com', 'windows')
+
+    res_template = os_mixins.ResourceTemplate('http://example.com', 'itsy')
+    res_template2 = os_mixins.ResourceTemplate('http://example.com', 'bitsy')
 
     def setUp(self):
         """
@@ -52,9 +58,11 @@ class TestComputeBackend(unittest.TestCase):
         """
         self.mox.UnsetStubs()
 
+    # Test for failure
+
     def test_create_for_failure(self):
         """
-        Test for proper error handling
+        Test for proper error handling.
         """
         # msg OS template
         res = core_model.Resource('/foo/bar', infrastructure.COMPUTE, [])
@@ -70,6 +78,55 @@ class TestComputeBackend(unittest.TestCase):
         self.assertRaises(AttributeError, self.backend.create, res,
             self.sec_obj)
 
+    def test_update_for_failure(self):
+        """
+        Test if correct errors are thrown.
+        """
+        # msg mixin
+        res1 = core_model.Resource('/foo/bar', infrastructure.COMPUTE, [])
+        res1.attributes = {'occi.core.id': 'bar'}
+        res2 = core_model.Resource('/foo/bar', infrastructure.COMPUTE, [])
+
+        self.assertRaises(AttributeError, self.backend.update, res1, res2,
+            self.sec_obj)
+
+        res2 = core_model.Resource('/foo/bar', infrastructure.COMPUTE,
+            [core_model.Category('http://foo.com', 'bar', '', '', '')])
+
+        self.assertRaises(AttributeError, self.backend.update, res1, res2,
+            self.sec_obj)
+
+    def test_action_for_failure(self):
+        """
+        Test if correct errors are thrown.
+        """
+        # wrong action
+        res1 = core_model.Resource('/foo/bar', infrastructure.COMPUTE, [])
+        res1.attributes = {'occi.core.id': 'bar'}
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'vm_state': vm_states.STOPPED
+            })
+        self.mox.ReplayAll()
+        self.assertRaises(AttributeError, self.backend.action, res1,
+            infrastructure.STOP, {}, self.sec_obj)
+        self.mox.VerifyAll()
+
+        # missing method!
+        self.mox.UnsetStubs()
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'vm_state': vm_states.ACTIVE
+            })
+        self.mox.ReplayAll()
+        self.assertRaises(AttributeError, self.backend.action, res1,
+                            infrastructure.RESTART, {}, self.sec_obj)
+        self.mox.VerifyAll()
+
+    # Test for Sanity
+
     def test_create_for_sanity(self):
         """
         Simulate a create call!
@@ -94,11 +151,180 @@ class TestComputeBackend(unittest.TestCase):
 
         self.backend.create(res, self.sec_obj)
 
-        # TODO check if all attrs are there!
+        # check if all attrs are there!
-        self.assertEqual(True, True)
+        self.assertIn('occi.compute.hostname', res.attributes)
+        self.assertIn('occi.compute.architecture', res.attributes)
+        self.assertIn('occi.compute.cores', res.attributes)
+        self.assertIn('occi.compute.speed', res.attributes)
+        self.assertIn('occi.compute.memory', res.attributes)
+        self.assertIn('occi.compute.state', res.attributes)
+
+        self.assertEqual('inactive', res.attributes['occi.compute.state'])
+
+        self.assertListEqual([infrastructure.STOP, infrastructure.SUSPEND,
+                              infrastructure.RESTART], res.actions)
 
         self.mox.VerifyAll()
 
+    def test_retrieve_for_sanity(self):
+        """
+        Simulate a retrieve call!
+        """
+        res = core_model.Resource('/foo/bar', infrastructure.COMPUTE,
+            [self.os_template])
+        res.attributes = {'occi.core.id': 'bar'}
 
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_occi_state')
+        nova_glue.vm.get_occi_state(mox.IsA(object),
+            mox.IsA(object)).AndReturn(('active', [infrastructure.STOP,
+                                                   infrastructure.SUSPEND,
+                                                   infrastructure.RESTART]))
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'hostname': 'bar',
+                'vcpus': 1,
+                'memory_mb': 256
+            })
+        self.mox.StubOutWithMock(nova_glue.storage, 'get_image_architecture')
+        nova_glue.storage.get_image_architecture(mox.IsA(object),
+            mox.IsA(object)).AndReturn(
+            'foo')
+        self.mox.ReplayAll()
 
+        self.backend.retrieve(res, self.sec_obj)
 
+        # check if all attrs are there!
+        self.assertIn('occi.compute.hostname', res.attributes)
+        self.assertIn('occi.compute.architecture', res.attributes)
+        self.assertIn('occi.compute.cores', res.attributes)
+        self.assertIn('occi.compute.speed', res.attributes)
+        self.assertIn('occi.compute.memory', res.attributes)
+        self.assertIn('occi.compute.state', res.attributes)
+
+        self.assertIn('occi.core.id', res.attributes)
+
+        self.assertEqual('active', res.attributes['occi.compute.state'])
+
+        self.assertListEqual([infrastructure.STOP, infrastructure.SUSPEND,
+                              infrastructure.RESTART], res.actions)
+
+        self.mox.VerifyAll()
+
+    def test_update_for_sanity(self):
+        """
+        Simulate a update call!
+        """
+        res1 = core_model.Resource('/foo/bar', infrastructure.COMPUTE,
+            [self.os_template])
+        res1.attributes = {'occi.core.id': 'bar'}
+
+        # case 1 - rebuild VM with different OS
+        res2 = core_model.Resource('/foo/bar', infrastructure.COMPUTE,
+            [self.os_template2])
+
+        self.mox.StubOutWithMock(nova_glue.vm, 'rebuild_vm')
+        nova_glue.vm.rebuild_vm(mox.IsA(object), mox.IsA(object),
+            mox.IsA(object))
+        self.mox.ReplayAll()
+        self.backend.update(res1, res2, self.sec_obj)
+
+        self.assertIn(self.os_template2, res1.mixins)
+
+        self.mox.VerifyAll()
+
+        # case 2 - resize the VM
+        res2 = core_model.Resource('/foo/bar', infrastructure.COMPUTE,
+            [self.res_template2])
+
+        self.mox.StubOutWithMock(nova_glue.vm, 'resize_vm')
+        nova_glue.vm.resize_vm(mox.IsA(object), mox.IsA(object),
+            mox.IsA(object))
+        self.mox.ReplayAll()
+        self.backend.update(res1, res2, self.sec_obj)
+
+        self.assertIn(self.res_template2, res1.mixins)
+
+        self.mox.VerifyAll()
+
+    def test_replace_for_sanity(self):
+        """
+        Simulate a replace call - does nothing atm.
+        """
+        self.backend.replace(None, None, self.sec_obj)
+
+    def test_delete_for_sanity(self):
+        """
+        Simulate a delete call.
+        """
+        res = core_model.Resource('/foo/bar', infrastructure.COMPUTE,
+            [self.os_template])
+        res.attributes = {'occi.core.id': 'bar'}
+
+        self.mox.StubOutWithMock(nova_glue.vm, 'delete_vm')
+        nova_glue.vm.delete_vm(mox.IsA(object), mox.IsA(object))
+        self.mox.ReplayAll()
+        self.backend.delete(res, self.sec_obj)
+
+        self.mox.VerifyAll()
+
+    def test_action_for_sanity(self):
+        """
+        Test actions
+        """
+        res1 = core_model.Resource('/foo/bar', infrastructure.COMPUTE, [])
+        res1.attributes = {'occi.core.id': 'bar'}
+
+        # start
+        self.mox.StubOutWithMock(nova_glue.vm, 'start_vm')
+        nova_glue.vm.start_vm(mox.IsA(object), mox.IsA(object))
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'vm_state': vm_states.STOPPED
+            })
+        self.mox.ReplayAll()
+        self.backend.action(res1, infrastructure.START, {}, self.sec_obj)
+        self.mox.VerifyAll()
+
+        # stop
+        self.mox.UnsetStubs()
+        self.mox.StubOutWithMock(nova_glue.vm, 'stop_vm')
+        nova_glue.vm.stop_vm(mox.IsA(object), mox.IsA(object))
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'vm_state': vm_states.ACTIVE
+            })
+        self.mox.ReplayAll()
+        self.backend.action(res1, infrastructure.STOP, {}, self.sec_obj)
+        self.mox.VerifyAll()
+
+        # reboot
+        self.mox.UnsetStubs()
+        self.mox.StubOutWithMock(nova_glue.vm, 'restart_vm')
+        nova_glue.vm.restart_vm(mox.IsA(object), mox.IsA(str),
+            mox.IsA(object))
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'vm_state': vm_states.ACTIVE
+            })
+        self.mox.ReplayAll()
+        self.backend.action(res1, infrastructure.RESTART,
+            {'method': 'graceful'},
+            self.sec_obj)
+        self.mox.VerifyAll()
+
+        # suspend
+        self.mox.UnsetStubs()
+        self.mox.StubOutWithMock(nova_glue.vm, 'suspend_vm')
+        nova_glue.vm.suspend_vm(mox.IsA(object), mox.IsA(object))
+        self.mox.StubOutWithMock(nova_glue.vm, 'get_vm')
+        nova_glue.vm.get_vm(mox.IsA(object), mox.IsA(object)).AndReturn(
+            {
+                'vm_state': vm_states.ACTIVE
+            })
+        self.mox.ReplayAll()
+        self.backend.action(res1, infrastructure.SUSPEND, {}, self.sec_obj)
+        self.mox.VerifyAll()