KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _is_online method in avocado

Best Python code snippet using avocado_python

fetchcliout.py

Source:fetchcliout.py

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

models.py

Source:models.py

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

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.