functional tests: add skip checker for absence of EC2-classic

Change-Id: I42e8b446c23a06197f7daffbb826033a87e33cb7
This commit is contained in:
Andrey Pavlov 2016-07-29 15:45:35 +03:00
parent bebfe547e5
commit b0882c6c6c
7 changed files with 80 additions and 19 deletions

View File

@ -42,6 +42,7 @@ class AddressTest(base.EC2TestCase):
data = self.client.release_address(AllocationId=id)
self.cancelResourceCleanUp(res_clean)
@base.skip_without_ec2()
def test_create_delete_standard_address(self):
data = self.client.allocate_address()
ip = data['PublicIp']
@ -183,6 +184,7 @@ class AddressTest(base.EC2TestCase):
self.client.release_address(AllocationId=id2)
self.cancelResourceCleanUp(res_clean2)
@base.skip_without_ec2()
def test_describe_standard_addresses(self):
data = self.client.allocate_address(*[], **{})
ip = data['PublicIp']
@ -306,6 +308,8 @@ class AddressTest(base.EC2TestCase):
@testtools.skipUnless(CONF.aws.image_id, "image id is not defined")
# skip this test for nova network due to bug #1607350
@base.skip_without_vpc()
# this is a correct skip
@base.skip_without_ec2()
def test_associate_disassociate_standard_addresses(self):
instance_id = self.run_instance()
@ -382,6 +386,7 @@ class AddressTest(base.EC2TestCase):
self.cancelResourceCleanUp(clean_vpc)
self.get_vpc_waiter().wait_delete(vpc_id)
@base.skip_without_ec2()
def test_disassociate_not_associated_standard_addresses(self):
data = self.client.allocate_address(Domain='standard')
ip = data['PublicIp']

View File

@ -128,9 +128,10 @@ class InstanceAttributeTest(base.EC2TestCase):
self.assertRaises('InvalidInstanceID.NotFound',
self.client.describe_instance_attribute,
InstanceId='i-0', Attribute='disableApiTermination')
self.assertRaises('InvalidParameterCombination',
self.client.describe_instance_attribute,
InstanceId=instance_id, Attribute='sourceDestCheck')
if base.TesterStateHolder().get_ec2_enabled():
self.assertRaises('InvalidParameterCombination',
self.client.describe_instance_attribute,
InstanceId=instance_id, Attribute='sourceDestCheck')
self.assertRaises('InvalidParameterValue',
self.client.modify_instance_attribute,
@ -146,12 +147,17 @@ class InstanceAttributeTest(base.EC2TestCase):
InstanceId=instance_id, Attribute='disableApiTermination',
Value='True', DisableApiTermination={'Value': False})
self.assertRaises('InvalidParameterCombination',
ex_str = ('InvalidParameterCombination'
if base.TesterStateHolder().get_ec2_enabled() else
'InvalidGroup.NotFound')
self.assertRaises(ex_str,
self.client.modify_instance_attribute,
InstanceId=instance_id, Groups=['sg-0'])
self.assertRaises('InvalidParameterCombination',
self.client.modify_instance_attribute,
InstanceId=instance_id, Attribute='sourceDestCheck', Value='False')
if base.TesterStateHolder().get_ec2_enabled():
self.assertRaises('InvalidParameterCombination',
self.client.modify_instance_attribute,
InstanceId=instance_id, Attribute='sourceDestCheck',
Value='False')
self.assertRaises('InvalidParameterValue',
self.client.reset_instance_attribute,
@ -169,9 +175,10 @@ class InstanceAttributeTest(base.EC2TestCase):
self.client.reset_instance_attribute,
InstanceId=instance_id, Attribute='instanceType')
self.assertRaises('InvalidParameterCombination',
self.client.reset_instance_attribute,
InstanceId=instance_id, Attribute='sourceDestCheck')
if base.TesterStateHolder().get_ec2_enabled():
self.assertRaises('InvalidParameterCombination',
self.client.reset_instance_attribute,
InstanceId=instance_id, Attribute='sourceDestCheck')
self.assertRaises('IncorrectInstanceState',
self.client.modify_instance_attribute,

View File

@ -54,9 +54,12 @@ class InstanceTest(base.EC2TestCase):
self.assertEqual(1, len(instances))
self.assertEqual(1, len(instances[0]['SecurityGroups']))
groups = reservations[0].get('Groups', [])
self.assertEqual(1, len(groups))
self.assertEqual(groups[0]['GroupName'],
instances[0]['SecurityGroups'][0]['GroupName'])
if base.TesterStateHolder().get_ec2_enabled():
self.assertEqual(1, len(groups))
self.assertEqual(groups[0]['GroupName'],
instances[0]['SecurityGroups'][0]['GroupName'])
else:
self.assertEqual(0, len(groups))
self.client.terminate_instances(InstanceIds=[instance_id])
self.cancelResourceCleanUp(res_clean)

View File

@ -209,7 +209,10 @@ class InstanceInVPCTest(base.EC2TestCase):
'MaxCount': 1,
'PrivateIpAddress': '10.16.1.2'
}
self.assertRaises('InvalidParameterCombination',
ex_str = ('InvalidParameterCombination'
if base.TesterStateHolder().get_ec2_enabled() else
'InvalidParameterValue')
self.assertRaises(ex_str,
self.client.run_instances, rollback_fn=_rollback,
**kwargs)

View File

@ -235,6 +235,13 @@ class SecurityGroupInVPCTest(SecurityGroupBaseTest):
class SecurityGroupEC2ClassicTest(SecurityGroupBaseTest):
@classmethod
@base.safe_setup
def setUpClass(cls):
super(SecurityGroupEC2ClassicTest, cls).setUpClass()
if not base.TesterStateHolder().get_ec2_enabled():
raise cls.skipException('EC2-classic is disabled')
def test_create_delete_security_group(self):
name = data_utils.rand_name('sgName')
desc = data_utils.rand_name('sgDesc')

30
base.py
View File

@ -189,12 +189,21 @@ class TesterStateHolder(object):
cls, *args, **kwargs)
return cls._instance
_ec2_enabled = None
_vpc_enabled = None
def get_vpc_enabled(self):
if self._vpc_enabled:
return self._vpc_enabled
def get_ec2_enabled(self):
if self._ec2_enabled is None:
self._fill_attributes()
return self._ec2_enabled
def get_vpc_enabled(self):
if self._vpc_enabled is None:
self._fill_attributes()
return self._vpc_enabled
def _fill_attributes(self):
self._ec2_enabled = False
self._vpc_enabled = False
data = self.ec2_client.describe_account_attributes()
for item in data.get('AccountAttributes', []):
@ -202,8 +211,21 @@ class TesterStateHolder(object):
for value in item['AttributeValues']:
if value['AttributeValue'] == 'VPC':
self._vpc_enabled = True
if value['AttributeValue'] == 'EC2':
self._ec2_enabled = True
return self._vpc_enabled
def skip_without_ec2(*args, **kwargs):
"""A decorator useful to skip tests if EC2-classic is not supported."""
def decorator(f):
@functools.wraps(f)
def wrapper(self, *func_args, **func_kwargs):
if not TesterStateHolder().get_ec2_enabled():
msg = "Skipped because EC2-classic is not enabled"
raise testtools.TestCase.skipException(msg)
return f(self, *func_args, **func_kwargs)
return wrapper
return decorator
def skip_without_vpc(*args, **kwargs):

View File

@ -32,6 +32,13 @@ class TagsPagingTest(scenario_base.BaseScenarioTest):
# NOTE(andrey-mp): limit for tags for one resource in amazon
TAGS_COUNT = 10
@classmethod
@base.safe_setup
def setUpClass(cls):
super(TagsPagingTest, cls).setUpClass()
if 'amazon' in CONF.aws.ec2_url:
raise cls.skipException('Paging is broken in Amazon.')
def _create_volume_and_tags(self):
data = self.client.create_volume(
Size=1, AvailabilityZone=CONF.aws.aws_zone)
@ -147,6 +154,9 @@ class VolumesPagingTest(scenario_base.BaseScenarioTest):
@base.safe_setup
def setUpClass(cls):
super(VolumesPagingTest, cls).setUpClass()
if 'amazon' in CONF.aws.ec2_url:
raise cls.skipException('Paging is broken in Amazon.')
zone = CONF.aws.aws_zone
cls.ids = list()
for dummy in xrange(0, cls.VOLUMES_COUNT):
@ -216,6 +226,9 @@ class SnapshotPagingTest(scenario_base.BaseScenarioTest):
@base.safe_setup
def setUpClass(cls):
super(SnapshotPagingTest, cls).setUpClass()
if 'amazon' in CONF.aws.ec2_url:
raise cls.skipException('Paging is broken in Amazon.')
zone = CONF.aws.aws_zone
data = cls.client.create_volume(Size=1, AvailabilityZone=zone)
@ -294,6 +307,8 @@ class InstancePagingTest(scenario_base.BaseScenarioTest):
@base.safe_setup
def setUpClass(cls):
super(InstancePagingTest, cls).setUpClass()
if 'amazon' in CONF.aws.ec2_url:
raise cls.skipException('Paging is broken in Amazon.')
if not CONF.aws.image_id:
raise cls.skipException('aws image_id does not provided')
@ -339,7 +354,6 @@ class InstancePagingTest(scenario_base.BaseScenarioTest):
self.assertEqual(max_results, self._count_instances(data))
def test_instances_paging(self):
real_count = 0
max_results = 5
kwargs = {'MaxResults': max_results}
instances = set()