Kashyap | First real tests, using nova client to act and boto to assert

credentials.py

@@ -0,0 +1,10 @@
+#!/usr/bin/env python
+import os
+ 
+def get_nova_creds():
+    d = {}
+    d['username'] = os.environ['OS_USERNAME']
+    d['api_key'] = os.environ['OS_PASSWORD']
+    d['auth_url'] = os.environ['OS_AUTH_URL']
+    d['project_id'] = os.environ['OS_TENANT_NAME']
+    return d

ec2driver.py

@@ -145,6 +145,7 @@ class EC2Driver(driver.ComputeDriver):
 
     def spawn(self, context, instance, image_meta, injected_files,
               admin_password, network_info=None, block_device_info=None):
+        LOG.info("__Context__ %s, __instance__ %s, __image_meta__ %s, __injected_files__ %s, __admin_password__ %s" % (context, instance, image_meta, injected_files, admin_password))
         name = instance['name']
         state = power_state.RUNNING
         ec2_instance = EC2Instance(name, state)

test_spawn.py

@@ -3,6 +3,9 @@ import requests
 import json
 import time
 
+from novaclient.v1_1 import client
+from credentials import get_nova_creds
+
 from boto import ec2
 from ec2driver_config import *
 
@@ -12,30 +15,52 @@ class TestSpawn(unittest.TestCase):
     def setUp(self):
         print "Establishing connection with AWS"
         self.ec2_conn = ec2.connect_to_region(aws_region, aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key)
-
+        self.creds = get_nova_creds()
+    
+    # @unittest.skip("For fun")
     def test_spawn(self):
-        reservation = self.ec2_conn.run_instances(aws_ami, instance_type=instance_type)
-        self.instance = reservation.instances[0]
-        print self.instance
-        self.assertTrue(len(reservation.instances) >= 1)
-
+        print "Spawning an instance"
+        nova = client.Client(**self.creds)
+        image = nova.images.find(name="cirros-0.3.1-x86_64-uec")
+        flavor = nova.flavors.find(name="m1.tiny")
+        self.server = nova.servers.create(name = "cirros-test",
+                        image = image.id,
+                        flavor = flavor.id)
+        time.sleep(15)
+        
+        instance = self.ec2_conn.get_only_instances(instance_ids=[self.server.metadata['ec2_id']], filters=None, dry_run=False, max_results=None)
+        
+        self.assertTrue(len(instance) == 1)
+    
     def tearDown(self):
         print "Cleanup: Destroying the instance used for testing"
-        time.sleep(60)
-        self.ec2_conn.terminate_instances(instance_ids=[self.instance.id])
-
+        time.sleep(15)
+        self.server.delete()
+        
 class TestDestroy(unittest.TestCase):
     def setUp(self):
         print "Establishing connection with AWS"
         self.ec2_conn = ec2.connect_to_region(aws_region, aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key)
-
+        self.creds = get_nova_creds()
+    
+    # @unittest.skip("For fun")
     def test_destroy(self):
-        reservation = self.ec2_conn.run_instances(aws_ami, instance_type=instance_type)
-        print "Waiting for 120 seconds for the instance to change to running state"
-        time.sleep(60)
-        instance_count = len(reservation.instances)
-        terminated_instances = self.ec2_conn.terminate_instances(instance_ids=[reservation.instances[0].id])
-        self.assertTrue(len(terminated_instances) == 1)
+        print "Spawning an instance"
+        nova = client.Client(**self.creds)
+        image = nova.images.find(name="cirros-0.3.1-x86_64-uec")
+        flavor = nova.flavors.find(name="m1.tiny")
+        server = nova.servers.create(name = "cirros-test",
+                        image = image.id,
+                        flavor = flavor.id)
+        time.sleep(20)
+        ec2_id = server.metadata['ec2_id']
+        server.delete()
+
+        time.sleep(10)
+        instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id], filters=None, dry_run=False, max_results=None)
+        
+        shutting_down_state_code = 32
+        self.assertEquals(instance[0].state_code, shutting_down_state_code)
 
 if __name__ == '__main__':
     unittest.main()