How to use _is_online method in avocado

Best Python code snippet using avocado_python

fetchcliout.py

Source:fetchcliout.py Github

copy

Full Screen

1'''2@summary: Methods for common device operation tasks3'''4from nxapi.nxapi_utils import NXAPITransport5from urlparse import urlparse6import time7import json8import os9import re10VERSION_OUT_FRMT = 'CHASSIS: {chassis}; IMAGE: {image}'11CPU_THRESHOLD = 50.0012MEMORY_USED_THRESHOLD = 50.0013class FetchCliOut(object):14 def __init__(self, target_url, username, password, **kwargs):15 '''16 @param target_url: URL format. For telnet/ssh use, telnet://hostip:port or ssh://hostip:port17 For NXAPI, use http://hostip/ins18 @param username: Username of device19 @param password: Password of device20 '''21 self.username = username22 self.password = password23 self.target_url = target_url24 25 parsed_url = urlparse(target_url)26 self.hostname = parsed_url.hostname27 self.port = parsed_url.port28 29 self.error_online = ''30 self.health_statuses = list() # Format: [(description, healthy)]31 32 self._is_online = False33 34 def _exception(self, reason):35 raise Exception(reason)36 37 def _log(self, msg):38 print msg39 @property40 def is_healthy(self):41 raise NotImplementedError()42 43 @property 44 def is_online(self):45 raise NotImplementedError()46 47 @property48 def sw_version(self):49 raise NotImplementedError()50 51class FetchViaRoutercli(FetchCliOut):52 '''53 Get output from CLI ssh/telnet as string54 55 TODO: Implement all functions like FetchViaNxapi56 '''57 def __init__(self, *args, **kwargs):58 from utils.routercli.router import Router59 super(FetchViaRoutercli, self).__init__(*args, **kwargs)60 self.routercli_obj = Router(self.hostname, logger=False, logfile_console=None)61 self.routercli_obj.login(username=self.username, password=self.password)62 63 def __del__(self):64 self.routercli_obj.logout()65 66 67class FetchViaNxapi(FetchCliOut):68 '''69 Get output from http/https as JSON70 '''71 def __init__(self, *args, **kwargs):72 super(FetchViaNxapi, self).__init__(*args, **kwargs)73 self.login()74 75 def login(self):76 try:77 NXAPITransport.init(target_url=self.target_url, username=self.username, 78 password=self.password)79 NXAPITransport.cli('! Testing if device is online')80 except Exception as e:81 self._is_online = False82 e = str(e).lower()83 if 'timed out' in e:84 self.error_online = ('Connection timed out!! Make sure that '85 'URL provided ({0}) is accurate'86 '').format(self.target_url)87 elif 'unauthorized' in e:88 self.error_online = ('Unauthorized!! Make sure that username '89 '({0}) and/or password provided are accurate'90 '').format(self.username)91 elif 'target machine actively refused it' in e:92 self.error_online = ('NXAPI TCP Port not open!! Make sure '93 '"feature nxapi" is configured on '94 'target Nexus switch')95 else:96 self.error_online = e97 else:98 self._is_online = True99 return self._is_online100 101 @property102 def sw_version(self):103 version = ''104 if self._is_online is True:105 version = VERSION_OUT_FRMT106 o = json.loads(NXAPITransport.clid('show version'))107 version = version.format(chassis=o['chassis_id'], 108 image=o['kick_file_name'])109 return version110 111 @property112 def is_healthy(self):113 healthy = True114 if self._is_online is True:115 o1 = json.loads(NXAPITransport.clid('show system resources'))116 loads = list()117 loads.append(float(o1['cpu_state_kernel']))118 loads.append(float(o1['load_avg_1min']))119 loads.append(float(o1['load_avg_15min']))120 loads.append(float(o1['cpu_state_user']))121 healthy_cpu = True122 cpu_used_percent = ('{0} (Kernel), {1} (1min), {2} '123 '(15min), {3} (User)'124 '').format(loads[0], loads[1], 125 loads[2], loads[3])126 for load in loads:127 if load > CPU_THRESHOLD:128 healthy_cpu = False129 healthy = False130 loggedin_msg = 'Logged in successfully to {0}'.format(self.target_url)131 self.health_statuses.append((loggedin_msg, True))132 self.health_statuses.append(('CPU Used (%): {0}'.format(cpu_used_percent), 133 healthy_cpu))134 135 memory_usage_used = float(o1['memory_usage_used'])136 memory_usage_total = float(o1['memory_usage_total'])137 mem_used_percent = memory_usage_used/memory_usage_total*100138 if mem_used_percent > MEMORY_USED_THRESHOLD:139 healthy_mem = False140 healthy = False141 else:142 healthy_mem = True143 self.health_statuses.append((('Memory Used (%): {0:.2f}'144 ''.format(mem_used_percent)), 145 healthy_mem))146 147 current_memory_status = o1['current_memory_status']148 if current_memory_status != 'OK':149 healthy_mem_txt = False150 healthy = False151 else:152 healthy_mem_txt = True153 self.health_statuses.append((('Current Memory Status: {0}'154 ''.format(current_memory_status)), 155 healthy_mem_txt))156 157 o2 = json.loads(NXAPITransport.clid('show module'))158 online_diag_statuses = o2['TABLE_moddiaginfo']['ROW_moddiaginfo']159 diagstatuses = True160 for r in online_diag_statuses:161 diagstatus = r['diagstatus']162 if diagstatus != 'Pass':163 diagstatuses = False164 healthy = False165 self.health_statuses.append((('Module no. {0} '166 'Diag Status: {1}'167 ''.format(r['mod'], diagstatus)), 168 diagstatuses))169 170 if diagstatuses:171 self.health_statuses.append(('Modules Diag Status: Pass', 172 diagstatuses))173 174 modinfo = o2['TABLE_modinfo']['ROW_modinfo']175 modstatus = True176 for r in modinfo:177 status = r['status']178 if status not in ['ok', 'active', 'active *', 'standby']:179 modstatus = False180 healthy = False181 self.health_statuses.append((('Module no. {0} Status: {1}'182 ''.format(r['modinf']), status), 183 modstatus))184 185 if modstatus:186 self.health_statuses.append(('Modules Status: Ok', modstatus))187 else:188 healthy = False189 return healthy190 191 @property192 def is_online(self):193 return self.login()194 195 def get_osinfo(self):196 '''197 @return: dict d with keys: 'osdevicename', 'osplatform',198 'ostime', 'osuptime'199 '''200 d = {'osdevicename': '',201 'osplatform': '',202 'ostime': '',203 'osuptime': ''}204 if self._is_online:205 version = VERSION_OUT_FRMT206 o = json.loads(NXAPITransport.clid('show version'))207 version = version.format(chassis=o['chassis_id'], 208 image=o['kick_file_name'])209 osuptime = ('days: {0}, hrs: {1}, mins: {2}, secs: {3}'210 ''.format(o['kern_uptm_days'], o['kern_uptm_hrs'],211 o['kern_uptm_mins'], o['kern_uptm_secs']))212 devicename = o['host_name']213 o = json.loads(NXAPITransport.clid('show clock'))214 ostime = o["simple_time"]215 d['osplatform'] = version216 d['osdevicename'] = devicename217 d['ostime'] = ostime218 d['osuptime'] = osuptime219 return d220 221 def get_cpustats(self):222 '''223 @return: dict d with keys matching exactly as 224 attributes of dashboardperdevice.models.CpuStats225 '''226 d = dict()227 if self._is_online:228 o = json.loads(NXAPITransport.clid('show system resources'))229 d['per1min'] = o['load_avg_1min']230 d['per5min'] = o['load_avg_5min']231 d['per15min'] = o['load_avg_15min']232 return d233 234 def get_memstats(self):235 '''236 @return: dict d with keys matching exactly as 237 attributes of dashboardperdevice.models.MemStats238 '''239 d = dict()240 if self._is_online:241 o = json.loads(NXAPITransport.clid('show system resources'))242 d['mem_used'] = o['memory_usage_used']243 d['mem_free'] = o['memory_usage_free']244 d['mem_total'] = o['memory_usage_total']245 return d246 247 def get_dirstats(self):248 '''249 @return: dict d with keys matching exactly as 250 attributes of dashboardperdevice.models.DirStats251 '''252 l = list()253 modules_dirs = [['sup-active', 'bootflash'], ['sup-standby', 'bootflash']]254 if self._is_online:255 for module, dirpath in modules_dirs:256 cmd = 'dir {0}://{1}'.format(dirpath, module)257 o = NXAPITransport.cli(cmd)258 d = dict()259 try:260 d['used'] = re.search('(\d+ bytes) used', o).group(1)261 d['free'] = re.search('(\d+ bytes) free', o).group(1)262 d['total'] = re.search('(\d+ bytes) total', o).group(1)263 264 used = float(d['used'].rstrip(' bytes'))265 total = float(d['total'].rstrip(' bytes'))266 used_percent = (used/total)*100267 d['used_percent'] = '{0}%'.format(int(used_percent))268 except:269 d['used'] = d['free'] = d['total'] = d['used_percent'] = 'NA'270 d['module'] = module271 d['dirpath'] = dirpath272 l.append(d)273 return l274 275 def get_modulestats(self):276 '''277 @return: dict d with keys matching exactly as 278 attributes of dashboardperdevice.models.ModulesStats279 '''280 l = list()281 if self._is_online:282 o = json.loads(NXAPITransport.clid('show module'))283 try:284 diaginfos = o['TABLE_moddiaginfo']['ROW_moddiaginfo'] 285 modinfos = o['TABLE_modinfo']['ROW_modinfo']286 modmacinfos = o['TABLE_modmacinfo']['ROW_modmacinfo']287 for r in modinfos:288 d = dict()289 d['status'] = r['status']290 d['hw_desc'] = r['modtype']291 d['hw_model'] = r['model']292 d['mod_id'] = r['modinf']293 d['ports'] = r['ports']294 d['serial_no'] = [i['serialnum'] for i in modmacinfos if i['modmac'] == d['mod_id']][0]295 d['diag_stat'] = [i['diagstatus'] for i in diaginfos if i['mod'] == d['mod_id']][0]296 l.append(d)297 except Exception as e:298 print 'Error while get_modulestats', e299 return l300 301 def get_intstats(self):302 '''303 @return: dict d with keys matching exactly as 304 attributes of dashboardperdevice.models.InterfacesStats305 '''306 list_of_ints = list()307 if self._is_online:308 rows = json.loads(NXAPITransport.clid('show interface'))['TABLE_interface']['ROW_interface']309 for o in rows:310 d = dict()311 int_name = o["interface"]312 d['int_name'] = int_name313 if 'vlan' in int_name.lower():314 int_state = o.get("svi_line_proto", '')315 int_adminstate = o.get("svi_admin_state", '')316 int_mtu = o.get("svi_mtu", '')317 int_bw = o.get("svi_bw", '')318 int_hwdesc = o.get("svi_hw_desc", "VLAN")319 int_hwaddr = o.get("svi_mac", '')320 int_ipaddr = o.get("svi_ip_addr", '')321 int_ipmask = o.get("svi_ip_mask", '')322 else:323 int_state = o.get("state", '')324 int_adminstate = o.get("admin_state", int_state)325 int_mtu = o.get("eth_mtu", '')326 int_bw = o.get("eth_bw", '')327 int_hwdesc = o.get("eth_hw_desc", '')328 int_hwaddr = o.get("eth_hw_addr", '')329 int_ipaddr = o.get("eth_ip_addr", '')330 int_ipmask = o.get("eth_ip_mask", '')331 if 'mgmt' in int_name.lower():332 int_bpsrate_rx = o.get("vdc_lvl_in_avg_bits", '')333 int_bpsrate_tx = o.get("vdc_lvl_out_avg_bits", '')334 int_loadinterval = 60335 else:336 int_bpsrate_rx = o.get("eth_inrate1_bits", '')337 int_bpsrate_tx = o.get("eth_outrate1_bits", '')338 int_loadinterval = o.get("eth_load_interval1_rx", 0)339 if 'down' in int_state.lower():340 int_bpsrate_tx = int_bpsrate_rx = 0341 342 d['int_state'] = int_state343 d['int_adminstate'] = int_adminstate344 d['int_mtu'] = int_mtu345 d['int_bw'] = int_bw346 d['int_desc'] = o.get("desc", '')347 d['int_hwdesc'] = int_hwdesc348 d['int_hwaddr'] = int_hwaddr349 d['int_ipaddr'] = int_ipaddr350 d['int_ipmask'] = int_ipmask351 d['int_bpsrate_rx'] = int_bpsrate_rx352 d['int_bpsrate_tx'] = int_bpsrate_tx353 d['int_loadinterval'] = int(int_loadinterval)354 list_of_ints.append(d)355 356 return list_of_ints357 358def fetchcli_wrapper(hostname):359 '''360 This function is wrapper for getting FetchVia* object and 361 query object for a hostname362 Only two things are modified in the HostNames DB here:363 * timestamp364 * polling_method365 366 The entry of hostname in HostNames is created by a POST to form LoginForm367 '''368 from django.conf import settings369 from hostnames.models import HostNames370 371 hostname_obj = HostNames.objects.get(hostname=hostname)372 hostname_obj.polling_timestamp = time.time()373 374 # Trying to connect375 target_url = hostname_obj.url376 username = hostname_obj.username377 password = HostNames._meta.get_field('password').value_from_object(hostname_obj) # @UndefinedVariable378 379 parsed_url = urlparse(target_url)380 381 if parsed_url.path == '/ins' and 'http' in parsed_url.scheme:382 polling_method = settings.POLLING_METHOD_NXAPI383 else:384 polling_method = settings.POLLING_METHOD_CLI385 hostname_obj.polling_method = polling_method386 387 if polling_method == settings.POLLING_METHOD_CLI:388 fetcho = FetchViaRoutercli(target_url=target_url, username=username, password=password)389 elif polling_method == settings.POLLING_METHOD_NXAPI:390 fetcho = FetchViaNxapi(target_url=target_url, username=username, password=password)391 return hostname_obj, fetcho392 393 394def main():395 pass396# fetcho = FetchViaRoutercli(target_url='telnet://comcast-term-1', username='cisco', password='cisco')397# print 'out1', repr(fetcho.get_os_info())398# fetcho = FetchViaNxapi(target_url='http://10.201.30.194/ins', username='CISCO\\ambarik', password='xxxxxx')399# print 'sw_version', repr(fetcho.sw_version)400# print 'is_online', repr(fetcho.is_online)401# print 'error_online', repr(fetcho.error_online)402# print 'is_healthy', repr(fetcho.is_healthy)403# print 'health_statuses', repr(fetcho.health_statuses)404# print 'get_osinfo', fetcho.get_osinfo()405# print 'get_intstats', fetcho.get_intstats()406# print 'get_cpustats', fetcho.get_cpustats()407# print 'get_memstats', fetcho.get_memstats()408# print 'get_dirstats', fetcho.get_dirstats()409# print 'get_dirstats', fetcho.get_dirstats()410# print 'get_modulestats', fetcho.get_modulestats()411# print 'test'412 413 414if __name__ == '__main__':415 main()...

Full Screen

Full Screen

models.py

Source:models.py Github

copy

Full Screen

1from solarsan import logging2logger = logging.getLogger(__name__)3from solarsan import conf4from solarsan.models import CreatedModifiedDocMixIn, ReprMixIn5from solarsan.configure.models import Nic6from solarsan.exceptions import ConnectionError7import mongoengine as m8import rpyc9import weakref10"""11Cluster12"""13class Peer(CreatedModifiedDocMixIn, ReprMixIn, m.Document):14 #uuid = m.UUIDField(binary=False)15 uuid = m.StringField(required=True, unique=True)16 hostname = m.StringField(required=True, unique=True)17 #hostname = m.StringField(required=True, unique=True)18 #is_local = m.BooleanField()19 ifaces = m.ListField()20 addrs = m.ListField()21 netmasks = m.ListField()22 # TODO Make this a list in desc priority or sort addrs by priority23 cluster_addr = m.StringField()24 cluster_iface = m.StringField()25 last_seen = m.DateTimeField()26 is_offline = m.BooleanField()27 # For ReprMixIn28 _repr_vars = ['hostname']29 """30 General31 """32 def __init__(self, **kwargs):33 super(Peer, self).__init__(**kwargs)34 self._is_online = None35 self._lost_count = 036 self._services = weakref.WeakValueDictionary()37 # TODO Manager38 @classmethod39 def get_local(cls):40 ret, created = Peer.objects.get_or_create(uuid=conf.config['uuid'], defaults={'hostname': conf.hostname})41 ret.hostname = conf.hostname42 ret.uuid = conf.config['uuid']43 ret.cluster_iface = conf.config['cluster_iface']44 return ret45 @property46 def is_local(self):47 return self.hostname == conf.hostname48 @property49 def cluster_nic(self):50 return Nic(self.cluster_iface)51 """52 Newer RPC53 """54 @property55 def is_online(self):56 if self._is_online is None:57 try:58 self.storage.ping()59 self._is_online = True60 except:61 self._is_online = False62 return self._is_online63 def get_service(self, name, default='exception', cache=True, client_service=None):64 """Looks for service on Peer. Returns an existing one if possible, otherwise instantiates one."""65 NAME = str(name).upper()66 check_service = None67 if cache:68 check_service = self._services.get(NAME)69 if not client_service:70 client_service = rpyc.VoidService71 if not check_service or check_service.closed:72 service = None73 try:74 service = rpyc.connect_by_service(NAME,75 host=self.cluster_addr,76 service=client_service,77 #config=conf.rpyc_conn_config,78 )79 # Remove existing aliases to old service80 if cache:81 if check_service:82 for alias in self._services.keys():83 if self._services[alias] == check_service:84 self._services.pop(alias)85 # Add in the new service's86 for alias in service.root.get_service_aliases():87 self._services[alias] = service88 except Exception, e:89 if check_service:90 check_service.close()91 if service:92 service.close()93 if default == 'exception':94 raise ConnectionError('Cannot get service "%s": "%s"', name, e.message)95 else:96 return default97 else:98 service = self._services[NAME]99 return service100 @property101 def storage(self):102 return self.get_service('storage')103 def __call__(self, method, *args, **kwargs):104 default_on_timeout = kwargs.pop('_default_on_timeout', 'exception')105 # TODO handle multiple service names106 service = self.get_service('storage')107 try:108 meth = getattr(service.root, method)109 ret = meth(*args, **kwargs)110 if not self.is_online:111 self._is_online = True112 self._lost_count = 0113 return ret114 except EOFError:115 if self.is_online:116 self._is_online = False117 self._lost_count += 1118 if default_on_timeout == 'exception':119 raise120 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 avocado 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