KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _get_server_ip method in tempest

Best Python code snippet using tempest_python
test_security_groups_basic_ops.py
Source:test_security_groups_basic_ops.py
...261        self._create_tenant_keypairs(tenant)262        self._create_tenant_network(tenant)263        self._create_tenant_security_groups(tenant)264        self._set_access_point(tenant)265    def _get_server_ip(self, server, floating=False):266        """returns the ip (floating/internal) of a server"""267        if floating:268            server_ip = self.floating_ips[server['id']].floating_ip_address269        else:270            server_ip = None271            network_name = self.tenants[server['tenant_id']].network.name272            if network_name in server['addresses']:273                server_ip = server['addresses'][network_name][0]['addr']274        return server_ip275    def _connect_to_access_point(self, tenant):276        """create ssh connection to tenant access point"""277        access_point_ssh = \278            self.floating_ips[tenant.access_point['id']].floating_ip_address279        private_key = tenant.keypair['private_key']280        access_point_ssh = self._ssh_to_server(access_point_ssh,281                                               private_key=private_key)282        return access_point_ssh283    def _check_connectivity(self, access_point, ip, should_succeed=True):284        if should_succeed:285            msg = "Timed out waiting for %s to become reachable" % ip286        else:287            msg = "%s is reachable" % ip288        self.assertTrue(self._check_remote_connectivity(access_point, ip,289                                                        should_succeed), msg)290    def _test_in_tenant_block(self, tenant):291        access_point_ssh = self._connect_to_access_point(tenant)292        for server in tenant.servers:293            self._check_connectivity(access_point=access_point_ssh,294                                     ip=self._get_server_ip(server),295                                     should_succeed=False)296    def _test_in_tenant_allow(self, tenant):297        ruleset = dict(298            protocol='icmp',299            remote_group_id=tenant.security_groups['default'].id,300            direction='ingress'301        )302        self._create_security_group_rule(303            secgroup=tenant.security_groups['default'],304            **ruleset305        )306        access_point_ssh = self._connect_to_access_point(tenant)307        for server in tenant.servers:308            self._check_connectivity(access_point=access_point_ssh,309                                     ip=self._get_server_ip(server))310    def _test_cross_tenant_block(self, source_tenant, dest_tenant):311        # if public router isn't defined, then dest_tenant access is via312        # floating-ip313        access_point_ssh = self._connect_to_access_point(source_tenant)314        ip = self._get_server_ip(dest_tenant.access_point,315                                 floating=self.floating_ip_access)316        self._check_connectivity(access_point=access_point_ssh, ip=ip,317                                 should_succeed=False)318    def _test_cross_tenant_allow(self, source_tenant, dest_tenant):319        """check for each direction:320        creating rule for tenant incoming traffic enables only 1way traffic321        """322        ruleset = dict(323            protocol='icmp',324            direction='ingress'325        )326        self._create_security_group_rule(327            secgroup=dest_tenant.security_groups['default'],328            client=dest_tenant.manager.network_client,329            **ruleset330        )331        access_point_ssh = self._connect_to_access_point(source_tenant)332        ip = self._get_server_ip(dest_tenant.access_point,333                                 floating=self.floating_ip_access)334        self._check_connectivity(access_point_ssh, ip)335        # test that reverse traffic is still blocked336        self._test_cross_tenant_block(dest_tenant, source_tenant)337        # allow reverse traffic and check338        self._create_security_group_rule(339            secgroup=source_tenant.security_groups['default'],340            client=source_tenant.manager.network_client,341            **ruleset342        )343        access_point_ssh_2 = self._connect_to_access_point(dest_tenant)344        ip = self._get_server_ip(source_tenant.access_point,345                                 floating=self.floating_ip_access)346        self._check_connectivity(access_point_ssh_2, ip)347    def _verify_mac_addr(self, tenant):348        """Verify that VM has the same ip, mac as listed in port"""349        access_point_ssh = self._connect_to_access_point(tenant)350        mac_addr = access_point_ssh.get_mac_address()351        mac_addr = mac_addr.strip().lower()352        # Get the fixed_ips and mac_address fields of all ports. Select353        # only those two columns to reduce the size of the response.354        port_list = self._list_ports(fields=['fixed_ips', 'mac_address'])355        port_detail_list = [356            (port['fixed_ips'][0]['subnet_id'],357             port['fixed_ips'][0]['ip_address'],358             port['mac_address'].lower())359            for port in port_list if port['fixed_ips']360        ]361        server_ip = self._get_server_ip(tenant.access_point)362        subnet_id = tenant.subnet.id363        self.assertIn((subnet_id, server_ip, mac_addr), port_detail_list)364    @test.idempotent_id('e79f879e-debb-440c-a7e4-efeda05b6848')365    @test.services('compute', 'network')366    def test_cross_tenant_traffic(self):367        if not self.credentials_provider.is_multi_tenant():368            raise self.skipException("No secondary tenant defined")369        try:370            # deploy new tenant371            self._deploy_tenant(self.alt_tenant)372            self._verify_network_details(self.alt_tenant)373            self._verify_mac_addr(self.alt_tenant)374            # cross tenant check375            source_tenant = self.primary_tenant376            dest_tenant = self.alt_tenant377            self._test_cross_tenant_block(source_tenant, dest_tenant)378            self._test_cross_tenant_allow(source_tenant, dest_tenant)379        except Exception:380            for tenant in self.tenants.values():381                self._log_console_output(servers=tenant.servers)382            raise383    @test.idempotent_id('63163892-bbf6-4249-aa12-d5ea1f8f421b')384    @test.services('compute', 'network')385    def test_in_tenant_traffic(self):386        try:387            self._create_tenant_servers(self.primary_tenant, num=1)388            # in-tenant check389            self._test_in_tenant_block(self.primary_tenant)390            self._test_in_tenant_allow(self.primary_tenant)391        except Exception:392            for tenant in self.tenants.values():393                self._log_console_output(servers=tenant.servers)394            raise395    @test.idempotent_id('f4d556d7-1526-42ad-bafb-6bebf48568f6')396    @test.services('compute', 'network')397    def test_port_update_new_security_group(self):398        """Verifies the traffic after updating the vm port399        With new security group having appropriate rule.400        """401        new_tenant = self.primary_tenant402        # Create empty security group and add icmp rule in it403        new_sg = self._create_empty_security_group(404            namestart='secgroup_new-',405            tenant_id=new_tenant.creds.tenant_id,406            client=new_tenant.manager.security_groups_client)407        icmp_rule = dict(408            protocol='icmp',409            direction='ingress',410        )411        self._create_security_group_rule(412            secgroup=new_sg,413            client=new_tenant.manager.network_client,414            **icmp_rule)415        new_tenant.security_groups.update(new_sg=new_sg)416        # Create server with default security group417        name = 'server-{tenant}-gen-1'.format(418               tenant=new_tenant.creds.tenant_name419        )420        name = data_utils.rand_name(name)421        server = self._create_server(name, new_tenant)422        # Check connectivity failure with default security group423        try:424            access_point_ssh = self._connect_to_access_point(new_tenant)425            self._check_connectivity(access_point=access_point_ssh,426                                     ip=self._get_server_ip(server),427                                     should_succeed=False)428            server_id = server['id']429            port_id = self._list_ports(device_id=server_id)[0]['id']430            # update port with new security group and check connectivity431            self.ports_client.update_port(port_id, security_groups=[432                new_tenant.security_groups['new_sg'].id])433            self._check_connectivity(434                access_point=access_point_ssh,435                ip=self._get_server_ip(server))436        except Exception:437            for tenant in self.tenants.values():438                self._log_console_output(servers=tenant.servers)439            raise440    @test.idempotent_id('d2f77418-fcc4-439d-b935-72eca704e293')441    @test.services('compute', 'network')442    def test_multiple_security_groups(self):443        """Verify multiple security groups and checks that rules444        provided in the both the groups is applied onto VM445        """446        tenant = self.primary_tenant447        ip = self._get_server_ip(tenant.access_point,448                                 floating=self.floating_ip_access)449        ssh_login = CONF.validation.image_ssh_user450        private_key = tenant.keypair['private_key']451        self.check_vm_connectivity(ip,452                                   should_connect=False)453        ruleset = dict(454            protocol='icmp',455            direction='ingress'456        )457        self._create_security_group_rule(458            secgroup=tenant.security_groups['default'],459            **ruleset460        )461        # NOTE: Vm now has 2 security groups one with ssh rule(462        # already added in setUp() method),and other with icmp rule463        # (added in the above step).The check_vm_connectivity tests464        # -that vm ping test is successful465        # -ssh to vm is successful466        self.check_vm_connectivity(ip,467                                   username=ssh_login,468                                   private_key=private_key,469                                   should_connect=True)470    @test.requires_ext(service='network', extension='port-security')471    @test.idempotent_id('7c811dcc-263b-49a3-92d2-1b4d8405f50c')472    @test.services('compute', 'network')473    def test_port_security_disable_security_group(self):474        """Verify the default security group rules is disabled."""475        new_tenant = self.primary_tenant476        # Create server477        name = 'server-{tenant}-gen-1'.format(478               tenant=new_tenant.creds.tenant_name479        )480        name = data_utils.rand_name(name)481        server = self._create_server(name, new_tenant)482        access_point_ssh = self._connect_to_access_point(new_tenant)483        server_id = server['id']484        port_id = self._list_ports(device_id=server_id)[0]['id']485        # Flip the port's port security and check connectivity486        try:487            self.ports_client.update_port(port_id,488                                          port_security_enabled=True,489                                          security_groups=[])490            self._check_connectivity(access_point=access_point_ssh,491                                     ip=self._get_server_ip(server),492                                     should_succeed=False)493            self.ports_client.update_port(port_id,494                                          port_security_enabled=False,495                                          security_groups=[])496            self._check_connectivity(497                access_point=access_point_ssh,498                ip=self._get_server_ip(server))499        except Exception:500            for tenant in self.tenants.values():501                self._log_console_output(servers=tenant.servers)...
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.