Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _check_server_connectivity method in tempest

Best Python code snippet using tempest_python

test_fwaas_v2.py

Source:test_fwaas_v2.py

...58                        self.os_admin.ports_client.list_ports(59                            tenant_id=server2['tenant_id'],60                            network_id=network2['id'])['ports']61                        if p['device_owner'].startswith('network'))62        self._check_server_connectivity(63            floating_ip1, keys1, internal_ips, should_connect)64    def _check_server_connectivity(self, floating_ip, keys1, address_list,65                                   should_connect=True):66        ip_address = floating_ip['floating_ip_address']67        private_key = keys168        ssh_source = self.get_remote_client(69            ip_address, private_key=private_key)70        for remote_ip in address_list:71            if should_connect:72                msg = ("Timed out waiting for %s to become "73                       "reachable") % remote_ip74            else:75                msg = "ip address %s is reachable" % remote_ip76            try:77                self.assertTrue(self._check_remote_connectivity78                                (ssh_source, remote_ip, should_connect),79                                msg)80            except Exception:81                LOG.exception("Unable to access {dest} via ssh to "82                              "floating-ip {src}".format(dest=remote_ip,83                                                         src=floating_ip))84                raise85    def _check_remote_connectivity(self, source, dest, should_succeed=True,86                                   nic=None):87        """check ping server via source ssh connection88        :param source: RemoteClient: an ssh connection from which to ping89        :param dest: and IP to ping against90        :param should_succeed: boolean should ping succeed or not91        :param nic: specific network interface to ping from92        :returns: boolean -- should_succeed == ping93        :returns: ping is false if ping failed94        """95        def ping_remote():96            try:97                source.ping_host(dest, nic=nic)98            except lib_exc.SSHExecCommandFailed:99                LOG.warning('Failed to ping IP: %s via a ssh connection '100                            'from: %s.', dest, source.ssh_client.host)101                return not should_succeed102            return should_succeed103        return test_utils.call_until_true(ping_remote,104                                          CONF.validation.ping_timeout,105                                          1)106    def _add_router_interface(self, router_id, subnet_id):107        resp = self.routers_client.add_router_interface(108            router_id, subnet_id=subnet_id)109        self.addCleanup(test_utils.call_and_ignore_notfound_exc,110                        self.routers_client.remove_router_interface, router_id,111                        subnet_id=subnet_id)112        return resp113    def _create_network_subnet(self):114        network = self._create_network()115        subnet_kwargs = dict(network=network)116        subnet = self._create_subnet(**subnet_kwargs)117        return network, subnet118    def _create_test_server(self, network, security_group):119        pub_network_id = CONF.network.public_network_id120        server, keys = self._create_server(121            network, security_group=security_group)122        private_key = keys['private_key']123        server_floating_ip = self.create_floating_ip(server, pub_network_id)124        fixed_ip = server['addresses'].values()[0][0]['addr']125        return server, private_key, fixed_ip, server_floating_ip126    def _create_topology(self):127        """128        +--------+             +-------------+129        |"server"|             | "subnet"    |130        |   VM-1 +-------------+ "network-1" |131        +--------+             +----+--------+132                                    |133                                    | router interface port134                               +----+-----+135                               | "router" |136                               +----+-----+137                                    | router interface port138                                    |139                                    |140        +--------+             +-------------+141        |"server"|             | "subnet"    |142        |   VM-2 +-------------+ "network-2" |143        +--------+             +----+--------+144        """145        LOG.debug('Starting Topology Creation')146        resp = {}147        # Create Network1 and Subnet1.148        network1, subnet1 = self._create_network_subnet()149        resp['network1'] = network1150        resp['subnet1'] = subnet1151        # Create Network2 and Subnet2.152        network2, subnet2 = self._create_network_subnet()153        resp['network2'] = network2154        resp['subnet2'] = subnet2155        # Create a router and attach Network1, Network2 and External Networks156        # to it.157        router = self._create_router(namestart='SCENARIO-TEST-ROUTER')158        pub_network_id = CONF.network.public_network_id159        kwargs = {'external_gateway_info': dict(network_id=pub_network_id)}160        router = self.routers_client.update_router(161            router['id'], **kwargs)['router']162        router_id = router['id']163        resp_add_intf = self._add_router_interface(164            router_id, subnet_id=subnet1['id'])165        router_portid_1 = resp_add_intf['port_id']166        resp_add_intf = self._add_router_interface(167            router_id, subnet_id=subnet2['id'])168        router_portid_2 = resp_add_intf['port_id']169        resp['router'] = router170        resp['router_portid_1'] = router_portid_1171        resp['router_portid_2'] = router_portid_2172        # Create a VM on each of the network and assign it a floating IP.173        security_group = self._create_security_group()174        server1, private_key1, server_fixed_ip_1, server_floating_ip_1 = (175            self._create_test_server(network1, security_group))176        server2, private_key2, server_fixed_ip_2, server_floating_ip_2 = (177            self._create_test_server(network2, security_group))178        resp['server1'] = server1179        resp['private_key1'] = private_key1180        resp['server_fixed_ip_1'] = server_fixed_ip_1181        resp['server_floating_ip_1'] = server_floating_ip_1182        resp['server2'] = server2183        resp['private_key2'] = private_key2184        resp['server_fixed_ip_2'] = server_fixed_ip_2185        resp['server_floating_ip_2'] = server_floating_ip_2186        return resp187    @decorators.idempotent_id('77fdf3ea-82c1-453d-bfec-f7efe335625d')188    def test_icmp_reachability_scenarios(self):189        topology = self._create_topology()190        ssh_login = CONF.validation.image_ssh_user191        self.check_vm_connectivity(192            ip_address=topology['server_floating_ip_1']['floating_ip_address'],193            username=ssh_login,194            private_key=topology['private_key1'])195        self.check_vm_connectivity(196            ip_address=topology['server_floating_ip_2']['floating_ip_address'],197            username=ssh_login,198            private_key=topology['private_key2'])199        # Scenario 1: Add allow ICMP rules between the two VMs.200        fw_allow_icmp_rule = self.create_firewall_rule(action="allow",201                                                       protocol="icmp")202        fw_allow_ssh_rule = self.create_firewall_rule(action="allow",203                                                      protocol="tcp",204                                                      destination_port=22)205        fw_policy = self.create_firewall_policy(206            firewall_rules=[fw_allow_icmp_rule['id'], fw_allow_ssh_rule['id']])207        fw_group = self.create_firewall_group(208            ports=[209                topology['router_portid_1'],210                topology['router_portid_2']],211            ingress_firewall_policy_id=fw_policy['id'],212            egress_firewall_policy_id=fw_policy['id'])213        self._wait_firewall_group_ready(fw_group['id'])214        LOG.debug('fw_allow_icmp_rule: %s\nfw_allow_ssh_rule: %s\n'215                  'fw_policy: %s\nfw_group: %s\n',216                  fw_allow_icmp_rule, fw_allow_ssh_rule, fw_policy, fw_group)217        # Check the connectivity between VM1 and VM2. It should Pass.218        self._check_server_connectivity(219            topology['server_floating_ip_1'],220            topology['private_key1'],221            address_list=[topology['server_fixed_ip_2']],222            should_connect=True)223        # Scenario 2: Now remove the allow_icmp rule add a deny_icmp rule and224        # check that ICMP gets blocked225        fw_deny_icmp_rule = self.create_firewall_rule(action="deny",226                                                      protocol="icmp")227        self.remove_firewall_rule_from_policy_and_wait(228            firewall_group_id=fw_group['id'],229            firewall_rule_id=fw_allow_icmp_rule['id'],230            firewall_policy_id=fw_policy['id'])231        self.insert_firewall_rule_in_policy_and_wait(232            firewall_group_id=fw_group['id'],233            firewall_rule_id=fw_deny_icmp_rule['id'],234            firewall_policy_id=fw_policy['id'])235        self._check_server_connectivity(236            topology['server_floating_ip_1'],237            topology['private_key1'],238            address_list=[topology['server_fixed_ip_2']],239            should_connect=False)240        # Scenario 3: Create a rule allowing ICMP only from server_fixed_ip_1241        # to server_fixed_ip_2 and check that traffic from opposite direction242        # is blocked.243        fw_allow_unidirectional_icmp_rule = self.create_firewall_rule(244            action="allow", protocol="icmp",245            source_ip_address=topology['server_fixed_ip_1'],246            destination_ip_address=topology['server_fixed_ip_2'])247        self.remove_firewall_rule_from_policy_and_wait(248            firewall_group_id=fw_group['id'],249            firewall_rule_id=fw_deny_icmp_rule['id'],250            firewall_policy_id=fw_policy['id'])251        self.insert_firewall_rule_in_policy_and_wait(252            firewall_group_id=fw_group['id'],253            firewall_rule_id=fw_allow_unidirectional_icmp_rule['id'],254            firewall_policy_id=fw_policy['id'])255        self._check_server_connectivity(256            topology['server_floating_ip_1'],257            topology['private_key1'],258            address_list=[topology['server_fixed_ip_2']],259            should_connect=True)260        self._check_server_connectivity(261            topology['server_floating_ip_2'],262            topology['private_key2'],263            address_list=[topology['server_fixed_ip_1']],...

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.