diff --git a/ec2api/tests/functional/api/test_customer_gateways.py b/ec2api/tests/functional/api/test_customer_gateways.py index 41ee24a3..8d1a7352 100644 --- a/ec2api/tests/functional/api/test_customer_gateways.py +++ b/ec2api/tests/functional/api/test_customer_gateways.py @@ -31,6 +31,7 @@ class CustomerGatewayTest(base.EC2TestCase): super(CustomerGatewayTest, cls).setUpClass() if not base.TesterStateHolder().get_vpc_enabled(): raise cls.skipException('VPC is disabled') + base.check_network_feature_enabled('vpnaas') def test_create_delete_customer_gateway(self): data = self.client.create_customer_gateway( diff --git a/ec2api/tests/functional/api/test_tags.py b/ec2api/tests/functional/api/test_tags.py index 8645fc1e..2733684c 100644 --- a/ec2api/tests/functional/api/test_tags.py +++ b/ec2api/tests/functional/api/test_tags.py @@ -441,6 +441,7 @@ class TagTest(base.EC2TestCase): self._test_tag_resource(cgw_id, 'customer-gateway', describe_func) @base.skip_without_vpc() + @base.skip_without_network_feature('vpnaas') def test_tag_vpn_gateway(self): data = self.client.create_vpn_gateway(Type='ipsec.1') vgw_id = data['VpnGateway']['VpnGatewayId'] @@ -456,6 +457,7 @@ class TagTest(base.EC2TestCase): self._test_tag_resource(vgw_id, 'vpn-gateway', describe_func) @base.skip_without_vpc() + @base.skip_without_network_feature('vpnaas') def test_tag_vpn_connection(self): data = self.client.create_customer_gateway( Type='ipsec.1', PublicIp='198.51.100.77', BgpAsn=65000) diff --git a/ec2api/tests/functional/api/test_vpn_connections.py b/ec2api/tests/functional/api/test_vpn_connections.py index 105a8270..e4d2fbc5 100644 --- a/ec2api/tests/functional/api/test_vpn_connections.py +++ b/ec2api/tests/functional/api/test_vpn_connections.py @@ -35,6 +35,7 @@ class VpnConnectionTest(base.EC2TestCase): super(VpnConnectionTest, cls).setUpClass() if not base.TesterStateHolder().get_vpc_enabled(): raise cls.skipException('VPC is disabled') + base.check_network_feature_enabled('vpnaas') data = cls.client.create_customer_gateway( Type='ipsec.1', PublicIp=cls.CUSTOMER_GATEWAY_IP, BgpAsn=65000) diff --git a/ec2api/tests/functional/api/test_vpn_gateways.py b/ec2api/tests/functional/api/test_vpn_gateways.py index b92a8594..148ef320 100644 --- a/ec2api/tests/functional/api/test_vpn_gateways.py +++ b/ec2api/tests/functional/api/test_vpn_gateways.py @@ -32,6 +32,7 @@ class VpnGatewayTest(base.EC2TestCase): super(VpnGatewayTest, cls).setUpClass() if not base.TesterStateHolder().get_vpc_enabled(): raise cls.skipException('VPC is disabled') + base.check_network_feature_enabled('vpnaas') data = cls.client.create_vpc(CidrBlock=cls.VPC_CIDR) cls.vpc_id = data['Vpc']['VpcId'] diff --git a/ec2api/tests/functional/base.py b/ec2api/tests/functional/base.py index 33db4876..a24503c2 100644 --- a/ec2api/tests/functional/base.py +++ b/ec2api/tests/functional/base.py @@ -241,6 +241,24 @@ def skip_without_vpc(*args, **kwargs): return decorator +def check_network_feature_enabled(ext_name): + ext_list = CONF['network-feature-enabled']['api_extensions'] + if 'all' not in ext_list and ext_name not in ext_list: + msg = ("Skipped network test as %s is not available" % ext_name) + raise testtools.TestCase.skipException(msg) + + +def skip_without_network_feature(ext_name, *args, **kwargs): + """A decorator useful to skip tests without specified network extension.""" + def decorator(f): + @functools.wraps(f) + def wrapper(self, *func_args, **func_kwargs): + check_network_feature_enabled(ext_name) + 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.""" diff --git a/ec2api/tests/functional/scenario/test_vpn.py b/ec2api/tests/functional/scenario/test_vpn.py index 362daf6b..7b0268d4 100644 --- a/ec2api/tests/functional/scenario/test_vpn.py +++ b/ec2api/tests/functional/scenario/test_vpn.py @@ -46,6 +46,7 @@ class VpnTest(scenario_base.BaseScenarioTest): super(VpnTest, cls).setUpClass() if not base.TesterStateHolder().get_vpc_enabled(): raise cls.skipException('VPC is disabled') + base.check_network_feature_enabled('vpnaas') def test_vpn_routing(self): vpc_id, _subnet_id = self.create_vpc_and_subnet('10.42.0.0/20')