x
/
ec2-driver
Code Issues Proposed changes

Cam & Ed | Extract setup/teardown/helper methods for tests into superclass

This commit is contained in:
Ed Thome 2014-10-21 19:42:11 +00:00
parent 8d0078e352
commit d5abe2973e
3 changed files with 110 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
ec2driver.py
View File

@ -16,6 +16,7 @@
"""Connection to the Amazon Web Services - EC2 service"""
from threading import Lock
import base64
import time
from ec2driver_config import *
from boto import ec2
import boto.ec2.cloudwatch

66
tests/ec2_test_base.py Normal file
View File

@ -0,0 +1,66 @@
import os
import unittest
import time
from boto.regioninfo import RegionInfo
from novaclient.v1_1 import client
from boto import ec2
from ..credentials import get_nova_creds
from ..ec2driver_config import *
class EC2TestBase(unittest.TestCase):
@staticmethod
def sleep_if_ec2_not_mocked(seconds):
if not os.environ.get('MOCK_EC2'):
time.sleep(seconds)
@classmethod
def setUp(self):
print "Establishing connection with AWS"
region = RegionInfo(name=aws_region, endpoint=aws_endpoint)
self.ec2_conn = ec2.EC2Connection(aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
host=host,
port=port,
region = region,
is_secure=secure)
self.creds = get_nova_creds()
self.nova = client.Client(**self.creds)
# nova client for cinder
self.creds['service_type'] = 'volume'
self.nova_volume = client.Client(**self.creds)
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)
for volume in self.volumes:
volume.delete()
def spawn_ec2_instance(self):
print "aws_region: " + aws_region
print "Spawning an instance"
image = self.nova.images.find(name="cirros-0.3.1-x86_64-uec")
flavor = self.nova.flavors.find(name="m1.tiny")
server = self.nova.servers.create(
name="cirros-test", image=image.id, flavor=flavor.id)
instance = self.nova.servers.get(server.id)
while instance.status != 'ACTIVE':
EC2TestBase.sleep_if_ec2_not_mocked(10)
instance = self.nova.servers.get(server.id)
self.servers.append(instance)
return instance, server.id

144
tests/test_ec2driver.py
View File

@ -1,59 +1,12 @@
import unittest
import time
import urllib2
from boto.regioninfo import RegionInfo
from novaclient.v1_1 import client
from boto import ec2
from ..credentials import get_nova_creds
from ..ec2driver_config import *
from ec2_test_base import EC2TestBase
def sleep_if_ec2_not_mocked(seconds):
if not os.environ.get('MOCK_EC2'):
time.sleep(seconds)
class EC2DriverTest(unittest.TestCase):
_multiprocess_shared_ = True
@classmethod
def setUp(self):
print "Establishing connection with AWS"
moto_region = RegionInfo(name=aws_region, endpoint=aws_endpoint)
self.ec2_conn = ec2.EC2Connection(aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
host=host,
port=port,
region = moto_region,
is_secure=secure)
self.creds = get_nova_creds()
self.nova = client.Client(**self.creds)
# nova client for cinder
self.creds['service_type'] = 'volume'
self.nova_volume = client.Client(**self.creds)
self.servers = []
self.volumes = []
def spawn_ec2_instance(self):
print "aws_region: " + aws_region
print "Spawning an instance"
image = self.nova.images.find(name="cirros-0.3.1-x86_64-uec")
flavor = self.nova.flavors.find(name="m1.tiny")
server = self.nova.servers.create(
name="cirros-test", image=image.id, flavor=flavor.id)
instance = self.nova.servers.get(server.id)
while instance.status != 'ACTIVE':
sleep_if_ec2_not_mocked(10)
instance = self.nova.servers.get(server.id)
self.servers.append(instance)
return instance, server.id
class TestEC2Driver(EC2TestBase):
def test_spawn(self):
print "******* Spawn Test ***********"
@ -75,14 +28,14 @@ class EC2DriverTest(unittest.TestCase):
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
# EC2 statecode: 16->Running, 32->Shutting Down
while ec2_instance.state != "running":
sleep_if_ec2_not_mocked(10)
self.sleep_if_ec2_not_mocked(10)
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
instance.delete()
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
# EC2 statecode: 16->Running, 32->Shutting Down
while ec2_instance.state not in ("shutting-down", "terminated"):
sleep_if_ec2_not_mocked(10)
self.sleep_if_ec2_not_mocked(10)
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
self.assertTrue(ec2_instance.state in ("shutting-down", "terminated"))
@ -95,7 +48,7 @@ class EC2DriverTest(unittest.TestCase):
self.nova.servers.stop(instance)
while instance.status != 'SHUTOFF':
sleep_if_ec2_not_mocked(5)
self.sleep_if_ec2_not_mocked(5)
instance = self.nova.servers.get(instance.id)
# assert power off
@ -116,7 +69,7 @@ class EC2DriverTest(unittest.TestCase):
instance = self.nova.servers.get(instance.id)
while instance.status != 'ACTIVE':
sleep_if_ec2_not_mocked(5)
self.sleep_if_ec2_not_mocked(5)
instance = self.nova.servers.get(instance.id)
#assert restarted
@ -132,11 +85,11 @@ class EC2DriverTest(unittest.TestCase):
# we are waiting for the status to actually get to 'Hard Reboot' before
# beginning to wait for it to go to 'Active' status
while instance.status != 'HARD_REBOOT':
sleep_if_ec2_not_mocked(5)
self.sleep_if_ec2_not_mocked(5)
instance = self.nova.servers.get(instance.id)
while instance.status != 'ACTIVE':
sleep_if_ec2_not_mocked(5)
self.sleep_if_ec2_not_mocked(5)
instance = self.nova.servers.get(instance.id)
#assert restarted
@ -161,14 +114,14 @@ class EC2DriverTest(unittest.TestCase):
# wait for the status to actually go to Verify_Resize, before
# confirming the resize.
while instance.status != 'VERIFY_RESIZE':
sleep_if_ec2_not_mocked(5)
self.sleep_if_ec2_not_mocked(5)
instance = self.nova.servers.get(instance.id)
# Confirm the resize
self.nova.servers.confirm_resize(instance)
while instance.status != 'ACTIVE':
sleep_if_ec2_not_mocked(5)
self.sleep_if_ec2_not_mocked(5)
instance = self.nova.servers.get(instance.id)
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']])[0]
@ -181,64 +134,53 @@ class EC2DriverTest(unittest.TestCase):
@unittest.skipIf(os.environ.get('MOCK_EC2'), 'Not supported by moto')
def test_user_data(self):
"""To test the spawn method by providing a file user_data for config drive.
Will bring up a LAMP server.
"""
content = open('user_data', 'r')
user_data_content = content.read()
image = self.nova.images.find(name="cirros-0.3.1-x86_64-uec")
flavor = self.nova.flavors.find(name="m1.tiny")
server = self.nova.servers.create(name="cirros-test", image=image.id, flavor=flavor.id,
userdata=user_data_content)
instance = self.nova.servers.get(server.id)
while instance.status != 'ACTIVE' and 'ec2_id' not in instance.metadata:
sleep_if_ec2_not_mocked(10)
instance = self.nova.servers.get(server.id)
self.servers.append(instance)
"""To test the spawn method by providing a file user_data for config drive.
Will bring up a LAMP server.
"""
content = open('user_data', 'r')
user_data_content = content.read()
image = self.nova.images.find(name="cirros-0.3.1-x86_64-uec")
flavor = self.nova.flavors.find(name="m1.tiny")
server = self.nova.servers.create(name="cirros-test", image=image.id, flavor=flavor.id,
userdata=user_data_content)
instance = self.nova.servers.get(server.id)
while instance.status != 'ACTIVE' and 'ec2_id' not in instance.metadata:
self.sleep_if_ec2_not_mocked(10)
instance = self.nova.servers.get(server.id)
self.servers.append(instance)
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']], filters=None,
dry_run=False, max_results=None)
print ec2_instance
print ec2_instance[0].ip_address
#getting the public ip of the ec2 instance
url = "http://"+ec2_instance[0].ip_address+"/phpinfo.php"
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']], filters=None,
dry_run=False, max_results=None)
print ec2_instance
print ec2_instance[0].ip_address
#getting the public ip of the ec2 instance
url = "http://"+ec2_instance[0].ip_address+"/phpinfo.php"
#wait for the instance to downalod all the dependencies for a LAMP server
sleep_if_ec2_not_mocked(300)
print url
raw_response = urllib2.urlopen(url)
print raw_response
self.assertEqual(raw_response.code, 200)
#wait for the instance to downalod all the dependencies for a LAMP server
self.sleep_if_ec2_not_mocked(300)
print url
raw_response = urllib2.urlopen(url)
print raw_response
self.assertEqual(raw_response.code, 200)
@unittest.skipIf(os.environ.get('MOCK_EC2'), 'Not supported by moto')
def test_diagnostics(self):
print "******* Diagnostics Test ***********"
instance, instance_ref = self.spawn_ec2_instance()
print "instance_ref: ", instance_ref
print "******* Diagnostics Test ***********"
instance, instance_ref = self.spawn_ec2_instance()
print "instance_ref: ", instance_ref
diagnostics = instance.diagnostics()[1]
self.assertEqual(diagnostics['instance.instance_type'], 't2.micro')
self.assertEqual(diagnostics['instance._state'], 'running(16)')
diagnostics = instance.diagnostics()[1]
self.assertEqual(diagnostics['instance.instance_type'], 't2.micro')
self.assertEqual(diagnostics['instance._state'], 'running(16)')
def test_attach_volume(self):
volume = self.nova_volume.volumes.create(1, snapshot_id=None, display_name='test', display_description=None, volume_type=None, availability_zone=None, imageRef=None)
self.volumes.append(volume)
instance, instance_ref = self.spawn_ec2_instance()
self.nova.volumes.create_server_volume(instance_ref, volume.id, "/dev/sdb")
time.sleep(30)
self.sleep_if_ec2_not_mocked(30)
volumes = self.nova.volumes.get_server_volumes(instance.id)
self.assertIn(volume, 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
time.sleep(120)
for volume in self.volumes:
volume.delete()
if __name__ == '__main__':
unittest.main()