diff --git a/vmware_nsx_tempest/tests/nsxv/scenario/test_spoofguard_policy.py b/vmware_nsx_tempest/tests/nsxv/scenario/test_spoofguard_policy.py index 846586e09b..83ff0d538f 100644 --- a/vmware_nsx_tempest/tests/nsxv/scenario/test_spoofguard_policy.py +++ b/vmware_nsx_tempest/tests/nsxv/scenario/test_spoofguard_policy.py @@ -16,8 +16,10 @@ import re +from tempest.common.utils.linux import remote_client from tempest import config from tempest.lib.common.utils import data_utils +from tempest.lib.common.utils import test_utils from tempest import test from vmware_nsx_tempest._i18n import _LI @@ -70,6 +72,10 @@ class TestSpoofGuardBasicOps(dmgr.TopoDeployScenarioManager): def resource_cleanup(cls): super(TestSpoofGuardBasicOps, cls).resource_cleanup() + def setUp(self): + super(TestSpoofGuardBasicOps, self).setUp() + self.keypairs = {} + def tearDown(self): self.remove_project_network() super(TestSpoofGuardBasicOps, self).tearDown() @@ -97,35 +103,63 @@ class TestSpoofGuardBasicOps(dmgr.TopoDeployScenarioManager): host_ip = serv_addr['addr'] self.waitfor_host_connected(host_ip) - def setup_vm_enviornment(self, client_mgr, t_id, + def _get_server_key(self, server): + return self.keypairs[server['key_name']]['private_key'] + + def _create_sec_group(self): + # Create security group + sg_name = data_utils.rand_name('disable-spoof') + sg_desc = sg_name + " description" + secgroup = self.compute_security_groups_client.create_security_group( + name=sg_name, description=sg_desc)['security_group'] + self.addCleanup( + test_utils.call_and_ignore_notfound_exc, + self.compute_security_groups_client.delete_security_group, + secgroup['id']) + + def setup_vm_environment(self, client_mgr, t_id, check_outside_world=True, cidr_offset=0): + sg_name = data_utils.rand_name('disable-spoof') + sg_desc = sg_name + " description" + t_security_group = \ + self.compute_security_groups_client.create_security_group( + name=sg_name, description=sg_desc)['security_group'] + self.addCleanup( + test_utils.call_and_ignore_notfound_exc, + self.compute_security_groups_client.delete_security_group, + t_security_group['id']) + rule = {'direction': 'ingress', 'protocol': 'tcp'} + self._create_security_group_rule(secgroup=t_security_group, **rule) + rule = {'direction': 'ingress'} + rule_id = self._create_security_group_rule(secgroup=t_security_group, + **rule)['id'] + keypair = self.create_keypair() + self.keypairs[keypair['name']] = keypair t_network, t_subnet, t_router = self.setup_project_network( self.public_network_id, namestart=("deploy-%s-spoofuard" % t_id), cidr_offset=0) - t_security_group = self._create_security_group( - security_groups_client=self.security_groups_client, - security_group_rules_client=self.security_group_rules_client, - namestart='adm') username, password = self.get_image_userpass() security_groups = [{'name': t_security_group['id']}] + key_name = keypair['name'] t_serv1 = self.create_server_on_network( - t_network, security_groups, + t_network, security_groups, key_name=key_name, image=self.get_server_image(), flavor=self.get_server_flavor(), name=t_network['name']) self.check_server_connected(t_serv1) t_floatingip = self.create_floatingip_for_server( - t_serv1, client_mgr=self.admin_manager) + t_serv1, client_mgr=client_mgr) msg = ("Associate t_floatingip[%s] to server[%s]" % (t_floatingip, t_serv1['name'])) self._check_floatingip_connectivity( t_floatingip, t_serv1, should_connect=True, msg=msg) - vm_enviornment = dict(security_group=t_security_group, + vm_environment = dict(security_group=t_security_group, network=t_network, subnet=t_subnet, router=t_router, client_mgr=client_mgr, - serv1=t_serv1, fip1=t_floatingip) - return vm_enviornment + serv1=t_serv1, fip1=t_floatingip, + rule_id=rule_id) + return vm_environment def get_port_id(self, port_client, vm_info): tenant_name = vm_info['name'] @@ -143,7 +177,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps): @test.idempotent_id('2804f55d-3221-440a-9fa8-ab16a8932634') def test_exclude_list_with_new_attach_port(self): port_client = self.manager.ports_client - self.green = self.setup_vm_enviornment(self.manager, 'green', True) + self.green = self.setup_vm_environment(self.manager, 'green', True) vm_id = self.green['serv1']['id'] net_id = self.green['network']['id'] name = 'disabled-port-security-port' @@ -173,13 +207,87 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps): # Detach interface from vm self.interface_client.delete_interface(vm_id, port_id) + @test.attr(type='nsxv') + @test.idempotent_id('a5420350-2658-47e4-9e2b-490b200e9f41') + def test_spoofguard_with_ping_between_servers_on_same_network(self): + username, password = self.get_image_userpass() + image = self.get_server_image() + flavor = self.get_server_flavor() + port_client = self.manager.ports_client + self.green = self.setup_vm_environment(self.manager, 'green', True) + security_groups = [{'name': self.green['security_group']['id']}] + # Boot instance vm2 + keypair = self.create_keypair() + self.keypairs[keypair['name']] = keypair + key_name = keypair['name'] + t_serv2 = self.create_server_on_network( + self.green['network'], security_groups, + key_name=key_name, + image=image, + flavor=flavor, + name=self.green['network']['name']) + self.check_server_connected(t_serv2) + t_floatingip2 = self.create_floatingip_for_server( + t_serv2, client_mgr=self.manager) + msg = ("Associate t_floatingip[%s] to server[%s]" + % (t_floatingip2, t_serv2['name'])) + self._check_floatingip_connectivity( + t_floatingip2, t_serv2, should_connect=True, msg=msg) + public_ip_vm_1 = self.green['fip1']['floating_ip_address'] + public_ip_vm_2 = t_floatingip2['floating_ip_address'] + private_ip_vm_1 = \ + self.green['fip1']['fixed_ip_address'] + private_ip_vm_2 = \ + t_floatingip2['fixed_ip_address'] + private_key_1 = self._get_server_key(self.green['serv1']) + client1 = remote_client.RemoteClient(public_ip_vm_1, username=username, + pkey=private_key_1, + password=password) + private_key_2 = self._get_server_key(t_serv2) + client2 = remote_client.RemoteClient(public_ip_vm_2, username=username, + pkey=private_key_2, + password=password) + self.assertEqual(True, dmgr.is_reachable(client1, private_ip_vm_2), + "Destination is reachable") + port1_id = self.green['fip1']['port_id'] + # Update vm1 port to disbale port security + port_client.update_port( + port_id=port1_id, + port_security_enabled='false') + self.compute_security_group_rules_client.\ + delete_security_group_rule(self.green['rule_id']) + self.assertEqual(False, dmgr.is_reachable(client1, private_ip_vm_2), + "Destination is not reachable") + self.assertEqual(True, dmgr.is_reachable(client2, private_ip_vm_1), + "Destination is reachable") + + def create_port(self, network_id): + port_client = self.manager.ports_client + return HELO.create_port(self, network_id=network_id, + client=port_client) + + def create_network_subnet_with_cidr(self, client_mgr=None, + tenant_id=None, name=None, cidr=None): + client_mgr = client_mgr or self.manager + tenant_id = tenant_id + name = name or data_utils.rand_name('topo-deploy-network') + net_network = self.create_network( + client=client_mgr.networks_client, + tenant_id=tenant_id, name=name) + cidr_offset = 16 + net_subnet = self.create_subnet( + client=client_mgr.subnets_client, + network=net_network, + cidr=cidr, cidr_offset=cidr_offset, name=net_network['name']) + return net_network, net_subnet + @test.attr(type='nsxv') @test.idempotent_id('38c213df-bfc2-4681-9c9c-3a31c05b0e6f') def test_exclude_with_multiple_vm(self): image = self.get_server_image() flavor = self.get_server_flavor() port_client = self.manager.ports_client - self.green = self.setup_vm_enviornment(self.manager, 'green', True) + self.green = self.setup_vm_environment(self.manager, 'green', True) vm_id = self.green['serv1']['id'] security_groups = [{'name': self.green['security_group']['id']}] # Boot instance vm2 @@ -244,7 +352,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps): @test.idempotent_id('f034d3e9-d717-4bcd-8e6e-18e9ada7b81a') def test_exclude_list_with_single_vm_port(self): port_client = self.manager.ports_client - self.green = self.setup_vm_enviornment(self.manager, 'green', True) + self.green = self.setup_vm_environment(self.manager, 'green', True) port_id = self.green['fip1']['port_id'] # Update vm port to disable port security port_client.update_port( @@ -301,7 +409,7 @@ class TestSpoofGuardFeature(TestSpoofGuardBasicOps): @test.idempotent_id('c8683cb7-4be5-4670-95c6-344a0aea3667') def test_exclude_list_with_multiple_ports(self): port_client = self.manager.ports_client - self.green = self.setup_vm_enviornment(self.manager, 'green', True) + self.green = self.setup_vm_environment(self.manager, 'green', True) vm_id = self.green['serv1']['id'] net_id = self.green['network']['id'] name = 'disabled-port-security-port1'