All testcase automation for feature Router

Dr component feature

Change-Id: I022c836a9257db0066ab0fedd032184c6e245292
This commit is contained in:
shubhamk 2019-07-19 13:11:10 +00:00
parent 2fb1aa210c
commit 18f592c506

View File

@ -11,6 +11,8 @@
# 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_log import log as logging
from tempest import config
from tempest.lib.common.utils import data_utils
@ -32,6 +34,17 @@ class TestTier1DRComponentDeployment(feature_manager.FeatureManager):
Adding test cases to test deploy tier1
on sepcific edge_cluster.
"""
@classmethod
def setup_clients(cls):
super(TestTier1DRComponentDeployment, 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')
cls.routers_client = cls.cmgr_adm.routers_client
cls.networks_client = cls.cmgr_adm.networks_client
cls.subnets_client = cls.cmgr_adm.subnets_client
cls.sec_rule_client = cls.cmgr_adm.security_group_rules_client
cls.sec_client = cls.cmgr_adm.security_groups_client
def setUp(self):
super(TestTier1DRComponentDeployment, self).setUp()
@ -96,8 +109,18 @@ class TestTier1DRComponentDeployment(feature_manager.FeatureManager):
on edge
"""
kwargs = {"enable_snat": False}
router_state = self.create_topology_router(set_gateway=True,
**kwargs)
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
set_gateway=True,
routers_client=self.cmgr_adm.routers_client, **kwargs)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
self.create_topology_subnet(
subnet_name, network_state, router_id=router_state["id"],
routers_client=self.cmgr_adm.routers_client,
subnets_client=self.cmgr_adm.subnets_client)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
@ -106,26 +129,55 @@ class TestTier1DRComponentDeployment(feature_manager.FeatureManager):
def test_tier1_sr_component_should_present(self):
"""
Check sr and dr component of router should be present
on edge
on edge, check N-S and E-W traffic
"""
kwargs = {"enable_snat": False}
router_state = self.create_topology_router(set_gateway=True,
**kwargs)
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
set_gateway=True,
routers_client=self.cmgr_adm.routers_client, **kwargs)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
self.create_topology_subnet(
subnet_name, network_state, router_id=router_state["id"],
routers_client=self.cmgr_adm.routers_client,
subnets_client=self.cmgr_adm.subnets_client)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id)}
self.routers_client.update_router(router_state['id'],
**public_network_info)
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
network_state = self.create_topology_network(network_name)
self.create_topology_subnet(subnet_name, network_state,
router_id=router_state["id"])
network_id=CONF.network.public_network_id,
enable_snat=True)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(True, result[1]['sr_present'])
security_group = self._create_security_group(
security_group_rules_client=self.cmgr_adm.
security_group_rules_client,
security_groups_client=self.cmgr_adm.security_groups_client)
image_id = self.get_glance_image_id(["cirros", "esx"])
security_groups = [{'name': security_group['name']}]
self.create_topology_instance(
"state_vm_1", [network_state],
create_floating_ip=True, image_id=image_id, clients=self.cmgr_adm,
security_groups=security_groups)
self.create_topology_instance(
"state_vm_2", [network_state],
create_floating_ip=True, image_id=image_id, clients=self.cmgr_adm,
security_groups=security_groups)
self.check_cross_network_connectivity(
network_state,
self.servers_details.get("state_vm_1").floating_ips[0],
self.servers_details.get("state_vm_1").server, should_connect=True)
self.check_cross_network_connectivity(
network_state,
self.servers_details.get("state_vm_2").floating_ips[0],
self.servers_details.get("state_vm_2").server, should_connect=True)
@decorators.idempotent_id('1206238b-02dd-1098-c228-10066ecc69e5')
def test_tier1_sr_should_create_when_service_is_enabled(self):
@ -135,10 +187,13 @@ class TestTier1DRComponentDeployment(feature_manager.FeatureManager):
"""
kwargs = {"enable_snat": False}
router_state_1 = self.create_topology_router(set_gateway=True,
routers_client=self.
cmgr_adm.routers_client,
**kwargs)
network_lbaas_1 = self.create_topology_network("network_lbaas")
sec_rule_client = self.manager.security_group_rules_client
sec_client = self.manager.security_groups_client
network_lbaas_1 = self.create_topology_network(
"network_lbaas", networks_client=self.cmgr_adm.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)
@ -149,30 +204,241 @@ class TestTier1DRComponentDeployment(feature_manager.FeatureManager):
dict(direction='ingress', protocol='tcp',
port_range_min=443, port_range_max=443, )]
for rule in lbaas_rules:
self.add_security_group_rule(self.sg, rule)
self.add_security_group_rule(
self.sg,
rule,
ruleclient=sec_rule_client,
secclient=sec_client,
tenant_id=network_lbaas_1['tenant_id'])
subnet_lbaas = self.create_topology_subnet(
"subnet_lbaas", network_lbaas_1, router_id=router_state_1["id"])
"subnet_lbaas", network_lbaas_1,
subnets_client=self.cmgr_adm.subnets_client,
routers_client=self.cmgr_adm.routers_client,
router_id=router_state_1["id"])
no_of_servers = 2
image_id = self.get_glance_image_id(["cirros", "esx"])
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, create_floating_ip=False)
image_id=image_id, create_floating_ip=False,
clients=self.cmgr_adm)
network = self.cmgr_adm.networks_client.show_network(
CONF.network.public_network_id)['network']
result = self.check_router_components_on_edge(router_state_1)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
lb_resource = self.create_project_lbaas(protocol_type="HTTP",
protocol_port="80",
vip_subnet_id=subnet_lbaas[
'id'],
lb_algorithm="ROUND_ROBIN",
hm_type='PING',
create_fip=False)
result = self.check_router_components_on_edge(router_state_1)
for i in range(0, 2):
lb_resource = self.create_project_lbaas(
protocol_type="HTTP", protocol_port="80",
vip_subnet_id=subnet_lbaas['id'],
lb_algorithm="ROUND_ROBIN", hm_type='PING',
create_fip=False, clean_up=True,
external_subnet=network['subnets'][0])
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state_1)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(True, result[1]['sr_present'])
self.delete_loadbalancer_resources(lb_resource['lb_id'],
admin=True)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state_1)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
@decorators.idempotent_id('2317349c-13ee-2198-c228-10066ecc69e5')
def test_tier1_sr_component_stail_entry_should_not_present(self):
"""
Check sr component stail entry should not present on edge after
multiple times enabling and disabling service on router.
"""
kwargs = {"enable_snat": False}
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
set_gateway=True,
routers_client=self.cmgr_adm.routers_client, **kwargs)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
self.create_topology_subnet(
subnet_name, network_state, router_id=router_state["id"],
routers_client=self.cmgr_adm.routers_client,
subnets_client=self.cmgr_adm.subnets_client)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
for i in range(1, 20):
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id,
enable_snat=True)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(True, result[1]['sr_present'])
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id,
enable_snat=False)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
@decorators.idempotent_id('2317349c-13ee-3209-d339-10066ecc69e5')
def test_sr_dr_componet_of_router_status_with_multiple_services(self):
"""
Check sr-dr component of router status
"""
kwargs = {"enable_snat": False}
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
set_gateway=True,
routers_client=self.cmgr_adm.routers_client, **kwargs)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
subnet_lbaas = self.create_topology_subnet(
subnet_name, network_state, router_id=router_state["id"],
routers_client=self.cmgr_adm.routers_client,
subnets_client=self.cmgr_adm.subnets_client)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id,
enable_snat=True)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(True, result[1]['sr_present'])
self.delete_loadbalancer_resources(lb_resource['lb_id'])
result = self.check_router_components_on_edge(router_state_1)
sec_rule_client = self.cmgr_adm.security_group_rules_client
sec_client = self.cmgr_adm.security_groups_client
kwargs = dict(tenant_id=network_state['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, )]
for rule in lbaas_rules:
self.add_security_group_rule(
self.sg,
rule,
ruleclient=sec_rule_client,
secclient=sec_client,
tenant_id=network_state['tenant_id'])
no_of_servers = 2
image_id = self.get_glance_image_id(["cirros", "esx"])
for instance in range(0, no_of_servers):
self.create_topology_instance(
"server_lbaas_%s" % instance, [network_state],
security_groups=[{'name': self.sg['name']}],
image_id=image_id, create_floating_ip=False,
clients=self.cmgr_adm)
network = self.cmgr_adm.networks_client.show_network(
CONF.network.public_network_id)['network']
lb_resource = self.create_project_lbaas(
protocol_type="HTTP", protocol_port="80",
vip_subnet_id=subnet_lbaas['id'],
lb_algorithm="ROUND_ROBIN", hm_type='PING',
create_fip=False, clean_up=True,
external_subnet=network['subnets'][0])
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(True, result[1]['sr_present'])
self.delete_loadbalancer_resources(lb_resource['lb_id'], admin=True)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(True, result[1]['sr_present'])
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id,
enable_snat=False)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
@decorators.idempotent_id('2317349c-24ff-4310-e440-21176ecc69e5')
def test_sr_component_status_after_deleting_multiple_services_once(self):
"""
Check only sr component should not present afetr
deleting multiple services at once
"""
kwargs = {"enable_snat": False}
network_name = data_utils.rand_name(name='tempest-net')
subnet_name = data_utils.rand_name(name='tempest-subnet')
router_state = self.create_topology_router(
set_gateway=True,
routers_client=self.cmgr_adm.routers_client, **kwargs)
network_state = self.create_topology_network(
network_name, networks_client=self.cmgr_adm.networks_client)
subnet_lbaas = self.create_topology_subnet(
subnet_name, network_state, router_id=router_state["id"],
routers_client=self.cmgr_adm.routers_client,
subnets_client=self.cmgr_adm.subnets_client)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id,
enable_snat=True)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
sec_rule_client = self.cmgr_adm.security_group_rules_client
sec_client = self.cmgr_adm.security_groups_client
kwargs = dict(tenant_id=network_state['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, )]
for rule in lbaas_rules:
self.add_security_group_rule(
self.sg,
rule,
ruleclient=sec_rule_client,
secclient=sec_client,
tenant_id=network_state['tenant_id'])
no_of_servers = 2
image_id = self.get_glance_image_id(["cirros", "esx"])
for instance in range(0, no_of_servers):
self.create_topology_instance(
"server_lbaas_%s" % instance, [network_state],
security_groups=[{'name': self.sg['name']}],
image_id=image_id, create_floating_ip=False,
clients=self.cmgr_adm)
network = self.cmgr_adm.networks_client.show_network(
CONF.network.public_network_id)['network']
lb_resource = self.create_project_lbaas(
protocol_type="HTTP", protocol_port="80",
vip_subnet_id=subnet_lbaas['id'],
lb_algorithm="ROUND_ROBIN", hm_type='PING',
create_fip=False, clean_up=True,
external_subnet=network['subnets'][0])
self.delete_loadbalancer_resources(lb_resource['lb_id'], admin=True)
public_network_info = {"external_gateway_info": dict(
network_id=CONF.network.public_network_id,
enable_snat=False)}
self.cmgr_adm.routers_client.update_router(router_state['id'],
**public_network_info)
time.sleep(constants.NSXP_BACKEND_SMALL_TIME_INTERVAL)
result = self.check_router_components_on_edge(router_state)
self.assertEqual(True, result[0]['dr_present'])
self.assertEqual(False, result[1]['sr_present'])