# Copyright 2019 VMware Inc # All Rights Reserved # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import time from oslo_utils import uuidutils from tempest import config from tempest.lib.common.utils import data_utils from tempest.lib.common.utils import test_utils from tempest.lib import decorators from tempest.lib import exceptions from tempest import test from vmware_nsx_tempest_plugin.common import constants from vmware_nsx_tempest_plugin.lib import feature_manager from vmware_nsx_tempest_plugin.services import nsxp_client from vmware_nsx_tempest_plugin.services import nsxv3_client LOG = constants.log.getLogger(__name__) CONF = config.CONF class OctaviaRoundRobin(feature_manager.FeatureManager): """Base class to support LBaaS ROUND-ROBIN test. It provides the methods to create loadbalancer network, and start web servers. Default lb_algorithm is ROUND_ROBIND. """ @classmethod def setup_clients(cls): super(OctaviaRoundRobin, cls).setup_clients() cls.cmgr_adm = cls.get_client_manager('admin') cls.cmgr_alt = cls.get_client_manager('alt') cls.cmgr_adm = cls.get_client_manager('admin') @classmethod def skip_checks(cls): super(OctaviaRoundRobin, cls).skip_checks() cfg = CONF.network if not test.is_extension_enabled('lbaasv2', 'network'): msg = 'lbaasv2 extension is not enabled.' raise cls.skipException(msg) if not (cfg.project_networks_reachable or cfg.public_network_id): msg = ('Either project_networks_reachable must be "true", or ' 'public_network_id must be defined.') raise cls.skipException(msg) @classmethod def resource_setup(cls): super(OctaviaRoundRobin, cls).resource_setup() cls.nsx = nsxv3_client.NSXV3Client(CONF.nsxv3.nsx_manager, CONF.nsxv3.nsx_user, CONF.nsxv3.nsx_password) cls.nsxp = nsxp_client.NSXPClient(CONF.nsxv3.nsx_manager, CONF.nsxv3.nsx_user, CONF.nsxv3.nsx_password) @classmethod def setup_credentials(cls): # Ask framework to not create network resources for these tests. cls.set_network_resources() super(OctaviaRoundRobin, cls).setup_credentials() def setUp(self): super(OctaviaRoundRobin, self).setUp() CONF.validation.ssh_shell_prologue = '' self.vip_ip_address = '' self.namestart = 'lbaas-ops' self.poke_counters = 12 self.hm_delay = 4 self.hm_max_retries = 3 self.hm_timeout = 10 self.server_names = [] self.loadbalancer = None self.vip_fip = None self.web_service_start_delay = 2.5 def tearDown(self, clean=None): if self.vip_fip: LOG.debug("tearDown lbass vip fip") self.disassociate_floatingip(self.vip_fip, and_delete=True) if self.loadbalancer: LOG.debug("tearDown lbass") lb_id = self.loadbalancer['id'] self.delete_octavia_lb_resources(lb_id) cert_file = open(constants.CERT_FILE, "r") cert_content = cert_file.read() self.check_certificate_at_backend(should_present=False, cert_conent=cert_content) LOG.debug("tearDown lbaas exiting...") super(OctaviaRoundRobin, self).tearDown() def deploy_octavia_topology(self, no_of_servers=2, image_id=None, slaac=False): kwargs = {'name': "router_lbaas", 'external_gateway_info': {"network_id": CONF.network.public_network_id}} router_lbaas = self.cmgr_adm.routers_client.create_router(**kwargs) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.routers_client.delete_router, router_lbaas['router']['id']) networks_client = self.cmgr_adm.networks_client name = "network_lbaas_1" network_lbaas_1 = self.\ create_topology_network(name, networks_client=networks_client) sec_rule_client = self.cmgr_adm.security_group_rules_client sec_client = self.cmgr_adm.security_groups_client kwargs = dict(tenant_id=network_lbaas_1['tenant_id'], security_group_rules_client=sec_rule_client, security_groups_client=sec_client) self.sg = self.create_topology_security_group(**kwargs) lbaas_rules = [dict(direction='ingress', protocol='tcp', port_range_min=constants.HTTP_PORT, port_range_max=constants.HTTP_PORT, ), dict(direction='ingress', protocol='tcp', port_range_min=443, port_range_max=443, )] t_id = network_lbaas_1['tenant_id'] for rule in lbaas_rules: self.add_security_group_rule(self.sg, rule, secclient=sec_client, ruleclient=sec_rule_client, tenant_id=t_id) if slaac: address_cidr = CONF.network.project_network_v6_cidr address_prefixlen = CONF.network.project_network_v6_mask_bits if ((address_prefixlen >= 126)): msg = ("Subnet %s isn't large" % address_cidr) raise exceptions.InvalidConfiguration(msg) body = {'ip_version': 6, 'ipv6_ra_mode': 'slaac', 'ipv6_address_mode': 'slaac', 'cidr': '2001:db8::/64', "network_id": network_lbaas_1['id'], 'allocation_pools': [{ 'start': str(address_cidr).split('/')[0] + '2', 'end': str(address_cidr).split('/')[0] + '70'}]} create_floating_ip = False else: body = {"network_id": network_lbaas_1['id'], "allocation_pools": [{"start": "2.0.0.2", "end": "2.0.0.254"}], "ip_version": 4, "cidr": "2.0.0.0/24"} create_floating_ip = True subnet_client = self.cmgr_adm.subnets_client subnet_lbaas = subnet_client.create_subnet(**body) self.addCleanup(test_utils.call_and_ignore_notfound_exc, subnet_client.delete_subnet, subnet_lbaas['subnet']['id']) self.cmgr_adm.routers_client.\ add_router_interface(router_lbaas['router']['id'], subnet_id=subnet_lbaas['subnet']['id']) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.routers_client.remove_router_interface, router_lbaas['router']['id'], subnet_id=subnet_lbaas['subnet']['id']) for instance in range(0, no_of_servers): self.create_topology_instance( "server_lbaas_%s" % instance, [network_lbaas_1], security_groups=[{'name': self.sg['name']}], image_id=image_id, clients=self.cmgr_adm, create_floating_ip=create_floating_ip) return dict(router=router_lbaas, subnet=subnet_lbaas, network=network_lbaas_1) @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_subnet_id_rr(self): """ This testcase creates an octavia Loadbalancer with vip-subnet-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_net_id_rr(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_net_id=net_id) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_net_id_LC(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_net_id=net_id) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('d6bd0657-6867-4b7a-8704-3844b11b1a34') def test_octavia_lb_vip_route_with_tenant_net(self): """ This testcase creates an octavia Loadbalancer with tenant net and create floating ip for lb vip, and check advertised route at tier1 backend """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] lb_dict = self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_net_id=net_id) self.check_project_lbaas() vip_address = lb_dict['vip_address'] vip_fip = lb_dict['vip_fip'] router_state_1 = diction['router']['router'] nsx_router_nat_rules = self.nsx.get_logical_router_nat_rule_ips( router_state_1['name'], router_state_1['id']) route_present = False for advertised_net in nsx_router_nat_rules['advertisedNetworks']: if len(advertised_net['networks']) > 0: if vip_address in advertised_net[ 'networks'][0]['network'] and\ advertised_net['networks'][ 0]['advertiseRouteType'] == 'T1_LB_VIP' and\ advertised_net['networks'][0]['advertiseAllow']: route_present = True self.assertEqual(True, route_present, 'Lb vip route is not advertised') kwargs = dict(port_id=None) self.cmgr_adm.floating_ips_client.\ update_floatingip(vip_fip['id'], **kwargs)['floatingip'] time.sleep(30) nsx_router_nat_rules = self.nsx.get_logical_router_nat_rule_ips( router_state_1['name'], router_state_1['id']) vip_address = vip_fip['fixed_ip_address'] route_present = False for advertised_net in nsx_router_nat_rules['advertisedNetworks']: if len(advertised_net['networks']) > 0: if vip_address in advertised_net[ 'networks'][0]['network'] and\ advertised_net['networks'][ 0]['advertiseRouteType'] == 'T1_LB_VIP' and\ advertised_net['networks'][0]['advertiseAllow']: route_present = True self.assertEqual(False, route_present, 'Lb vip route is advertised') @decorators.attr(type='nsxv3') @decorators.idempotent_id('d6bd0657-7078-4b7a-8704-3844b11b1a34') def test_octavia_lb_vip_route_with_external_net(self): """ This testcase creates an octavia Loadbalancer with external net and check lb vip route should get advertised at tier1 backend """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) network = self.cmgr_adm.networks_client.show_network( CONF.network.public_network_id)['network'] lb_dict = self.create_project_octavia( protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", external_subnet=network['subnets'][0], external=True) self.check_project_lbaas() vip_address = lb_dict['vip_address'] router_state_1 = diction['router']['router'] nsx_router_nat_rules = self.nsx.get_logical_router_nat_rule_ips( router_state_1['name'], router_state_1['id']) route_present = False for advertised_net in nsx_router_nat_rules['advertisedNetworks']: if len(advertised_net['networks']) > 0: if vip_address in advertised_net[ 'networks'][0]['network'] and\ advertised_net['networks'][ 0]['advertiseRouteType'] == 'T1_LB_VIP' and\ advertised_net['networks'][0]['advertiseAllow']: route_present = True self.assertEqual(True, route_present, 'Lb vip route is not advertised') @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_subnet_id_LC(self): """ This testcase creates an octavia Loadbalancer with vip-subnet-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_subnet_id=subnet_id) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_subnet_id(self): """ This testcase creates an octavia Loadbalancer with vip-subnet-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_subnet_id=subnet_id) self.check_project_lbaas() get_lb = self.octavia_admin_client.list_octavia_load_balancers() lb_id = get_lb['loadbalancers'][0]['id'] stat = self.octavia_admin_client.\ show_octavia_load_balancer_stats(lb_id) assert (stat['stats']['bytes_in'] == 0 and stat['stats']['bytes_out'] == 0) self.check_lbaas_project_weight_values(constants.NO_OF_VMS_2) stat = self.octavia_admin_client.\ show_octavia_load_balancer_stats(lb_id) assert (stat['stats']['bytes_in'] >= 0 and stat['stats']['bytes_out'] >= 0) @decorators.attr(type='nsxv3') @decorators.idempotent_id('60e9ecaf-b8d6-48a9-b0d2-942e5bb38f38') def test_octavia_http_round_robin_with_session_persistence(self): """ To verify the server count for LB pool with SOURCE_IP session persistence and ROUND_ROBIN lb-algorithm expected outcome is only one server responds to the client requests """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, persistence=True, persistence_type="SOURCE_IP") self.check_lbaas_project_weight_values(constants.NO_OF_VMS_2, hash_persistence=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('60e9ecaf-b8d6-48a9-b0d2-942e5bb38f38') def test_octavia_http_round_robin_with_net_id_session_persistence(self): """ To verify the server count for LB pool with SOURCE_IP session persistence and ROUND_ROBIN lb-algorithm, expected outcome is only one server responds to the client requests. """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_net_id=net_id, persistence=True, persistence_type="SOURCE_IP") self.check_lbaas_project_weight_values(constants.NO_OF_VMS_2, hash_persistence=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6768-4b7a-8704-3844b11b1a34') def test_create_verify_lb_rr_with_allowed_cidr_on_listener(self): """ The Loadbalancer listener is created with allowed_cidrs specified """ diction = self.deploy_octavia_topology() cidr_list = ['192.92.0.0/24', '192.94.0.0/16'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, allowed_cidrs=cidr_list) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c8546-6768-4b7a-8704-3844b11b1a34') def test_create_verify_lb_rr_with_allowed_cidr_0_0_0_0_on_listener(self): """ The Loadbalancer listener is created with allowed_cidrs specified """ diction = self.deploy_octavia_topology() cidr_list = ['0.0.0.0/0'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, allowed_cidrs=cidr_list) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4b7a-8704-3844b11b1a34') def test_create_REJECT_l7policies_listeneres(self): """ The Loadbalancer listener is created with allowed_cidrs specified """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT') @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c4368-6768-4b7a-8704-3844b11b1a34') def test_create_REDIRECT_TO_URL_l7policies_listeneres(self): """ The Loadbalancer listener is created with redirect_url l7policy with no url specified. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.assertRaises(exceptions.BadRequest, self.create_project_octavia, protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REDIRECT_TO_URL') @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c4368-6768-4b7a-8704-3844b11b1b34') def test_create_REDIRECT_TO_URL_url_l7policies_listeneres(self): """ The Loadbalancer listener is created with redirect_url l7policy with url specified. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REDIRECT_TO_URL', redirect_url='http://www.vmware.com') @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c4368-6768-4a7b-8704-3844b11b1b34') def test_create_REDIRECT_TO_POOL_l7policies_listeneres(self): """ The Loadbalancer listener is created with redirect_pool l7policy with url specified. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] lb = self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id) listener = lb['listener_id'] self.octavia_admin_l7policies_client.create_octavia_l7policies( listener_id=listener, action='REDIRECT_TO_POOL', redirect_pool_id=lb['pool_id']) self.octavia_admin_client.wait_for_load_balancer_status(lb['lb_id']) l7p = self.octavia_admin_l7policies_client.list_octavia_l7policies( lb['listener_id']) for i in l7p['l7policies']: if lb['listener_id'] == i['listener_id']: l7p_id = i['id'] self.octavia_admin_l7policies_client.delete_octavia_l7policy( l7p_id) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4a7a-8704-3844b11b1a34') def test_create_REJECT_l7policies_with_rules_REGEX_compare_type(self): """ This testcase creates a l7policy to reject particular regex i.e regex rule is added to the l7policy. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT', compare_type='REGEX', type='PATH', value='.*', l7rule=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4a7a-8704-3844b11b1a34') def test_create_REJECT_l7policies_with_rules_CONTAINS_compare_type(self): """ This testcase creates a l7policy to reject particular regex i.e regex rule is added to the l7policy. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT', compare_type='CONTAINS', type='PATH', value='.*', l7rule=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4a7a-8704-3844b11b1a34') def test_create_REJECT_l7policies_with_rules_ENDS_WITH_compare_type(self): """ This testcase creates a l7policy to reject particular regex i.e regex rule is added to the l7policy. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT', compare_type='ENDS_WITH', type='PATH', value='*', l7rule=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4a7a-8704-3844b11b1a34') def test_create_REJECT_l7policies_with_rules_STARTS_WITH_cmpr_type(self): """ This testcase creates a l7policy to reject particular regex i.e regex rule is added to the l7policy. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.assertRaises(exceptions.BadRequest, self.create_project_octavia, protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT', compare_type='STARTS_WITH', type='PATH', value='', l7rule=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4a7a-8704-3844b11b1a34') def test_create_REJECT_l7policies_with_invert_enabled(self): """ This testcase creates a l7policy to reject particular regex i.e regex rule is added to the l7policy. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT', invert=True) self.assertIn("KeyError", self.check_project_lbaas()) @decorators.attr(type='nsxv3') @decorators.idempotent_id('74f022d6-a6ef-4458-96a7-541deadacf99') def test_octavia_http_traffic_with_barbican_secrets(self): """ Create octavia loadbalancer with http traffic with barbican enabled. """ diction = self.deploy_octavia_topology() subnet_id = diction['subnet']['subnet']['id'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) barbican_secrets = self.create_barbican_secret_conatainer( constants.CERT_FILE, constants.KEY_FILE) barbican_container = barbican_secrets['secret_container'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, hm_type='HTTP', member_count=2, weight=5, pool_protocol='HTTP', pool_port='80', barbican_container=barbican_container, count=0, barbican=True, delay=self.hm_delay, max_retries=self.hm_max_retries, timeout=self.hm_timeout) self.check_lbaas_project_weight_values(barbican_http=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('74f022d6-a6ef-4458-96a7-541deadacf99') def test_octavia_https_traffic_with_barbican_secrets(self): """ Create octavia loadbalancer with http traffic with barbican enabled. """ diction = self.deploy_octavia_topology() subnet_id = diction['subnet']['subnet']['id'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) barbican_secrets = self.create_barbican_secret_conatainer( constants.CERT_FILE, constants.KEY_FILE) barbican_container = barbican_secrets['secret_container'] self.create_project_octavia(protocol_type="TERMINATED_HTTPS", protocol_port="443", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, hm_type='HTTP', member_count=2, weight=5, pool_protocol='HTTP', pool_port='80', barbican_container=barbican_container, count=0, barbican=True, delay=self.hm_delay, max_retries=self.hm_max_retries, timeout=self.hm_timeout) self.check_lbaas_project_weight_values(HTTPS=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('d6bd0657-7078-5c8b-8704-3844b11b1a34') def test_octavia_multiple_listeners_with_secrets(self): """ Create multiple terminated https protocol based listener with same loadbalancer """ diction = self.deploy_octavia_topology() subnet_id = diction['subnet']['subnet']['id'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) barbican_secrets = self.create_barbican_secret_conatainer( constants.CERT_FILE, constants.KEY_FILE) barbican_container = barbican_secrets['secret_container'] protocol_type = "TERMINATED_HTTPS" protocol_port = 443 lb_name = data_utils.rand_name("tempest_lb") self.loadbalancer = self.\ octavia_admin_client.\ create_octavia_load_balancer(name=lb_name, vip_subnet_id=subnet_id )['loadbalancer'] lb_id = self.loadbalancer['id'] self.octavia_admin_client.wait_for_load_balancer_status(lb_id) for i in range(1, 20): tls_id = barbican_container["container_ref"] self.listener = self.octavia_admin_listener_client.\ create_octavia_listener(loadbalancer_id=lb_id, protocol=protocol_type, protocol_port=protocol_port, name=lb_name, allowed_cidrs=None, default_tls_container_ref=tls_id )['listener'] self.octavia_admin_client.wait_for_load_balancer_status(lb_id) protocol_port = protocol_port + 1 @decorators.idempotent_id('d6bd0657-7078-5c8b-0815-3844b11b1a34') def test_octavia_create_listener_with_empty_secrets(self): """ Try to create listener with terminated https protocol and empty secret , it should fail. """ secret_name1 = data_utils.rand_name(name='tempest-cert-secret') kwargs = {"secret_type": constants.SECRET_TYPE, "algorithm": constants.ALGORITHM, "name": secret_name1} barbican_secret1 = self.create_barbican_secret(**kwargs) secret_name2 = data_utils.rand_name(name='tempest-key-secret') kwargs = {"secret_type": constants.SECRET_TYPE, "algorithm": constants.ALGORITHM, "name": secret_name2} barbican_secret2 = self.create_barbican_secret(**kwargs) container_name = data_utils.rand_name(name='tempest-container') kwargs = {"type": constants.CONTAINER_TYPE, "name": container_name, "secret_refs": [{"secret_ref": barbican_secret1['secret_ref'], "name": 'certificate'}, {"secret_ref": barbican_secret2['secret_ref'], "name": 'private_key'}]} barbican_container = self.create_barbican_container(**kwargs) diction = self.deploy_octavia_topology() subnet_id = diction['subnet']['subnet']['id'] protocol_type = "TERMINATED_HTTPS" protocol_port = 443 lb_name = data_utils.rand_name("tempest_lb") self.loadbalancer = self.\ octavia_admin_client.\ create_octavia_load_balancer(name=lb_name, vip_subnet_id=subnet_id )['loadbalancer'] lb_id = self.loadbalancer['id'] self.octavia_admin_client.wait_for_load_balancer_status(lb_id) self.assertRaises(exceptions.BadRequest, self.octavia_admin_listener_client. create_octavia_listener, loadbalancer_id=lb_id, protocol=protocol_type, protocol_port=protocol_port, name=lb_name, allowed_cidrs=None, default_tls_container_ref=barbican_container[ 'container_ref']) @decorators.attr(type='nsxv3') @decorators.idempotent_id('d6bd0657-7078-5c8b-0815-4055c21b1a34') def test_octavia_check_certificate_at_backend(self): """ Create octavia loadbalancer with http traffic with barbican enabled. """ diction = self.deploy_octavia_topology() subnet_id = diction['subnet']['subnet']['id'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) barbican_secrets = self.create_barbican_secret_conatainer( constants.CERT_FILE, constants.KEY_FILE) barbican_container = barbican_secrets['secret_container'] self.create_project_octavia( protocol_type="TERMINATED_HTTPS", protocol_port="443", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, hm_type='HTTP', member_count=2, weight=5, pool_protocol='HTTP', pool_port='80', barbican_container=barbican_container, count=0, barbican=True, delay=self.hm_delay, max_retries=self.hm_max_retries, timeout=self.hm_timeout) self.check_lbaas_project_weight_values(HTTPS=True) cert_file = open(constants.CERT_FILE, "r") cert_content = cert_file.read() self.check_certificate_at_backend(cert_conent=cert_content.rstrip()) @decorators.attr(type='nsxv3') @decorators.idempotent_id('d6bd0657-7078-5c8b-0815-4055c21c2b45') def test_octavia_create_lbaas_listener_with_invalid_container_uuid(self): """ Create octavia loadbalancer with http traffic with barbican enabled. """ diction = self.deploy_octavia_topology() subnet_id = diction['subnet']['subnet']['id'] if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) barbican_secrets = self.create_barbican_secret_conatainer( constants.CERT_FILE, constants.KEY_FILE) container_ref = barbican_secrets["secret_container"]['container_ref']\ .split('/') container_ref.remove(container_ref[len(container_ref) - 1]) container_ref.append(uuidutils.generate_uuid()) container_ref = '/'.join(str(e) for e in container_ref) protocol_type = "TERMINATED_HTTPS" protocol_port = 443 lb_name = data_utils.rand_name("tempest_lb") self.loadbalancer = self.\ octavia_admin_client.\ create_octavia_load_balancer(name=lb_name, vip_subnet_id=subnet_id )['loadbalancer'] lb_id = self.loadbalancer['id'] self.octavia_admin_client.wait_for_load_balancer_status(lb_id) self.assertRaises(exceptions.BadRequest, self.octavia_admin_listener_client. create_octavia_listener, loadbalancer_id=lb_id, protocol=protocol_type, protocol_port=protocol_port, name=lb_name, allowed_cidrs=None, default_tls_container_ref=container_ref) @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_net_id_LC_hm_ping(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_net_id_LC_hm_ping_https(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTPS", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_net_id_RR_hm_ping(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1a34') def test_create_verify_octavia_lb_with_vip_net_id_SOURCE_IP_hm_ping(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1b43') def test_verify_octavia_lb_vip_net_id_ROUND_ROBIN_default_pool(self): """ This testcase is for verifying the loadbalancer with net-id and the pool is created using lb option and attached to a listener with default-pool option """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11a1b34') def test_verify_octavia_lb_vip_subnet_id_rr_default_pool(self): """ This testcase is for verifying the loadbalancer with subnet-id and the pool is created using lb option and attached to a listener with default-pool option """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3843a11b1a34') def test_verify_octavia_lb_port_id_rr_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3843a11a2b34') def test_verify_octavia_lb_port_id_lc_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option, lease connection algorithm """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3843a12b1a34') def test_verify_octavia_lb_port_id_source_ip_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option, source_ip algorithm. """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="SOURCE_IP", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-5677-4b7a-8704-3843a12b1a34') def test_verify_octavia_https_lb_port_id_source_ip_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option, source_ip algorithm, https protocol. """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTPS", protocol_port="80", lb_algorithm="SOURCE_IP", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8544-3843a11b1a34') def test_verify_octavia_https_lb_port_id_rr_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option, https traffic """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTPS", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8544-3843a11b1a34') def test_verify_octavia_https_lb_port_id_lc_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option,least connections https traffic """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTPS", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b21b1a34') def test_create_verify_octavia_lb_with_vip_net_id_qos_lc_hm_ping(self): """ This testcase creates an octavia Loadbalancer with vip-net-ip option, and verifies the traffic on the loadbalancer vip """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] policy = self.cmgr_adm.qos_client.create_qos_policy( name='test-policy', description='test policy desc1', shared=False) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.qos_client.delete_qos_policy, policy['policy']['id']) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, qos_policy_id=policy['policy']['id']) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8544-3843a11b1a34') def test_verify_octavia_https_lb_port_id_lc_default_pool_qos(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option,least connections https traffic """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] policy = self.cmgr_adm.qos_client.create_qos_policy( name='test-policy', description='test policy desc1', shared=False) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.qos_client.delete_qos_policy, policy['policy']['id']) port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="LEAST_CONNECTIONS", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True, qos_policy_id=policy['policy']['id']) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8544-3843a11b1a23') def test_verify_octavia_https_lb_port_id_rr_default_pool_qos(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option,round_robin https traffic """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] policy = self.cmgr_adm.qos_client.create_qos_policy( name='test-policy', description='test policy desc1', shared=False) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.qos_client.delete_qos_policy, policy['policy']['id']) port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True, qos_policy_id=policy['policy']['id']) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8544-3843a11b1a24') def test_verify_octavia_https_lb_port_id_source_ip_default_pool_qos(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option,SOURCE_IP https traffic """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] policy = self.cmgr_adm.qos_client.create_qos_policy( name='test-policy', description='test policy desc1', shared=False) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.qos_client.delete_qos_policy, policy['policy']['id']) port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="SOURCE_IP", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True, qos_policy_id=policy['policy']['id']) self.check_project_lbaas() @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-5677-4b7a-8704-3843a12b1a98') def test_verify_octavia_https_lb_port_id_source_ipv6_default_pool(self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option, source_ip algorithm, https protocol. """ diction = self.deploy_octavia_topology(slaac=True) net_id = diction['network']['id'] port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTPS", protocol_port="80", lb_algorithm="SOURCE_IP", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True, create_fip=False) @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8704-3844b11b1b43') def test_ipv6__verify_octavia_lb_vip_net_id_ROUND_ROBIN_default_pool(self): """ This testcase is for verifying the loadbalancer with net-id and the pool is created using lb option and attached to a listener with default-pool option """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_net_id=net_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True, fip_disassociate=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('c5ac8546-6867-4b7a-8544-3843a11b1a24') def test_ipv6_verify_octavia_https_lb_port_id_source_ip_default_pool_qos( self): """ This testcase is for verifying the loadbalancer with port-id and the pool is created using lb option and attached to a listener with default-pool option,SOURCE_IP https traffic """ diction = self.deploy_octavia_topology() self.start_web_servers(constants.HTTP_PORT) net_id = diction['network']['id'] policy = self.cmgr_adm.qos_client.create_qos_policy( name='test-policy', description='test policy desc1', shared=False) self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.qos_client.delete_qos_policy, policy['policy']['id']) port_id = self.cmgr_adm.ports_client.create_port( network_id=net_id)['port']['id'] self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.cmgr_adm.ports_client.delete_port, port_id) self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="SOURCE_IP", vip_port_id=port_id, hm_type='PING', timeout=self.hm_timeout, max_retries=self.hm_max_retries, delay=self.hm_delay, default_pool=True, qos_policy_id=policy['policy']['id'], fip_disassociate=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c5468-6768-4b7a-8704-3844b11b1a34') def test_ipv6_create_REJECT_l7policies_listeneres(self): """ The Loadbalancer listener is created with allowed_cidrs specified """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.create_project_octavia(protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REJECT', fip_disassociate=True) @decorators.attr(type='nsxv3') @decorators.idempotent_id('ca5c4368-6768-4b7a-8704-3844b11b1a34') def test_ipv6_create_REDIRECT_TO_URL_l7policies_listeneres(self): """ The Loadbalancer listener is created with redirect_url l7policy with no url specified. """ diction = self.deploy_octavia_topology() if not CONF.nsxv3.ens: self.start_web_servers(constants.HTTP_PORT) subnet_id = diction['subnet']['subnet']['id'] self.assertRaises(exceptions.BadRequest, self.create_project_octavia, protocol_type="HTTP", protocol_port="80", lb_algorithm="ROUND_ROBIN", vip_subnet_id=subnet_id, l7policy=True, action='REDIRECT_TO_URL', fip_disassociate=True)