openstack
/
ec2-api
Code Issues Proposed changes

Reworked mechanism of disabling VPC tests 

Change-Id: I2438610e595325919d4b3d7eb17aea67c11f0ba2
This commit is contained in:
Andrey Pavlov 2015-02-11 22:16:15 +03:00
parent 7377e3f7d9
commit 0ece6267b0
12 changed files with 88 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
ec2api/tests/contrib/post_test_hook.sh
View File

@ -23,10 +23,6 @@ export TEST_CONFIG="functional_tests.conf"
if [[ ! -f $TEST_CONFIG_DIR/$TEST_CONFIG ]]; then
IMAGE_ID=$(euca-describe-images | grep "cirros" | grep "ami-" | head -n 1 | awk '{print $2}')
VPC_ENABLED="False"
if [[ $DEVSTACK_GATE_NEUTRON -eq "1" ]]; then
VPC_ENABLED="True"
fi
sudo bash -c "cat > $TEST_CONFIG_DIR/$TEST_CONFIG <<EOF
[aws]
@ -34,7 +30,6 @@ ec2_url = $EC2_URL
aws_access = $EC2_ACCESS_KEY
aws_secret = $EC2_SECRET_KEY
image_id = $IMAGE_ID
vpc_enabled = $VPC_ENABLED
EOF"
fi

19
ec2api/tests/functional/api/test_addresses.py
View File

@ -16,7 +16,6 @@
import time
from tempest_lib.openstack.common import log
import testtools
from ec2api.tests.functional import base
from ec2api.tests.functional import config
@ -27,7 +26,7 @@ LOG = log.getLogger(__name__)
class AddressTest(base.EC2TestCase):
@testtools.skipUnless(CONF.aws.run_incompatible_tests, "VPC is disabled")
@base.skip_without_vpc()
def test_create_delete_vpc_address(self):
kwargs = {
'Domain': 'vpc',
@ -54,7 +53,7 @@ class AddressTest(base.EC2TestCase):
self.assertEqual(200, resp.status_code, base.EC2ErrorConverter(data))
self.cancelResourceCleanUp(res_clean)
@testtools.skipUnless(CONF.aws.run_incompatible_tests, "VPC is disabled")
@base.skip_without_vpc()
def test_invalid_delete_vpc_address(self):
kwargs = {
'Domain': 'vpc',
@ -106,12 +105,18 @@ class AddressTest(base.EC2TestCase):
}
resp, data = self.client.AllocateAddress(*[], **kwargs)
if resp.status_code == 200:
self.addResourceCleanUp(self.client.ReleaseAddress,
AllocationId=data['AllocationId'])
allocation_id = data.get('AllocationId')
if allocation_id:
self.addResourceCleanUp(self.client.ReleaseAddress,
AllocationId=allocation_id)
else:
public_ip = data.get('PublicIp')
self.addResourceCleanUp(self.client.ReleaseAddress,
PublicIp=public_ip)
self.assertEqual(400, resp.status_code)
self.assertEqual('InvalidParameterValue', data['Error']['Code'])
@testtools.skipUnless(CONF.aws.run_incompatible_tests, "VPC is disabled")
@base.skip_without_vpc()
def test_describe_vpc_addresses(self):
resp, data = self.client.DescribeAddresses(*[], **{})
self.assertEqual(200, resp.status_code, base.EC2ErrorConverter(data))
@ -239,7 +244,7 @@ class AddressTest(base.EC2TestCase):
self.assertEqual(200, resp.status_code, base.EC2ErrorConverter(data))
self.cancelResourceCleanUp(res_clean)
@testtools.skipUnless(CONF.aws.run_incompatible_tests, "VPC is disabled")
@base.skip_without_vpc()
def test_associate_disassociate_vpc_addresses(self):
instance_type = CONF.aws.instance_type
image_id = CONF.aws.image_id

2
ec2api/tests/functional/api/test_dhcp_options.py
View File

@ -30,7 +30,7 @@ class DhcpOptionsTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(DhcpOptionsTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
def test_create_delete_dhcp_options(self):

2
ec2api/tests/functional/api/test_instances.py
View File

@ -36,7 +36,7 @@ class InstanceTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(InstanceTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
if not CONF.aws.image_id:
raise cls.skipException('aws image_id does not provided')

2
ec2api/tests/functional/api/test_internet_gateways.py
View File

@ -36,7 +36,7 @@ class InternetGatewayTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(InternetGatewayTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
resp, data = cls.client.CreateVpc(CidrBlock=cls.VPC_CIDR)

2
ec2api/tests/functional/api/test_network_interfaces.py
View File

@ -36,7 +36,7 @@ class NetworkInterfaceTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(NetworkInterfaceTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
resp, data = cls.client.CreateVpc(CidrBlock=cls.VPC_CIDR)

2
ec2api/tests/functional/api/test_routes.py
View File

@ -32,7 +32,7 @@ class RouteTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(RouteTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
resp, data = cls.client.CreateVpc(CidrBlock=cls.VPC_CIDR)

2
ec2api/tests/functional/api/test_security_groups.py
View File

@ -35,7 +35,7 @@ class SecurityGroupTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(SecurityGroupTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
resp, data = cls.client.CreateVpc(CidrBlock=cls.VPC_CIDR)

2
ec2api/tests/functional/api/test_subnets.py
View File

@ -32,7 +32,7 @@ class SubnetTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(SubnetTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
resp, data = cls.client.CreateVpc(CidrBlock=cls.VPC_CIDR)

2
ec2api/tests/functional/api/test_vpcs.py
View File

@ -27,7 +27,7 @@ class VPCTest(base.EC2TestCase):
@base.safe_setup
def setUpClass(cls):
super(VPCTest, cls).setUpClass()
if not CONF.aws.vpc_enabled:
if not base.TesterStateHolder().get_vpc_enabled():
raise cls.skipException('VPC is disabled')
def test_create_delete_vpc(self):

89
ec2api/tests/functional/base.py
View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import functools
import logging
import random
import sys
@ -46,27 +47,6 @@ def rand_name(name=''):
return randbits
def safe_setup(f):
"""A decorator used to wrap the setUpClass for safe setup."""
def decorator(cls):
try:
f(cls)
except Exception as se:
etype, value, trace = sys.exc_info()
LOG.exception("setUpClass failed: %s" % se)
try:
cls.tearDownClass()
except Exception as te:
LOG.exception("tearDownClass failed: %s" % te)
try:
raise etype(value), None, trace
finally:
del trace # for avoiding circular refs
return decorator
class EC2ErrorConverter(object):
_data = ''
@ -92,6 +72,9 @@ class EC2ResponceException(Exception):
self.resp = resp
self.data = data
def __str__(self):
return str(self.data)
class EC2Waiter(object):
@ -185,6 +168,69 @@ class EC2Waiter(object):
interval += self.default_check_interval
def safe_setup(f):
"""A decorator used to wrap the setUpClass for safe setup."""
def decorator(cls):
try:
f(cls)
except Exception as se:
etype, value, trace = sys.exc_info()
LOG.exception("setUpClass failed: %s" % se)
try:
cls.tearDownClass()
except Exception as te:
LOG.exception("tearDownClass failed: %s" % te)
try:
raise etype(value), None, trace
finally:
del trace # for avoiding circular refs
return decorator
class TesterStateHolder(object):
ec2_client = None
_instance = None
def __new__(cls, *args, **kwargs):
if not cls._instance:
cls._instance = super(TesterStateHolder, cls).__new__(
cls, *args, **kwargs)
return cls._instance
_vpc_enabled = None
def get_vpc_enabled(self):
if self._vpc_enabled:
return self._vpc_enabled
self._vpc_enabled = False
resp, data = self.ec2_client.DescribeAccountAttributes()
if resp.status_code == 200:
for item in data.get('AccountAttributes', []):
if item['AttributeName'] == 'supported-platforms':
for value in item['AttributeValues']:
if value['AttributeValue'] == 'VPC':
self._vpc_enabled = True
return self._vpc_enabled
def skip_without_vpc(*args, **kwargs):
"""A decorator useful to skip tests if VPC is not supported."""
def decorator(f):
@functools.wraps(f)
def wrapper(self, *func_args, **func_kwargs):
if not TesterStateHolder().get_vpc_enabled():
msg = "Skipped because VPC is disabled"
raise testtools.TestCase.skipException(msg)
return f(self, *func_args, **func_kwargs)
return wrapper
return decorator
class EC2TestCase(base.BaseTestCase):
"""Recommended to use as base class for boto related test."""
@ -198,6 +244,7 @@ class EC2TestCase(base.BaseTestCase):
def setUpClass(cls):
super(EC2TestCase, cls).setUpClass()
cls.client = botocoreclient.APIClientEC2()
TesterStateHolder().ec2_client = cls.client
@classmethod
def addResourceCleanUpStatic(cls, function, *args, **kwargs):

3
ec2api/tests/functional/config.py
View File

@ -59,9 +59,6 @@ AWSGroup = [
cfg.StrOpt('image_id',
default=None,
help="Image ID for instance running"),
cfg.BoolOpt('vpc_enabled',
default=False,
help='Will run all tests that works with VPC.'),
cfg.BoolOpt('run_incompatible_tests',
default=False,
help='Will run all tests plus incompatible with Amazon.'),