How to use _check_public_network_connectivity method in tempest

Best Python code snippet using tempest_python
test_network_basic_ops.py
Source:test_network_basic_ops.py
...144        self.servers.append(server)145        return server146    def _get_server_key(self, server):147        return self.keypairs[server['key_name']]['private_key']148    def _check_public_network_connectivity(149            self, should_connect=True, msg=None,150            should_check_floating_ip_status=True, mtu=None):151        """Verifies connectivty to a VM via public network and floating IP152        and verifies floating IP has resource status is correct.153        :param should_connect: bool. determines if connectivity check is154        negative or positive.155        :param msg: Failure message to add to Error message. Should describe156        the place in the test scenario where the method was called,157        to indicate the context of the failure158        :param should_check_floating_ip_status: bool. should status of159        floating_ip be checked or not160        :param mtu: int. MTU network to use for connectivity validation161        """162        ssh_login = CONF.validation.image_ssh_user163        floating_ip, server = self.floating_ip_tuple164        ip_address = floating_ip['floating_ip_address']165        private_key = None166        floatingip_status = 'DOWN'167        if should_connect:168            private_key = self._get_server_key(server)169            floatingip_status = 'ACTIVE'170        # Check FloatingIP Status before initiating a connection171        if should_check_floating_ip_status:172            self.check_floating_ip_status(floating_ip, floatingip_status)173        message = 'Public network connectivity check failed'174        if msg:175            message += '. Reason: %s' % msg176        self.check_vm_connectivity(177            ip_address, ssh_login, private_key, should_connect,178            message, server, mtu=mtu)179    def _disassociate_floating_ips(self):180        floating_ip, _ = self.floating_ip_tuple181        floating_ip = self.floating_ips_client.update_floatingip(182            floating_ip['id'], port_id=None)['floatingip']183        self.assertIsNone(floating_ip['port_id'])184        self.floating_ip_tuple = Floating_IP_tuple(floating_ip, None)185    def _reassociate_floating_ips(self):186        floating_ip, server = self.floating_ip_tuple187        # create a new server for the floating ip188        server = self._create_server(self.network)189        port_id, _ = self._get_server_port_id_and_ip4(server)190        floating_ip = self.floating_ips_client.update_floatingip(191            floating_ip['id'], port_id=port_id)['floatingip']192        self.assertEqual(port_id, floating_ip['port_id'])193        self.floating_ip_tuple = Floating_IP_tuple(floating_ip, server)194    def _create_new_network(self, create_gateway=False):195        self.new_net = self._create_network()196        if create_gateway:197            self.new_subnet = self.create_subnet(198                network=self.new_net)199        else:200            self.new_subnet = self.create_subnet(201                network=self.new_net,202                gateway_ip=None)203    def _hotplug_server(self):204        old_floating_ip, server = self.floating_ip_tuple205        ip_address = old_floating_ip['floating_ip_address']206        private_key = self._get_server_key(server)207        ssh_client = self.get_remote_client(208            ip_address, private_key=private_key, server=server)209        old_nic_list = self._get_server_nics(ssh_client)210        # get a port from a list of one item211        port_list = self.os_admin.ports_client.list_ports(212            device_id=server['id'])['ports']213        self.assertEqual(1, len(port_list))214        old_port = port_list[0]215        interface = self.interface_client.create_interface(216            server_id=server['id'],217            net_id=self.new_net['id'])['interfaceAttachment']218        self.addCleanup(self.ports_client.wait_for_resource_deletion,219                        interface['port_id'])220        self.addCleanup(test_utils.call_and_ignore_notfound_exc,221                        self.interface_client.delete_interface,222                        server['id'], interface['port_id'])223        def check_ports():224            self.new_port_list = [225                port for port in226                self.os_admin.ports_client.list_ports(227                    device_id=server['id'])['ports']228                if port['id'] != old_port['id']229            ]230            return len(self.new_port_list) == 1231        if not test_utils.call_until_true(232                check_ports, CONF.network.build_timeout,233                CONF.network.build_interval):234            raise exceptions.TimeoutException(235                "No new port attached to the server in time (%s sec)! "236                "Old port: %s. Number of new ports: %d" % (237                    CONF.network.build_timeout, old_port,238                    len(self.new_port_list)))239        new_port = self.new_port_list[0]240        def check_new_nic():241            new_nic_list = self._get_server_nics(ssh_client)242            self.diff_list = [n for n in new_nic_list if n not in old_nic_list]243            return len(self.diff_list) == 1244        if not test_utils.call_until_true(245                check_new_nic, CONF.network.build_timeout,246                CONF.network.build_interval):247            raise exceptions.TimeoutException("Interface not visible on the "248                                              "guest after %s sec"249                                              % CONF.network.build_timeout)250        _, new_nic = self.diff_list[0]251        ip_output = ssh_client.exec_command('ip a')252        ip_address = new_port['fixed_ips'][0]['ip_address']253        ip_mask = CONF.network.project_network_mask_bits254        # check if the address is not already in use, if not, set it255        if ' ' + ip_address + '/' + str(ip_mask) not in ip_output:256            ssh_client.exec_command("sudo ip addr add %s/%s dev %s" % (257                                    ip_address, ip_mask, new_nic))258            ssh_client.exec_command("sudo ip link set %s up" % new_nic)259    def _get_server_nics(self, ssh_client):260        reg = re.compile(r'(?P<num>\d+): (?P<nic_name>\w+)[@]?.*:')261        ipatxt = ssh_client.exec_command("ip address")262        return reg.findall(ipatxt)263    def _check_network_internal_connectivity(self, network,264                                             should_connect=True):265        """via ssh check VM internal connectivity:266        - ping internal gateway and DHCP port, implying in-tenant connectivity267        pinging both, because L3 and DHCP agents might be on different nodes268        - ping internal compute port, implying connectivity to other VMs on269        this network270        """271        floating_ip, server = self.floating_ip_tuple272        # get internal ports' ips:273        # get all network and compute ports in the new network274        internal_ips = (275            p['fixed_ips'][0]['ip_address'] for p in276            self.os_admin.ports_client.list_ports(277                tenant_id=server['tenant_id'],278                network_id=network['id'])['ports']279            if p['device_owner'].startswith('network') or280            p['device_owner'].startswith('compute')281        )282        self._check_server_connectivity(floating_ip,283                                        internal_ips,284                                        should_connect)285    def _check_network_external_connectivity(self):286        """ping default gateway to imply external connectivity"""287        if not CONF.network.public_network_id:288            msg = 'public network not defined.'289            LOG.info(msg)290            return291        # We ping the external IP from the instance using its floating IP292        # which is always IPv4, so we must only test connectivity to293        # external IPv4 IPs if the external network is dualstack.294        v4_subnets = [295            s for s in self.os_admin.subnets_client.list_subnets(296                network_id=CONF.network.public_network_id)['subnets']297            if s['ip_version'] == 4298        ]299        self.assertEqual(1, len(v4_subnets),300                         "Found %d IPv4 subnets" % len(v4_subnets))301        external_ips = [v4_subnets[0]['gateway_ip']]302        self._check_server_connectivity(self.floating_ip_tuple.floating_ip,303                                        external_ips)304    def _check_server_connectivity(self, floating_ip, address_list,305                                   should_connect=True):306        ip_address = floating_ip['floating_ip_address']307        private_key = self._get_server_key(self.floating_ip_tuple.server)308        ssh_source = self.get_remote_client(309            ip_address, private_key=private_key,310            server=self.floating_ip_tuple.server)311        for remote_ip in address_list:312            self.check_remote_connectivity(ssh_source, remote_ip,313                                           should_connect)314    def _update_router_admin_state(self, router, admin_state_up):315        kwargs = dict(admin_state_up=admin_state_up)316        router = self.routers_client.update_router(317            router['id'], **kwargs)['router']318        self.assertEqual(admin_state_up, router['admin_state_up'])319    @decorators.attr(type='smoke')320    @decorators.idempotent_id('f323b3ba-82f8-4db7-8ea6-6a895869ec49')321    @utils.services('compute', 'network')322    def test_network_basic_ops(self):323        """Basic network operation test324        For a freshly-booted VM with an IP address ("port") on a given network:325        - the Tempest host can ping the IP address.  This implies, but326            does not guarantee (see the ssh check that follows), that the327            VM has been assigned the correct IP address and has328            connectivity to the Tempest host.329        - the Tempest host can perform key-based authentication to an330            ssh server hosted at the IP address.  This check guarantees331            that the IP address is associated with the target VM.332        - the Tempest host can ssh into the VM via the IP address and333            successfully execute the following:334            - ping an external IP address, implying external connectivity.335            - ping an external hostname, implying that dns is correctly336                configured.337            - ping an internal IP address, implying connectivity to another338               VM on the same network.339        - detach the floating-ip from the VM and verify that it becomes340            unreachable341        - associate detached floating ip to a new VM and verify connectivity.342            VMs are created with unique keypair so connectivity also asserts343            that floating IP is associated with the new VM instead of the old344            one345        Verifies that floating IP status is updated correctly after each change346        """347        self._setup_network_and_servers()348        self._check_public_network_connectivity(should_connect=True)349        self._check_network_internal_connectivity(network=self.network)350        self._check_network_external_connectivity()351        self._disassociate_floating_ips()352        self._check_public_network_connectivity(should_connect=False,353                                                msg="after disassociate "354                                                    "floating ip")355        self._reassociate_floating_ips()356        self._check_public_network_connectivity(should_connect=True,357                                                msg="after re-associate "358                                                    "floating ip")359    @decorators.idempotent_id('b158ea55-472e-4086-8fa9-c64ac0c6c1d0')360    @testtools.skipUnless(utils.is_extension_enabled('net-mtu', 'network'),361                          'No way to calculate MTU for networks')362    @decorators.attr(type='slow')363    @utils.services('compute', 'network')364    def test_mtu_sized_frames(self):365        """Validate that network MTU sized frames fit through."""366        self._setup_network_and_servers()367        # first check that connectivity works in general for the instance368        self._check_public_network_connectivity(should_connect=True)369        # now that we checked general connectivity, test that full size frames370        # can also pass between nodes371        self._check_public_network_connectivity(372            should_connect=True, mtu=self.network['mtu'])373    @decorators.idempotent_id('1546850e-fbaa-42f5-8b5f-03d8a6a95f15')374    @testtools.skipIf(CONF.network.shared_physical_network,375                      'Connectivity can only be tested when in a '376                      'multitenant network environment')377    @decorators.attr(type='slow')378    @utils.services('compute', 'network')379    def test_connectivity_between_vms_on_different_networks(self):380        """Test connectivity between VMs on different networks381        For a freshly-booted VM with an IP address ("port") on a given network:382        - the Tempest host can ping the IP address.383        - the Tempest host can ssh into the VM via the IP address and384            successfully execute the following:385            - ping an external IP address, implying external connectivity.386            - ping an external hostname, implying that dns is correctly387               configured.388            - ping an internal IP address, implying connectivity to another389               VM on the same network.390        - Create another network on the same tenant with subnet, create391            an VM on the new network.392            - Ping the new VM from previous VM failed since the new network393                was not attached to router yet.394            - Attach the new network to the router, Ping the new VM from395                previous VM succeed.396        """397        self._setup_network_and_servers()398        self._check_public_network_connectivity(should_connect=True)399        self._check_network_internal_connectivity(network=self.network)400        self._check_network_external_connectivity()401        self._create_new_network(create_gateway=True)402        new_server = self._create_server(self.new_net)403        new_server_ips = [addr['addr'] for addr in404                          new_server['addresses'][self.new_net['name']]]405        # Assert that pinging the new VM fails since the new network is not406        # connected to a router407        self._check_server_connectivity(self.floating_ip_tuple.floating_ip,408                                        new_server_ips, should_connect=False)409        router_id = self.router['id']410        self.routers_client.add_router_interface(411            router_id, subnet_id=self.new_subnet['id'])412        self.addCleanup(test_utils.call_and_ignore_notfound_exc,413                        self.routers_client.remove_router_interface,414                        router_id, subnet_id=self.new_subnet['id'])415        self._check_server_connectivity(self.floating_ip_tuple.floating_ip,416                                        new_server_ips, should_connect=True)417    @decorators.idempotent_id('c5adff73-e961-41f1-b4a9-343614f18cfa')418    @testtools.skipUnless(CONF.compute_feature_enabled.interface_attach,419                          'NIC hotplug not available')420    @testtools.skipIf(CONF.network.port_vnic_type in ['direct', 'macvtap'],421                      'NIC hotplug not supported for '422                      'vnic_type direct or macvtap')423    @utils.services('compute', 'network')424    def test_hotplug_nic(self):425        """Test hotplug network interface426        1. Create a network and a VM.427        2. Check connectivity to the VM via a public network.428        3. Create a new network, with no gateway.429        4. Bring up a new interface430        5. check the VM reach the new network431        """432        self._setup_network_and_servers()433        self._check_public_network_connectivity(should_connect=True)434        self._create_new_network()435        self._hotplug_server()436        self._check_network_internal_connectivity(network=self.new_net)437    @decorators.idempotent_id('04b9fe4e-85e8-4aea-b937-ea93885ac59f')438    @testtools.skipIf(CONF.network.shared_physical_network,439                      'Router state can be altered only with multitenant '440                      'networks capabilities')441    @decorators.attr(type='slow')442    @utils.services('compute', 'network')443    def test_update_router_admin_state(self):444        """Test to update admin state up of router445        1. Check public connectivity before updating446                admin_state_up attribute of router to False447        2. Check public connectivity after updating448                admin_state_up attribute of router to False449        3. Check public connectivity after updating450                admin_state_up attribute of router to True451        """452        self._setup_network_and_servers()453        self._check_public_network_connectivity(454            should_connect=True, msg="before updating "455            "admin_state_up of router to False")456        self._update_router_admin_state(self.router, False)457        # TODO(alokmaurya): Remove should_check_floating_ip_status=False check458        # once bug 1396310 is fixed459        self._check_public_network_connectivity(460            should_connect=False, msg="after updating "461            "admin_state_up of router to False",462            should_check_floating_ip_status=False)463        self._update_router_admin_state(self.router, True)464        self._check_public_network_connectivity(465            should_connect=True, msg="after updating "466            "admin_state_up of router to True")467    @decorators.idempotent_id('d8bb918e-e2df-48b2-97cd-b73c95450980')468    @testtools.skipIf(CONF.network.shared_physical_network,469                      'network isolation not available')470    @testtools.skipUnless(CONF.scenario.dhcp_client,471                          "DHCP client is not available.")472    @decorators.attr(type='slow')473    @utils.services('compute', 'network')474    def test_subnet_details(self):475        """Tests that subnet's extra configuration details are affecting VMs.476        This test relies on non-shared, isolated tenant networks.477        NOTE: Neutron subnets push data to servers via dhcp-agent, so any478        update in subnet requires server to actively renew its DHCP lease.479        1. Configure subnet with dns nameserver480        2. retrieve the VM's configured dns and verify it matches the one481           configured for the subnet.482        3. update subnet's dns483        4. retrieve the VM's configured dns and verify it matches the new one484           configured for the subnet.485        TODO(yfried): add host_routes486        any resolution check would be testing either:487        * l3 forwarding (tested in test_network_basic_ops)488        * Name resolution of an external DNS nameserver - out of scope for489          Tempest490        """491        # this test check only updates (no actual resolution) so using492        # arbitrary ip addresses as nameservers, instead of parsing CONF493        initial_dns_server = '1.2.3.4'494        alt_dns_server = '9.8.7.6'495        # Original timeouts are suggested by salvatore-orlando in496        # https://bugs.launchpad.net/neutron/+bug/1412325/comments/3497        #498        # Compared to that renew_delay was increased, because499        # busybox's udhcpc accepts SIGUSR1 as a renew request. Internally500        # it goes into RENEW_REQUESTED state. If it receives a 2nd SIGUSR1501        # signal while in that state then it calls the deconfig script502        # ("/sbin/cirros-dhcpc deconfig" in sufficiently new cirros versions)503        # which leads to the address being transiently deconfigured which504        # for our case is unwanted.505        renew_delay = 3 * CONF.network.build_interval506        renew_timeout = CONF.network.build_timeout507        self._setup_network_and_servers(dns_nameservers=[initial_dns_server])508        self._check_public_network_connectivity(should_connect=True)509        floating_ip, server = self.floating_ip_tuple510        ip_address = floating_ip['floating_ip_address']511        private_key = self._get_server_key(server)512        ssh_client = self.get_remote_client(513            ip_address, private_key=private_key, server=server)514        dns_servers = [initial_dns_server]515        servers = ssh_client.get_dns_servers()516        self.assertEqual(set(dns_servers), set(servers),517                         'Looking for servers: {trgt_serv}. '518                         'Retrieved DNS nameservers: {act_serv} '519                         'From host: {host}.'520                         .format(host=ssh_client.ssh_client.host,521                                 act_serv=servers,522                                 trgt_serv=dns_servers))523        self.subnet = self.subnets_client.update_subnet(524            self.subnet['id'], dns_nameservers=[alt_dns_server])['subnet']525        # asserts that Neutron DB has updated the nameservers526        self.assertEqual([alt_dns_server], self.subnet['dns_nameservers'],527                         "Failed to update subnet's nameservers")528        def check_new_dns_server():529            # NOTE: Server needs to renew its dhcp lease in order to get new530            # definitions from subnet531            # NOTE(amuller): we are renewing the lease as part of the retry532            # because Neutron updates dnsmasq asynchronously after the533            # subnet-update API call returns.534            ssh_client.renew_lease(fixed_ip=floating_ip['fixed_ip_address'],535                                   dhcp_client=CONF.scenario.dhcp_client)536            if ssh_client.get_dns_servers() != [alt_dns_server]:537                LOG.debug("Failed to update DNS nameservers")538                return False539            return True540        self.assertTrue(test_utils.call_until_true(check_new_dns_server,541                                                   renew_timeout,542                                                   renew_delay),543                        msg="DHCP renewal failed to fetch "544                            "new DNS nameservers")545    @decorators.idempotent_id('f5dfcc22-45fd-409f-954c-5bd500d7890b')546    @testtools.skipUnless(CONF.network_feature_enabled.port_admin_state_change,547                          "Changing a port's admin state is not supported "548                          "by the test environment")549    @decorators.attr(type='slow')550    @utils.services('compute', 'network')551    def test_update_instance_port_admin_state(self):552        """Test to update admin_state_up attribute of instance port553        1. Check public and project connectivity before updating554                admin_state_up attribute of instance port to False555        2. Check public and project connectivity after updating556                admin_state_up attribute of instance port to False557        3. Check public and project connectivity after updating558                admin_state_up attribute of instance port to True559        """560        self._setup_network_and_servers()561        _, server = self.floating_ip_tuple562        server_id = server['id']563        port_id = self.os_admin.ports_client.list_ports(564            device_id=server_id)['ports'][0]['id']565        server_pip = server['addresses'][self.network['name']][0]['addr']566        server2 = self._create_server(self.network)567        server2_fip = self.create_floating_ip(server2)568        private_key = self._get_server_key(server2)569        ssh_client = self.get_remote_client(server2_fip['floating_ip_address'],570                                            private_key=private_key,571                                            server=server2)572        self._check_public_network_connectivity(573            should_connect=True, msg="before updating "574            "admin_state_up of instance port to False")575        self.check_remote_connectivity(ssh_client, dest=server_pip,576                                       should_succeed=True)577        self.ports_client.update_port(port_id, admin_state_up=False)578        self._check_public_network_connectivity(579            should_connect=False, msg="after updating "580            "admin_state_up of instance port to False",581            should_check_floating_ip_status=False)582        self.check_remote_connectivity(ssh_client, dest=server_pip,583                                       should_succeed=False)584        self.ports_client.update_port(port_id, admin_state_up=True)585        self._check_public_network_connectivity(586            should_connect=True, msg="after updating "587            "admin_state_up of instance port to True")588        self.check_remote_connectivity(ssh_client, dest=server_pip,589                                       should_succeed=True)590    @decorators.idempotent_id('759462e1-8535-46b0-ab3a-33aa45c55aaa')591    @decorators.attr(type='slow')592    @utils.services('compute', 'network')593    def test_preserve_preexisting_port(self):594        """Test preserve pre-existing port595        Tests that a pre-existing port provided on server boot is not deleted596        if the server is deleted.597        Nova should unbind the port from the instance on delete if the port was598        not created by Nova as part of the boot request.599        We should also be able to boot another server with the same port.600        """601        # Setup the network, create a port and boot the server from that port.602        self._setup_network_and_servers(boot_with_port=True)603        _, server = self.floating_ip_tuple604        self.assertEqual(1, len(self.ports),605                         'There should only be one port created for '606                         'server %s.' % server['id'])607        port_id = self.ports[0]['port']608        self.assertIsNotNone(port_id,609                             'Server should have been created from a '610                             'pre-existing port.')611        # Assert the port is bound to the server.612        port_list = self.os_admin.ports_client.list_ports(613            device_id=server['id'], network_id=self.network['id'])['ports']614        self.assertEqual(1, len(port_list),615                         'There should only be one port created for '616                         'server %s.' % server['id'])617        self.assertEqual(port_id, port_list[0]['id'])618        # Delete the server.619        self.servers_client.delete_server(server['id'])620        waiters.wait_for_server_termination(self.servers_client, server['id'])621        # Assert the port still exists on the network but is unbound from622        # the deleted server.623        port = self.ports_client.show_port(port_id)['port']624        self.assertEqual(self.network['id'], port['network_id'])625        self.assertEqual('', port['device_id'])626        self.assertEqual('', port['device_owner'])627        # Boot another server with the same port to make sure nothing was628        # left around that could cause issues.629        server = self._create_server(self.network, port['id'])630        port_list = self.os_admin.ports_client.list_ports(631            device_id=server['id'], network_id=self.network['id'])['ports']632        self.assertEqual(1, len(port_list),633                         'There should only be one port created for '634                         'server %s.' % server['id'])635        self.assertEqual(port['id'], port_list[0]['id'])636    @utils.requires_ext(service='network', extension='l3_agent_scheduler')637    @decorators.idempotent_id('2e788c46-fb3f-4ac9-8f82-0561555bea73')638    @decorators.attr(type='slow')639    @utils.services('compute', 'network')640    def test_router_rescheduling(self):641        """Tests that router can be removed from agent and add to a new agent.642        1. Verify connectivity643        2. Remove router from all l3-agents644        3. Verify connectivity is down645        4. Assign router to new l3-agent (or old one if no new agent is646           available)647        5. Verify connectivity648        """649        # TODO(yfried): refactor this test to be used for other agents (dhcp)650        # as well651        list_hosts = (self.os_admin.routers_client.652                      list_l3_agents_hosting_router)653        schedule_router = (self.os_admin.network_agents_client.654                           create_router_on_l3_agent)655        unschedule_router = (self.os_admin.network_agents_client.656                             delete_router_from_l3_agent)657        agent_list_alive = set(658            a["id"] for a in659            self.os_admin.network_agents_client.list_agents(660                agent_type="L3 agent")['agents'] if a["alive"] is True661        )662        self._setup_network_and_servers()663        # NOTE(kevinbenton): we have to use the admin credentials to check664        # for the distributed flag because self.router only has a project view.665        admin = self.os_admin.routers_client.show_router(666            self.router['id'])667        if admin['router'].get('distributed', False):668            msg = "Rescheduling test does not apply to distributed routers."669            raise self.skipException(msg)670        self._check_public_network_connectivity(should_connect=True)671        # remove resource from agents672        hosting_agents = set(a["id"] for a in673                             list_hosts(self.router['id'])['agents'])674        no_migration = agent_list_alive == hosting_agents675        LOG.info("Router will be assigned to {mig} hosting agent".676                 format(mig="the same" if no_migration else "a new"))677        for hosting_agent in hosting_agents:678            unschedule_router(hosting_agent, self.router['id'])679            self.assertNotIn(hosting_agent,680                             [a["id"] for a in681                              list_hosts(self.router['id'])['agents']],682                             'unscheduling router failed')683        # verify resource is un-functional684        self._check_public_network_connectivity(685            should_connect=False,686            msg='after router unscheduling',687        )688        # schedule resource to new agent689        target_agent = list(hosting_agents if no_migration else690                            agent_list_alive - hosting_agents)[0]691        schedule_router(target_agent,692                        router_id=self.router['id'])693        self.assertEqual(694            target_agent,695            list_hosts(self.router['id'])['agents'][0]['id'],696            "Router failed to reschedule. Hosting agent doesn't match "697            "target agent")698        # verify resource is functional699        self._check_public_network_connectivity(700            should_connect=True,701            msg='After router rescheduling')702    @utils.requires_ext(service='network', extension='port-security')703    @testtools.skipUnless(CONF.compute_feature_enabled.interface_attach,704                          'NIC hotplug not available')705    @decorators.idempotent_id('7c0bb1a2-d053-49a4-98f9-ca1a1d849f63')706    @decorators.attr(type='slow')707    @utils.services('compute', 'network')708    def test_port_security_macspoofing_port(self):709        """Tests port_security extension enforces mac spoofing710        Neutron security groups always apply anti-spoof rules on the VMs. This711        allows traffic to originate and terminate at the VM as expected, but712        prevents traffic to pass through the VM. Anti-spoof rules are not713        required in cases where the VM routes traffic through it.714        The test steps are:715        1. Create a new network.716        2. Connect (hotplug) the VM to a new network.717        3. Check the VM can ping a server on the new network ("peer")718        4. Spoof the mac address of the new VM interface.719        5. Check the Security Group enforces mac spoofing and blocks pings via720           spoofed interface (VM cannot ping the peer).721        6. Disable port-security of the spoofed port- set the flag to false.722        7. Retest 3rd step and check that the Security Group allows pings via723           the spoofed interface.724        """725        spoof_mac = "00:00:00:00:00:01"726        # Create server727        self._setup_network_and_servers()728        self._check_public_network_connectivity(should_connect=True)729        self._create_new_network()730        self._hotplug_server()731        fip, server = self.floating_ip_tuple732        new_ports = self.os_admin.ports_client.list_ports(733            device_id=server["id"], network_id=self.new_net["id"])['ports']734        spoof_port = new_ports[0]735        private_key = self._get_server_key(server)736        ssh_client = self.get_remote_client(fip['floating_ip_address'],737                                            private_key=private_key,738                                            server=server)739        spoof_nic = ssh_client.get_nic_name_by_mac(spoof_port["mac_address"])740        peer = self._create_server(self.new_net)741        peer_address = peer['addresses'][self.new_net['name']][0]['addr']742        self.check_remote_connectivity(ssh_client, dest=peer_address,...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.