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

How to use get_device_slots method in lisa

Best Python code snippet using lisa_python

nic.py

Source:nic.py

...127    def get_upper_nics(self) -> List[str]:128        return list(self.nics.keys())129    def get_lower_nics(self) -> List[str]:130        return [x.lower for x in self.nics.values() if x.lower]131    def get_device_slots(self) -> List[str]:132        return [x.pci_slot for x in self.nics.values() if x.pci_slot]133    # update the current nic driver in the NicInfo instance134    # grabs the driver short name and the driver sysfs path135    def get_nic_driver(self, nic_name: str) -> str:136        # get the current driver for the nic from the node137        # sysfs provides a link to the driver entry at device/driver138        nic = self.get_nic(nic_name)139        cmd = f"readlink -f /sys/class/net/{nic_name}/device/driver"140        # ex return value:141        # /sys/bus/vmbus/drivers/hv_netvsc142        found_link = self._node.execute(cmd, expected_exit_code=0).stdout143        assert_that(found_link).described_as(144            f"sysfs check for NIC device {nic_name} driver returned no output"145        ).is_not_equal_to("")...

device.py

Source:device.py

1'''2Created on Apr 25, 20113@author: jono4'''5from ..sql import Query6from ..ssh import parse_version_file7from ....utils.version import Version8class DeviceNotFound(Exception):9    pass10class TaskNotFound(Exception):11    pass12filter_device_info = None13class FilterDeviceInfo(Query): #@IgnorePep814    def __init__(self, filter=None, mgmtip_for_local_em=False,  # @ReservedAssignment15                 textual_localhost=True, *args, **kwargs):16        assert filter is None or callable(filter), "Filter must be callable or None"17        super(FilterDeviceInfo, self).__init__(query=None, *args, **kwargs)18        self.filter = filter19        fields = ['uid', 'is_local_em',20                  'mgmt_address', 'host_name', 'platform',21                  'active_slot', 'chassis_serial', 'perfmon_state',22                  'monitoring_bytes_per_second', 'is_clustered',23                  'monitoring_counters_per_second', 'discovery_status',24                  'product_name', 'version', 'build_number',25                  'disk_partition_scheme', 'current_job_uid']26        fields = ['d.%s' % x for x in fields]27        if self.ifc.version < 'em 2.3.0' or not textual_localhost:28            fields.append('d.access_address')29        elif mgmtip_for_local_em:30            fields.append("IF(is_local_em, mgmt_address, access_address) AS access_address")31        else:32            fields.append("IF(is_local_em, 'localhost', access_address) AS access_address")33        # Starting with solar-em-em the 'supports_asm' column has been removed.34        # Faking it using a JOIN with device_component.35        if parse_version_file(ifc=self.ifc).get('project') in ('solar-em-em', 'em-nsd'):36            fields.append('COUNT(dc1.uid) AS supports_asm')37            query = "SELECT %s FROM device d " \38                    "LEFT JOIN device_component dc1 " \39                    "ON d.uid = dc1.device_uid AND dc1.component_type = 'ASM' " \40                    "GROUP BY d.uid" % ','.join(fields)41        else:42            fields.append('d.supports_asm')43            query = "SELECT %s FROM device d" % ','.join(fields)44        self.query = query45    def setup(self):46        return filter(self.filter, super(FilterDeviceInfo, self).setup())47get_device_info = None48class GetDeviceInfo(FilterDeviceInfo): #@IgnorePep849    def __init__(self, mgmtip, *args, **kwargs):50        self.mgmtip = mgmtip51        super(GetDeviceInfo, self).__init__(filter=lambda x: x.mgmt_address == mgmtip,52                                            *args, **kwargs)53    def setup(self):54        try:55            return super(GetDeviceInfo, self).setup()[0]56        except IndexError:57            raise DeviceNotFound("%s" % self.mgmtip)58get_device_version = None59class GetDeviceVersion(GetDeviceInfo): #@IgnorePep860    def setup(self):61        try:62            ret = super(GetDeviceInfo, self).setup()[0]63        except IndexError:64            raise DeviceNotFound("%s" % self.mgmtip)65        return Version("%(product_name)s %(version)s %(build_number)s" % ret)66get_device_state = None67class GetDeviceState(Query): #@IgnorePep868    def __init__(self, mgmtip=None, uids=None, *args, **kwargs):69        self.mgmtip = mgmtip70        fields = ['d.uid', 'access_address', 'mgmt_address', 'refresh_failed_at',71                  'refresh_state', 'status', 'substatus_message', 'host_name']72        if isinstance(mgmtip, basestring):73            where = "WHERE access_address = '%s'" % mgmtip74        elif isinstance(mgmtip, (tuple, list)):75            where = "WHERE access_address IN (%s)" % ','.join(("'%s'" % str(x) for x in mgmtip))76        elif isinstance(uids, (tuple, list)):77            where = "WHERE d.uid IN (%s)" % ','.join(("%s" % str(x) for x in uids))78        else:79            where = ''80        query = """SELECT %s FROM device d81                       LEFT JOIN device_2_substatus d2s ON (d.uid = d2s.device_uid)82                       LEFT JOIN device_substatus ds ON (d2s.substatus_uid = ds.uid)83                       %s84                """ % (','.join(fields), where)85        super(GetDeviceState, self).__init__(query=query, *args, **kwargs)86    def setup(self):87        try:88            return super(GetDeviceState, self).setup()89        except IndexError:90            raise DeviceNotFound("%s" % self.mgmtip)91get_reachable_devices = None92class GetReachableDevices(Query): #@IgnorePep893    def __init__(self, *args, **kwargs):94        fields = ['d.uid', 'access_address', 'mgmt_address', 'refresh_failed_at',95                  'refresh_state', 'status', 'substatus_message', 'is_local_em']96        query = """SELECT %s FROM device d97                       LEFT JOIN device_2_substatus d2s ON (d.uid = d2s.device_uid)98                       LEFT JOIN device_substatus ds ON (d2s.substatus_uid = ds.uid)99                   WHERE refresh_failed_at is NULL""" % \100                    (','.join(fields))101        super(GetReachableDevices, self).__init__(query=query, *args, **kwargs)102    def setup(self):103        return super(GetReachableDevices, self).setup()104get_device_slots = None105class GetDeviceSlots(Query): #@IgnorePep8106    def __init__(self, mgmtip, *args, **kwargs):107        fields = ['product', 'slot_key', 'slot_num', 'visible_name',108                  'ds.version', 'ds.build', 'ds.is_cf', 'ds.reg_key', 'ds.uid']109        query = """SELECT %s FROM device_slot ds110                             JOIN device d ON (ds.device_id = d.uid)111                             WHERE d.access_address = '%s'""" % \112                    (','.join(fields), mgmtip)113        super(GetDeviceSlots, self).__init__(query=query, *args, **kwargs)114get_device_active_slot = None115class GetDeviceActiveSlot(Query): #@IgnorePep8116    def __init__(self, mgmtip, *args, **kwargs):117        self.mgmtip = mgmtip118        fields = ['product', 'slot_key', 'slot_num', 'visible_name',119                  'ds.version', 'ds.build', 'ds.is_cf', 'ds.reg_key', 'ds.uid']120        query = """SELECT %s FROM device_slot ds121                             JOIN device d ON (ds.device_id = d.uid AND122                                               ds.slot_num = d.active_slot)123                             WHERE d.access_address = '%s'""" % \124                    (','.join(fields), mgmtip)125        super(GetDeviceActiveSlot, self).__init__(query=query, *args, **kwargs)126    def setup(self):127        try:128            return super(GetDeviceActiveSlot, self).setup()[0]129        except IndexError:130            raise DeviceNotFound("%s" % self.mgmtip)131get_task = None132class GetTask(Query): #@IgnorePep8133    def __init__(self, task_id, *args, **kwargs):134        self.task_id = int(task_id)135        super(GetTask, self).__init__(query=task_id, *args, **kwargs)136    def prep(self):137        fields = ['uid', 'task_name', 'create_time', 'start_time', 'finish_time',138                  'error_code', 'error_message', 'user_name', 'detail_table',139                  'status']140        self.query = """SELECT %s FROM job_header WHERE uid = '%d'""" % \141                    (','.join(fields), self.task_id)142        super(GetTask, self).prep()143get_discovery_task = None144class GetDiscoveryTask(GetTask): #@IgnorePep8145    def setup(self):146        fields = ['jh.uid', 'task_name', 'create_time', 'start_time', 'finish_time',147                  'error_code', 'error_message', 'user_name', 'detail_table',148                  'status', 'progress_percent']149        self.query = """SELECT %s FROM job_header jh150                                  JOIN discovery_job dj ON (dj.uid = jh.uid)151                                  WHERE jh.uid = '%d'""" % \152                    (','.join(fields), self.task_id)153        try:154            task = super(GetDiscoveryTask, self).setup()[0]155        except IndexError:156            raise TaskNotFound("%s" % self.task_id)157        # 'device-busy'158        self.query = """SELECT COUNT(*) AS count FROM discovery_results159                        WHERE uid = %s AND discovery_status IN160                            ('auth-failed', 'comm-failed', 'unsupported',161                             'unsupported-product-version','db-save-failed',162                             'exceeds-license', 'device-busy')""" % \163                    task['uid']164        error_count = super(GetDiscoveryTask, self).setup()[0]['count']165        task['error_count'] = int(error_count)166        fields = ['uid', 'access_address', 'system_id', 'product_name',167                  'version', 'build_number', 'discovery_status',168                  'discovery_status_message']169        self.query = """SELECT %s FROM discovery_results170                                  WHERE uid = %d""" % \171                    (','.join(fields), self.task_id)172        details = super(GetDiscoveryTask, self).setup()173        task.details = details174        task.progress_percent = int(task.progress_percent)175        return task176get_big3d_task = None177class GetBig3dTask(GetTask): #@IgnorePep8178    def setup(self):179        fields = ['jh.uid', 'dj.task_name', 'create_time', 'start_time', 'finish_time',180                  'error_code', 'error_message', 'user_name', 'detail_table',181                  'status']182        self.query = """SELECT %s FROM job_header jh183                                  JOIN big3d_install_job dj ON (dj.job_header_uid = jh.uid)184                                  WHERE jh.uid = '%d'""" % \185                    (','.join(fields), self.task_id)186        try:187            task = super(GetBig3dTask, self).setup()[0]188        except IndexError:189            raise TaskNotFound("%s" % self.task_id)190        # 'device-busy'191        self.query = """SELECT COUNT(*) AS count FROM big3d_install_device_job192                        WHERE big3d_install_job_uid = %s AND error_code != 0""" % \193                    task['uid']194        error_count = super(GetBig3dTask, self).setup()[0]['count']195        task['error_count'] = int(error_count)196        fields = ['uid', 'progress_percent', 'status', 'display_device_name',197                  'display_device_address', 'display_initial_big3d_version',198                  'error_code', 'error_message', 'device_uid']199        self.query = """SELECT %s FROM big3d_install_device_job200                                  WHERE big3d_install_job_uid = %d""" % \201                    (','.join(fields), self.task_id)202        details = super(GetBig3dTask, self).setup()203        task.details = details204        task.progress_percent = sum([100 if int(x['progress_percent']) < -1 else int(x['progress_percent'])205                                    for x in details]) / len(details)206        return task207get_changeset_task = None208class GetChangesetTask(GetTask): #@IgnorePep8209    def setup(self):210        fields = ['jh.uid', 'dscj.task_name', 'create_time', 'start_time', 'finish_time',211                  'error_code', 'error_message', 'user_name', 'detail_table',212                  'status']213        self.query = """SELECT %s FROM job_header jh214                                  JOIN deploy_staged_changeset_job dscj ON (dscj.job_header_uid = jh.uid)215                                  WHERE jh.uid = '%d'""" % \216                    (','.join(fields), self.task_id)217        try:218            task = super(GetChangesetTask, self).setup()[0]219        except IndexError:220            raise TaskNotFound("%s" % self.task_id)221        # 'device-busy'222        self.query = """SELECT COUNT(*) AS count FROM deploy_staged_changeset_changeset_job223                        WHERE job_header_uid = %s AND error_code != 0""" % \224                    task['uid']225        error_count = super(GetChangesetTask, self).setup()[0]['count']226        task['error_count'] = int(error_count)227        fields = ['uid', 'progress_percent', 'status', 'target_device', 'target_partition',228                  'display_source_name', 'display_source_description',229                  'error_code', 'error_message', 'source_uid']230        self.query = """SELECT %s FROM deploy_staged_changeset_changeset_job231                                  WHERE job_header_uid = %d""" % \232                    (','.join(fields), self.task_id)233        details = super(GetChangesetTask, self).setup()234        task.details = details235        task.progress_percent = sum([100 if int(x['progress_percent']) < -1 else int(x['progress_percent'])236                                    for x in details]) / len(details)237        return task238count_device__tasks = None239class CountDeviceTasks(Query): #@IgnorePep8240    def __init__(self, mgmtips=None, *args, **kwargs):241        query = """SELECT COUNT(*) AS count FROM device d242                       JOIN device_2_job d2j ON (d.uid = d2j.device_uid)"""243        if mgmtips:244            query += """245                   WHERE d.access_address IN (%s)""" % \246                                    ','.join(("'%s'" % str(x) for x in mgmtips))247        super(CountDeviceTasks, self).__init__(query=query, *args, **kwargs)248    def setup(self):249        return int(super(CountDeviceTasks, self).setup()[0]['count'])250count_active_tasks = None251class CountActiveTasks(Query): #@IgnorePep8252    def __init__(self, type=None, *args, **kwargs):  # @ReservedAssignment253        query = """SELECT COUNT(*) AS count FROM f5em.job_header jh254        WHERE status IN ('pending', 'started', 'running', 'canceling')"""255        if type:256            query += " AND jh.detail_table IN (%s)" % ','.join(("'%s'" %257                                                          str(x) for x in type))258        super(CountActiveTasks, self).__init__(query=query, *args, **kwargs)259    def setup(self):260        return int(super(CountActiveTasks, self).setup()[0]['count'])261get_device_archives = None262class GetDeviceArchives(Query): #@IgnorePep8263    def __init__(self, device_uid, pinned=True, *args, **kwargs):264        query = """SELECT * FROM config_archive d WHERE device_uid = %s AND pinned = %d""" % (device_uid, int(pinned))265        super(GetDeviceArchives, self).__init__(query=query, *args, **kwargs)266get_diff_filenames = None267class GetDiffFilenames(Query): #@IgnorePep8268    def __init__(self, *args, **kwargs):269        query = """SELECT * FROM diff_file_names"""270        super(GetDiffFilenames, self).__init__(query=query, *args, **kwargs)271    def setup(self):...

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.