How to use verify_connection_status method in lisa

Best Python code snippet using lisa_python

analytics_tests.py

Source:analytics_tests.py Github

copy

Full Screen

...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(...

Full Screen

Full Screen

drm.py

Source:drm.py Github

copy

Full Screen

...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:...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful