214 lines
9.4 KiB
Python
214 lines
9.4 KiB
Python
import unittest
|
|
import urllib2
|
|
|
|
from boto.exception import EC2ResponseError
|
|
from novaclient.v1_1 import client
|
|
|
|
from ...ec2driver_config import *
|
|
from tests.integration.ec2_test_base import EC2TestBase
|
|
|
|
|
|
class TestEC2Driver(EC2TestBase):
|
|
|
|
def test_spawn(self):
|
|
print "******* Spawn Test ***********"
|
|
instance, instance_ref = self.spawn_ec2_instance()
|
|
|
|
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']])
|
|
ec2_eip = self.ec2_conn.get_all_addresses(addresses=instance.metadata['public_ip_address'])[0]
|
|
|
|
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()
|
|
|
|
ec2_id = instance.metadata['ec2_id']
|
|
|
|
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":
|
|
self.sleep_if_ec2_not_mocked(10)
|
|
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[ec2_id])[0]
|
|
|
|
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
|
|
while ec2_instance.state not in ("shutting-down", "terminated"):
|
|
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"))
|
|
|
|
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 ***********"
|
|
instance, instance_ref = self.spawn_ec2_instance()
|
|
# Send poweroff to the instance
|
|
self.nova.servers.stop(instance)
|
|
|
|
while instance.status != 'SHUTOFF':
|
|
self.sleep_if_ec2_not_mocked(5)
|
|
instance = self.nova.servers.get(instance.id)
|
|
|
|
# assert power off
|
|
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']])[0]
|
|
self.assertEqual(ec2_instance.state, "stopped")
|
|
|
|
def test_soft_reboot(self):
|
|
print "******* Soft Reboot Test ***********"
|
|
instance, instance_ref = self.spawn_ec2_instance()
|
|
# Send reboot to the instance with reboot_type = 'soft'
|
|
self.nova.servers.reboot(instance, client.servers.REBOOT_SOFT)
|
|
|
|
# we are waiting for the status to actually get to 'Reboot' before
|
|
# beginning to wait for it to go to 'Active' status
|
|
while instance.status != 'REBOOT':
|
|
# We don't sleep here because the soft reboot may take less than a
|
|
# second
|
|
instance = self.nova.servers.get(instance.id)
|
|
|
|
while instance.status != 'ACTIVE':
|
|
self.sleep_if_ec2_not_mocked(5)
|
|
instance = self.nova.servers.get(instance.id)
|
|
|
|
#assert restarted
|
|
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']])[0]
|
|
self.assertEqual(ec2_instance.state, "running")
|
|
|
|
def test_hard_reboot(self):
|
|
print "******* Hard Reboot Test ***********"
|
|
instance, instance_ref = self.spawn_ec2_instance()
|
|
# Send reboot to the instance with reboot_type = 'soft'
|
|
self.nova.servers.reboot(instance, client.servers.REBOOT_HARD)
|
|
|
|
# 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':
|
|
self.sleep_if_ec2_not_mocked(5)
|
|
instance = self.nova.servers.get(instance.id)
|
|
|
|
while instance.status != 'ACTIVE':
|
|
self.sleep_if_ec2_not_mocked(5)
|
|
instance = self.nova.servers.get(instance.id)
|
|
|
|
#assert restarted
|
|
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']])[0]
|
|
self.assertEqual(ec2_instance.state, "running")
|
|
|
|
def test_resize(self):
|
|
print "******* Resize Test ***********"
|
|
instance, instance_ref = self.spawn_ec2_instance()
|
|
|
|
ec2_instance = self.ec2_conn.get_only_instances(instance_ids=[instance.metadata['ec2_id']])[0]
|
|
|
|
ip_before_resize = self.ec2_conn.get_all_addresses(addresses=instance.metadata['public_ip_address'])[0]
|
|
|
|
self.assertEqual(ec2_instance.instance_type, "t2.micro")
|
|
|
|
new_flavor = self.nova.flavors.find(name="m1.small")
|
|
|
|
# Resize instance with flavor = m1.small
|
|
self.nova.servers.resize(instance, new_flavor)
|
|
|
|
# wait for the status to actually go to Verify_Resize, before
|
|
# confirming the resize.
|
|
while instance.status != 'VERIFY_RESIZE':
|
|
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':
|
|
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]
|
|
# ip_after_resize = ec2_instance.ip_address
|
|
ip_after_resize = self.ec2_conn.get_all_addresses(addresses=instance.metadata['public_ip_address'])[0]
|
|
|
|
self.assertEqual(ec2_instance.instance_type, "t2.small")
|
|
|
|
self.assertEqual(ip_before_resize.public_ip, ip_after_resize.public_ip)
|
|
|
|
@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:
|
|
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"
|
|
|
|
#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
|
|
|
|
diagnostics = instance.diagnostics()[1]
|
|
self.assertEqual(diagnostics['instance.instance_type'], 't2.micro')
|
|
self.assertEqual(diagnostics['instance._state'], 'running(16)')
|
|
|
|
@unittest.skipIf(os.environ.get('MOCK_EC2'), 'Not supported by moto')
|
|
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")
|
|
self.sleep_if_ec2_not_mocked(30)
|
|
volumes = self.nova.volumes.get_server_volumes(instance.id)
|
|
self.assertIn(volume, volumes)
|
|
|
|
@unittest.skipIf(os.environ.get('MOCK_EC2'), 'Not supported by moto')
|
|
def test_detach_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")
|
|
self.sleep_if_ec2_not_mocked(30)
|
|
volumes = self.nova.volumes.get_server_volumes(instance.id)
|
|
if volume in volumes:
|
|
self.nova.volumes.delete_server_volume(instance_ref, volume.id)
|
|
self.sleep_if_ec2_not_mocked(30)
|
|
volumes = self.nova.volumes.get_server_volumes(instance.id)
|
|
self.assertNotIn(volume, volumes)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|