How to use verify_connection_status method in lisa

Best Python code snippet using lisa_python

analytics_tests.py

Source:analytics_tests.py

...184        for k,v in GENERATORS.items():185            if (k == 'Compute'):186                for name in self.inputs.compute_names:187                    for elem in v:188                        assert self.verify_connection_status(189                                name,elem,k) 190            if (k == 'Analytics'):191                for name in self.inputs.collector_names:192                    for elem in v:193                        assert self.verify_connection_status(194                                name,elem,k) 195            if (k == 'Database'):196                for name in self.inputs.database_names:197                    for elem in v:198                        assert  self.verify_connection_status(199                                name,elem,k) 200            if (k == 'Config'):201                202                for name in self.inputs.cfgm_names:203                    result = False204                    for elem in v:205                        result = result or self.verify_connection_status(206                                name,elem,k)207                assert result        208                         209            if (k == 'Control'):210                for name in self.inputs.bgp_names:211                    for elem in v:212                        assert self.verify_connection_status(213                                name,elem,k) 214    @retry(delay=5, tries=4)215    def verify_connection_status(self, generator, moduleid, node_type, instanceid='0'):216        '''Verify if connection status with collector and generator:node_type:moduleid:instance217            is established218        '''219        self.g = generator220        self.m = moduleid221        result = True222        for collector_ip in self.inputs.collector_ips:223            self.logger.info("Verifying through opserver in %s" %224                             (collector_ip))225            status = self.get_connection_status(226                collector_ip, self.g, self.m, node_type, instanceid)227            if (status == 'Established'):228                self.logger.info("%s:%s:%s:%s is connected to collector %s" %229                                 (self.g, node_type, self.m, instanceid, collector_ip))230                result = result & True231            else:232                self.logger.warn(233                    "%s:%s:%s:%s is NOT connected to collector %s" %234                    (self.g, node_type, self.m, instanceid, collector_ip))235                result = result & False236        return result237    def get_collector_of_gen(self, collector, gen, module, node_type, instance='0'):238        '''Gets the collector node of a generator239        '''240        connobj = self.get_connection_dict(241            collector, gen, module, node_type, instance)242        return connobj['collector_name']243    def get_all_generator_links(self, module=None):244        '''Get all links for a particular generator'''245        ret = []246        try:247            links = self.ops_inspect[self.inputs.collector_ips[248                0]].get_hrefs_to_all_UVEs_of_a_given_UVE_type(uveType='generators')249            if links:250                pattern = '%s(.*)' % module251                compiled = re.compile(pattern)252                for elem in links:253                    if compiled.search(str(elem)):254                        ret.append(elem)255        except Exception as e:256            self.logger.warn("Got exception as %s" % (e))257        finally:258            return ret259    def get_module_instances(self, module):260        '''Return the module instances from analytics/genarators url'''261        ret = []262        try:263            links = self.get_all_generator_links(module=module)264            if links:265                for elem in links:266                    inst = str(elem['name']).split(":")[-1]267                    ret.append(inst)268        except Exception as e:269            self.logger.warn("Got exception as %s" % (e))270        finally:271            return ret272    def get_uve_key(self, uve=None):273        '''{274            href: "http://10.204.216.14:8081/analytics/uves/virtual-machine/292c7779-c085-4079-91f6-440272bd2922?flat",275            name: "292c7779-c085-4079-91f6-440272bd2922"276        }'''277        ret = []278        try:279            links = self.ops_inspect[self.inputs.collector_ips[0]280                                     ].get_hrefs_to_all_UVEs_of_a_given_UVE_type(uveType=uve)281            if links:282                for elem in links:283                    ret.append(elem['name'])284        except Exception as e:285            self.logger.warn("Got exception as %s" % (e))286        finally:287            return ret288# Collector uve functions#289# ------------------------#290  #  @retry(delay=5, tries=1)291    def verify_collector_uve(self):292        '''Verify that all generators are connected to collector'''293        result = True294        # Verify module-ids correctly shown in the  collector uve for respective generators295         # verify module-id for bgp node in collector uve - should be296         # 'Contrail-Control'297        for ip in self.inputs.bgp_ips:298            assert self.verify_collector_connection_introspect(ip,http_introspect_ports['HttpPortControl'])299        for ip in self.inputs.cfgm_ips:300            assert self.verify_collector_connection_introspect(ip,http_introspect_ports['HttpPortApiServer'])301        result = False302        for ip in self.inputs.cfgm_ips:303            result= result or self.verify_collector_connection_introspect(ip,http_introspect_ports['HttpPortSchemaTransformer'])304        assert result305        result = False306        for ip in self.inputs.cfgm_ips:307            result = result or self.verify_collector_connection_introspect(ip,http_introspect_ports['HttpPortSvcMonitor'])308        assert result309        for ip in self.inputs.collector_ips:310            assert self.verify_collector_connection_introspect(ip,http_introspect_ports['HttpPortOpserver'])311        for ip in self.inputs.collector_ips:312            assert self.verify_collector_connection_introspect(ip,http_introspect_ports['HttpPortQueryEngine'])313        for ip in self.inputs.collector_ips:314            self.logger.info("Verifying through opserver in %s" % (ip))315            expected_module_id = ['contrail-control', 'contrail-dns']316            expected_node_type = 'Control'317            expected_instance_id = '0'318            for bgp_host in self.bgp_hosts:319                for module in expected_module_id:320                    is_established = self.verify_connection_status(321                        bgp_host, module, expected_node_type, expected_instance_id)322                    # collector=self.output['collector_name']323                    if is_established:324                    #self.logger.info("%s:%s connected to collector %s"%(bgp_host,module,collector))325                        result = result and True326                    else:327                        result = result and False328            expected_module_id = 'contrail-vrouter-agent'329            expected_node_type = 'Compute'330            expected_instance_id = '0'331            for compute_host in self.compute_hosts:332                is_established = self.verify_connection_status(333                    compute_host, expected_module_id, expected_node_type, expected_instance_id)334                # collector=self.output['collector_name']335                if is_established:336                    result = result and True337                else:338                    result = result and False339            # Verifying module_id from ApiServer340            expected_cfgm_modules = 'contrail-schema'341            expected_node_type = 'Config'342            expected_instance_id = '0'343            for cfgm_node in self.inputs.cfgm_names:344                result1 = True345                is_established = self.verify_connection_status(346                    cfgm_node, expected_cfgm_modules, expected_node_type, expected_instance_id)347                if is_established:348                    # collector=self.output['collector_name']349                    result1 = result1 and True350                    break351                else:352                    result1 = result1 and False353                    st = self.ops_inspect[self.inputs.collector_ips[0]].send_trace_to_database(354                                                            node=self.inputs.collector_names[0], \355                                                            module='Contrail-Analytics-Api', trace_buffer_name='DiscoveryMsg')356                    self.logger.info("status: %s" % (st))357            result = result and result1358            # Verifying module_id from DiscoveryService 359            expected_cfgm_modules = 'contrail-discovery'360            expected_node_type = 'Config'361            expected_instance_id = '0'362            for cfgm_node in self.inputs.cfgm_names:363                result1 = True364                is_established = self.verify_connection_status(365                    cfgm_node, expected_cfgm_modules, expected_node_type, expected_instance_id)366                if is_established:367                    # collector=self.output['collector_name']368                    result1 = result1 and True369                    break370                else:371                    result1 = result1 and False372                    st = self.ops_inspect[self.inputs.collector_ips[0]].send_trace_to_database(373                                                            node=self.inputs.collector_names[0], \374                                                            module='Contrail-Analytics-Api', trace_buffer_name='DiscoveryMsg')375                    self.logger.info("status: %s" % (st))376            result = result and result1377            #Verifying for ServiceMonitor378            expected_cfgm_modules = 'contrail-svc-monitor'379            expected_node_type = 'Config'380            expected_instance_id = '0'381            for cfgm_node in self.inputs.cfgm_names:382                result1 = True383                is_established = self.verify_connection_status(384                    cfgm_node, expected_cfgm_modules, expected_node_type, expected_instance_id)385                if is_established:386                    # collector=self.output['collector_name']387                    resulti1 = result1 and True388                    break389                else:390                    result1 = result1 and False391                    st = self.ops_inspect[self.inputs.collector_ips[0]].send_trace_to_database(392                                                            node=self.inputs.collector_names[0], \393                                                            module='Contrail-Analytics-Api', trace_buffer_name='DiscoveryMsg')394            result = result and result1395            # Verifying module_id  ApiServer396            expected_apiserver_module = 'Contrail-Api'397            expected_apiserver_instances = self.get_module_instances(398                expected_apiserver_module)399            expected_node_type = 'Config'400            # expected_cfgm_modules=['Contrail-Schema','contrail-svc-monitor']401            for cfgm_node in self.inputs.cfgm_names:402                for inst in expected_apiserver_instances:403                    result1 = True404                    is_established = self.verify_connection_status(405                        cfgm_node, expected_apiserver_module, expected_node_type, inst)406                    if is_established:407                        result1 = result1 and True408                        break409                    else:410                        result = result and False411                        st = self.ops_inspect[self.inputs.collector_ips[0]].send_trace_to_database(412                                                            node=self.inputs.collector_names[0], \413                                                            module='Contrail-Analytics-Api', trace_buffer_name='DiscoveryMsg')414                result = result1 and result415            # Verifying module_id Contrail-Analytics-Api416            expected_opserver_module = 'Contrail-Analytics-Api'417            expected_opserver_instances = self.get_module_instances(418                expected_opserver_module)419            expected_node_type = 'Analytics'420            for c_host in self.collector_hosts:421                for inst in expected_opserver_instances:422                    is_established = self.verify_connection_status(423                        c_host, expected_opserver_module, expected_node_type, inst)424                    if is_established:425                        # collector=self.output['collector_name']426                        result = result and True427                    else:428                        result = result and False429            # Verifying collector:moduleid430            expected_collector_module = ['contrail-collector', 'contrail-query-engine']431            expected_node_type = 'Analytics'432            expected_instance_id = '0'433            for c_host in self.collector_hosts:434                for module in expected_collector_module:435                    is_established = self.verify_connection_status(436                        c_host, module, expected_node_type, expected_instance_id)437                    # collector=self.output['collector_name']438                    if is_established:439                        result = result and True440                    else:441                        result = result and False442        return result443    @retry(delay=3, tries=15)444    def verify_hrefs_to_all_uves_of_a_given_uve_type(self):445        '''Verify all analytics links446        '''447        result = True448        for ip in self.inputs.collector_ips:449            self.logger.info(...

drm.py

Source:drm.py

...82        2. Verify the status return from modetest is connected.83        """,84        priority=2,85    )86    def verify_connection_status(self, node: Node, log: Logger) -> None:87        is_status_connected = node.tools[Modetest].is_status_connected("hyperv_drm")88        assert_that(is_status_connected).described_as(89            "dri connector status should be 'connected'"90        ).is_true()91    def before_case(self, log: Logger, **kwargs: Any) -> None:92        node = kwargs["node"]93        kernel_version = node.tools[Uname].get_linux_information().kernel_version94        if (95            isinstance(node.os, Redhat)96            and node.os.information.version >= "9.0.0"97            and kernel_version > "5.13.0"98        ):99            log.debug("Currently only RHEL9 enables drm driver.")100        else:...

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.