Best Python code snippet using avocado_python
host.py
Source:host.py
...45 reply_t = ret["exec_time"]46 return reply_t47 def get_info(self) -> dict:48 self._logger.info("Reading SUT information")49 ret = ltp.sut.collect_sysinfo(self)50 ret.pop("kernel_tained")51 self._logger.debug(ret)52 return ret53 def get_tained_info(self) -> set:54 self._logger.info("Checking for tained kernel")55 code, messages = ltp.sut.collect_sysinfo(self)["kernel_tained"]56 self._logger.debug("code=%d, messages=%s", code, messages)57 return code, messages58 def communicate(self,59 timeout: float = 3600,60 iobuffer: IOBuffer = None) -> None:61 if self.is_running:62 raise SUTError("SUT is running")63 self._initialized = True64 # some pylint versions don't recognize threading.Lock.locked()65 # pylint: disable=no-member66 def _inner_stop(self, sig: int, timeout: float = 30) -> None:67 """68 Wait process to stop.69 """...
__init__.py
Source:__init__.py
...52 yield from self.collect_usinfo()53 yield from self.collect_dsinfo()54 yield from self.collect_uptime()55 yield from self.collect_network()56 yield from self.collect_sysinfo()57 yield from self.collect_docsis()58 def collect_usinfo(self):59 usinfo_sigstr = prometheus_client.core.GaugeMetricFamily('hitron_channel_upstream_signal_strength_dbmv', '', labels=['port', 'channel', 'frequency'])60 usinfo_bw = prometheus_client.core.GaugeMetricFamily('hitron_channel_upstream_bandwidth', '', labels=['port', 'channel', 'frequency'])61 for uschannel in self.__usinfo:62 key = [uschannel['portId'], uschannel['channelId'], uschannel['frequency']]63 usinfo_sigstr.add_metric(key, float(uschannel['signalStrength']))64 usinfo_bw.add_metric(key, int(uschannel['bandwidth']))65 yield usinfo_sigstr66 yield usinfo_bw67 def collect_dsinfo(self):68 dsinfo_sigstr = prometheus_client.core.GaugeMetricFamily('hitron_channel_downstream_signal_strength_dbmv', '', labels=['port', 'channel', 'frequency'])69 dsinfo_snr = prometheus_client.core.GaugeMetricFamily('hitron_channel_downstream_snr', '', labels=['port', 'channel', 'frequency'])70 for dschannel in self.__dsinfo:71 key = [dschannel['portId'], dschannel['channelId'], dschannel['frequency']]72 dsinfo_sigstr.add_metric(key, float(dschannel['signalStrength']))73 dsinfo_snr.add_metric(key, float(dschannel['snr']))74 yield dsinfo_sigstr75 yield dsinfo_snr76 77 def collect_uptime(self):78 m = re.match(r'(\d+) Days,(\d+) Hours,(\d+) Minutes,(\d+) Seconds', self.__sysinfo[0]['systemUptime'])79 if m:80 td = datetime.timedelta(days=int(m.group(1)), hours=int(m.group(2)), minutes=int(m.group(3)), seconds=int(m.group(4)))81 yield prometheus_client.core.CounterMetricFamily('hitron_system_uptime_seconds_total', '', value=td.total_seconds())82 def collect_network(self):83 nw_tx = prometheus_client.core.CounterMetricFamily('hitron_network_transmit_bytes', '', labels=['device'])84 nbytes = self.parse_pkt(self.__sysinfo[0]['LSendPkt'])85 if nbytes:86 nw_tx.add_metric(['lan'], nbytes)87 nbytes = self.parse_pkt(self.__sysinfo[0]['WSendPkt'])88 if nbytes:89 nw_tx.add_metric(['wan'], nbytes)90 yield nw_tx91 nw_rx = prometheus_client.core.CounterMetricFamily('hitron_network_receive_bytes', '', labels=['device'])92 nbytes = self.parse_pkt(self.__sysinfo[0]['LRecPkt'])93 if nbytes:94 nw_rx.add_metric(['lan'], nbytes)95 nbytes = self.parse_pkt(self.__sysinfo[0]['WRecPkt'])96 if nbytes:97 nw_rx.add_metric(['wan'], nbytes)98 yield nw_rx99 def parse_pkt(self, pkt):100 m = re.match(r'(\d+(?:\.\d+)?)([A-Z]?) Bytes', pkt)101 if not m:102 LOGGER.error("Couldn't parse %r as pkt", pkt)103 return None104 factor = {105 '': 1,106 'K': 1e3,107 'M': 1e6,108 }.get(m.group(2))109 if not factor:110 LOGGER.error("Unknown pkt factor %r", m.group(2))111 return None112 return float(m.group(1)) * factor113 def collect_sysinfo(self):114 yield prometheus_client.core.InfoMetricFamily('hitron_system', '', value={115 'serial_number': self.__sysinfo[0]['serialNumber'],116 'software_version': self.__sysinfo[0]['swVersion'],117 'hardware_version': self.__sysinfo[0]['hwVersion'],118 'model_name': self.__system_model['modelName'],119 })120 def collect_docsis(self):121 bpi = {}122 for element in self.__cminit[0]['bpiStatus'].split(','):123 k, _, v = element.strip().partition(':')124 bpi[k.lower()] = v.lower()...
sysinfo.py
Source:sysinfo.py
...31 parser.add_argument('sysinfodir', type=str,32 help='Dir where to dump sysinfo',33 nargs='?', default='')34 def run(self, args):...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!