Cam & Ed | Wait for elastic ip to be released in test teardown, assert on releasing elastic ip in destroy instance test

tests/ec2_test_base.py

@@ -1,7 +1,9 @@
 import unittest
 import time
+from boto.exception import EC2ResponseError
 
 from boto.regioninfo import RegionInfo
+import datetime
 from novaclient.v1_1 import client
 from boto import ec2
 
@@ -15,8 +17,6 @@ class EC2TestBase(unittest.TestCase):
         if not os.environ.get('MOCK_EC2'):
             time.sleep(seconds)
 
-
-    @classmethod
     def setUp(self):
         print "Establishing connection with AWS"
 
@@ -38,20 +38,17 @@ class EC2TestBase(unittest.TestCase):
         self.servers = []
         self.volumes = []
 
-    @classmethod
     def tearDown(self):
         print "Cleanup: Destroying the instance used for testing"
         for instance in self.servers:
-            instance.delete()
-
-        # wait for all instances to completely shut down and detach volumes if any
-        self.sleep_if_ec2_not_mocked(120)
+            self.destroy_instance_and_release_elastic_ip(instance)
 
         for volume in self.volumes:
             volume.delete()
 
     def spawn_ec2_instance(self):
         print "aws_region: " + aws_region
+        print "time: " + str(datetime.datetime.now().time())
 
         print "Spawning an instance"
         image = self.nova.images.find(name="cirros-0.3.1-x86_64-uec")
@@ -63,4 +60,29 @@ class EC2TestBase(unittest.TestCase):
             EC2TestBase.sleep_if_ec2_not_mocked(10)
             instance = self.nova.servers.get(server.id)
         self.servers.append(instance)
-        return instance, server.id
+        return instance, server.id
+
+    def destroy_instance_and_release_elastic_ip(self, instance):
+        instance.delete()
+        self.servers.remove(instance)
+        self._wait_for_instance_to_be_destroyed(instance.metadata['ec2_id'])
+        self._wait_for_elastic_ip_to_be_released(instance.metadata['public_ip_address'])
+
+    def _wait_for_elastic_ip_to_be_released(self, public_ip):
+        while True:
+            try:
+                print 'Waiting for EC2 elastic ip to be released...'
+                self.sleep_if_ec2_not_mocked(10)
+                eip = self.ec2_conn.get_all_addresses(public_ip)
+            except EC2ResponseError:
+                print 'Elastic ip released.'
+                break
+
+    def _wait_for_instance_to_be_destroyed(self, ec2_id):
+        ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
+        # while ec2_instance.state not in ("shutting-down", "terminated"):
+        while ec2_instance.state is not "terminated":
+            print 'Waiting for EC2 instance to shut down...'
+            self.sleep_if_ec2_not_mocked(10)
+            ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
+        print 'Instance has been shut down.'

tests/test_ec2driver.py

@@ -1,5 +1,6 @@
 import unittest
 import urllib2
+from boto.exception import EC2ResponseError
 
 from novaclient.v1_1 import client
 from ..ec2driver_config import *
@@ -18,7 +19,6 @@ class TestEC2Driver(EC2TestBase):
         self.assertEqual(ec2_instance[0].id, instance.metadata['ec2_id'])
         self.assertEqual(ec2_eip.instance_id, instance.metadata['ec2_id'])
 
-
     def test_destroy(self):
         print "******* Destroy Test ***********"
         instance, instance_ref = self.spawn_ec2_instance()
@@ -30,7 +30,10 @@ class TestEC2Driver(EC2TestBase):
         while ec2_instance.state != "running":
             self.sleep_if_ec2_not_mocked(10)
             ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
-        instance.delete()
+
+        elastic_ip = self.ec2_conn.get_all_addresses(addresses=instance.metadata['public_ip_address'])[0]
+
+        self.destroy_instance_and_release_elastic_ip(instance)
 
         ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
         # EC2 statecode: 16->Running, 32->Shutting Down
@@ -40,6 +43,12 @@ class TestEC2Driver(EC2TestBase):
 
         self.assertTrue(ec2_instance.state in ("shutting-down", "terminated"))
 
+        with self.assertRaises(EC2ResponseError):
+            for i in xrange(0,20):
+                self.sleep_if_ec2_not_mocked(5)
+                eip = self.ec2_conn.get_all_addresses(elastic_ip.public_ip)
+                print "Elastic IP for i=", i, ": "
+                print eip
 
     def test_power_off(self):
         print "******* Power Off Test ***********"