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

How to use heartbeat method in locust

Best Python code snippet using locust

heartbeatmonitor_061018-1107.py

Source:heartbeatmonitor_061018-1107.py

...262        elif self.heartbeat_monitor == 'testing':263            logger.info('Alert Message:    ' + alert_message)264            logger.info('Alert Submessage: ' + str(alert_submessage))265        logger.info('Heartbeat monitor stopped successfully.')266    def heartbeat(self, message=None):267        if self.use_json_storage == True:268            self.heartbeat_delta = datetime.datetime.now() - self.heartbeat_last269            logger.debug('self.heartbeat_delta: ' + str(self.heartbeat_delta))270            self.heartbeat_last = datetime.datetime.now()271            logger.debug('self.heartbeat_last: ' + str(self.heartbeat_last))272            self.json_save_data['heartbeat_delta'] = self.heartbeat_delta273            self.json_save_data['heartbeat_last'] = self.heartbeat_last274            logger.debug('Dumping heartbeat data to json file.')275            #with open(self.json_save_file, 'w', encoding='utf-8') as file:276                #json.dump(self.json_save_data, file, indent=4, sort_keys=True, ensure_ascii=False)277            json_status = HeartbeatMonitor.read_write_json(self, self.json_save_data)278            logger.debug('json_status[\'status\']: ' + str(json_status['status']))279            if self.flatline_alerts_only == False:280                alert_message = str(self.heartbeat_last)281                if message != None:282                    alert_message += ' - (' + message + ')'283                logger.debug('alert_message: ' + alert_message)284                heartbeat_last_delta = "{:.2f}".format(float(self.heartbeat_delta.total_seconds()) / 60)285                alert_submessage = '*Last heartbeat:* ' + heartbeat_last_delta + ' minutes ago.'286                logger.debug('alert_submessage: ' + alert_submessage)287                if self.heartbeat_monitor == 'slack':288                    alert_result = HeartbeatMonitor.send_slack_alert(self,289                                                                     channel_id=self.slack_alert_channel_id_heartbeat,290                                                                     message=alert_message,291                                                                     submessage=alert_submessage,292                                                                     flatline=False)293                    logger.debug('alert_result: ' + str(alert_result))294                elif self.heartbeat_monitor == 'testing':295                    logger.info('Alert Message:    ' + alert_message)296                    logger.info('Alert Submessage: ' + alert_submessage)297            else:298                logger.debug('Skipping Slack alert for regular heartbeat trigger.')299        else:300            #self.heartbeat_delta = (datetime.datetime.now() - self.monitor_states['heartbeat_last']).total_seconds()301            self.heartbeat_delta = datetime.datetime.now() - self.monitor_states['heartbeat_last']302            logger.debug('self.heartbeat_delta: ' + str(self.heartbeat_delta))303            self.monitor_states['heartbeat_last'] = datetime.datetime.now()304            logger.debug('self.monitor_states[\'heartbeat_last\']: ' + str(self.monitor_states['heartbeat_last']))305            if self.flatline_alerts_only == False:306                heartbeat_last_delta = "{:.2f}".format(float((datetime.datetime.now() - self.monitor_states['heartbeat_last']).total_seconds()) / 60)307                alert_message = str(self.monitor_states['heartbeat_last'])308                if message != None:309                    alert_message += ' - (' + message + ')'310                logger.debug('alert_message: ' + alert_message)311                alert_submessage = '*Last heartbeat:* ' + heartbeat_last_delta + ' minutes ago.'312                logger.debug('alert_submessage: ' + alert_submessage)313                if self.heartbeat_monitor == 'slack':314                    alert_result = HeartbeatMonitor.send_slack_alert(self,315                                                                     channel_id=self.slack_alert_channel_id_heartbeat,316                                                                     message=alert_message,317                                                                     submessage=alert_submessage,318                                                                     flatline=False)319                    logger.debug('alert_result: ' + str(alert_result))320                elif self.heartbeat_monitor == 'testing':321                    logger.info('Alert Message:    ' + alert_message)322                    logger.info('Alert Submessage: ' + alert_submessage)323            else:324                logger.debug('Skipping Slack alert for regular heartbeat trigger.')325    def monitor(self):326        if self.use_json_storage == True:327            try:328                json_modified_time = os.stat(self.json_save_file).st_mtime329                self.monitor_isrunning = True330                while (True):331                    json_current_modified_time = os.stat(self.json_save_file).st_mtime332                    if json_current_modified_time != json_modified_time:333                        logger.debug('JSON heartbeat save file updated.')334                        #with open(self.json_save_file, 'r', encoding='utf-8') as file:335                            #self.json_save_data = json.load(file)336                        json_data = HeartbeatMonitor.read_write_json(self)337                        logger.debug('json_data[\'status\']: ' + str(json_data['status']))338                        self.json_save_data = json_data['data']339                        self.heartbeat_last = self.json_save_data['heartbeat_last']340                        self.flatline_last = self.json_save_data['flatline_last']341                        json_modified_time = json_current_modified_time342                    if (datetime.datetime.now() - self.heartbeat_last) > self.heartbeat_timeout:343                        if (datetime.datetime.now() - self.flatline_last) > self.flatline_timeout:344                            heartbeat_last_delta = "{:.2f}".format(float((datetime.datetime.now() - self.heartbeat_last).total_seconds()) / 60)345                            alert_message = '*Last heartbeat:* ' + heartbeat_last_delta + ' minutes ago.'346                            if self.heartbeat_monitor == 'slack':347                                alert_result = HeartbeatMonitor.send_slack_alert(self, channel_id=self.slack_alert_channel_id_heartbeat,348                                                                                 message=alert_message, flatline=True)349                                logger.debug('alert_result: ' + str(alert_result))350                            elif self.heartbeat_monitor == 'testing':351                                logger.info('Alert Message:    ' + alert_message)352                                #logger.info('Alert Submessage: ' + alert_submessage)353                            self.flatline_last = datetime.datetime.now()354                            logger.debug('self.flatline_last: ' + str(self.flatline_last))355                            self.json_save_data['flatline_last'] = self.flatline_last356                            logger.debug('Saving flatline alert time to json file.')357                            #with open(self.json_save_file, 'w', encoding='utf-8') as file:358                                #json.dump(self.json_save_data, file, indent=4, sort_keys=True, ensure_ascii=False)359                            json_data = HeartbeatMonitor.read_write_json(self, self.json_save_data)360                            logger.debug('json_data[\'status\']: ' + str(json_data['status']))361                    if self.kill_monitor == True:   # This doesn't get executed?362                        logger.debug('Killing monitor loop.')363                        break364                    time.sleep(1)365                logger.debug('Exited monitor loop.')366            except multiprocessing.ProcessError as e:367                logger.exception('multiprocessing.ProcessError raised in heartbeat monitor.')368                logger.exception(e)369                #raise370            except Exception as e:371                logger.exception('Exception raised in heartbeat monitor.')372                logger.exception(e)373                #raise374            except KeyboardInterrupt:375                logger.debug('KeyboardInterrupt in heartbeat monitor.')376                #raise377            finally:378                self.monitor_isrunning = False379                logger.debug('self.monitor_isrunning: ' + str(self.monitor_isrunning))380                """381                alert_message = 'Heartbeat monitor *_DEACTIVATED_* at ' + str(datetime.datetime.now()) + '.'382                if self.heartbeat_monitor == 'slack':383                    alert_result = HeartbeatMonitor.send_slack_alert(self, channel_id=self.slack_alert_channel_id_heartbeat,384                                                                     message=alert_message, submessage=alert_submessage, status_message=True)385                    logger.debug('alert_result: ' + str(alert_result))386                elif self.heartbeat_monitor == 'testing':387                    logger.info('Alert Message:    ' + alert_message)388                    logger.info('Alert Submessage: ' + alert_submessage)389                """390        else:391            self.monitor_states['kill'] = False392            logger.debug('[stop_monitor] self.monitor_states[\'kill\']: ' + str(self.monitor_states['kill']))393            self.monitor_states['isrunning'] = True394            logger.debug('self.monitor_states[\'isrunning\']: ' + str(self.monitor_states['isrunning']))395            try:396                self.monitor_states['heartbeat_last'] = datetime.datetime.now()397                logger.debug('self.monitor_states[\'heartbeat_last\']: ' + str(self.monitor_states['heartbeat_last']))398                alert_message = 'Heartbeat monitor *_ACTIVATED_* at ' + str(self.monitor_states['heartbeat_last']) + '.'399                if self.flatline_alerts_only == True:400                    alert_submessage = 'Regular heartbeat alerts disabled. Only sending alerts on flatline detection.'401                else:402                    alert_submessage = None403                if self.heartbeat_monitor == 'slack':404                    alert_result = HeartbeatMonitor.send_slack_alert(self, channel_id=self.slack_alert_channel_id_heartbeat,405                                                                     message=alert_message, submessage=alert_submessage, status_message=True)406                    logger.debug('alert_result: ' + str(alert_result))407                elif self.heartbeat_monitor == 'testing':408                    logger.info('Alert Message:    ' + alert_message)409                    logger.info('Alert Submessage: ' + str(alert_submessage))410                while (True):411                    if (datetime.datetime.now() - self.monitor_states['heartbeat_last']) > self.heartbeat_timeout and (datetime.datetime.now() - self.monitor_states['flatline_last']) > self.flatline_timeout:412                        # ALERT REQUIRED (HEARTBEAT TIME RESET BY CALLING HeartbeatMonitor.heartbeat())413                        logger.warning('Flatline event detected for ' + self.module_name + '!')414                        heartbeat_last_delta = "{:.2f}".format(float((datetime.datetime.now() - self.monitor_states['heartbeat_last']).total_seconds()) / 60)415                        alert_message = '*Last heartbeat:* ' + heartbeat_last_delta + ' minutes ago.'416                        if self.heartbeat_monitor == 'slack':417                            alert_result = HeartbeatMonitor.send_slack_alert(self, channel_id=self.slack_alert_channel_id_heartbeat, message=alert_message, flatline=True)418                            logger.debug('alert_result: ' + str(alert_result))419                        elif self.heartbeat_monitor == 'testing':420                            logger.info('Alert Message:    ' + alert_message)421                            #logger.info('Alert Submessage: ' + alert_submessage)422                        self.monitor_states['flatline_last'] = datetime.datetime.now()423                        logger.debug('self.monitor_states[\'flatline_last\']: ' + str(self.monitor_states['flatline_last']))424                    #if self.kill_monitor == True:425                    if self.monitor_states['kill'] == True:426                        #logger.debug('self.kill_monitor: ' + str(self.kill_monitor))427                        logger.debug('self.monitor_states[\'kill\']: ' + str(self.monitor_states['kill']))428                        logger.debug('Breaking from monitor loop.')429                        break430                    time.sleep(0.1)431                self.monitor_states['heartbeat_last'] = datetime.datetime.now()432                logger.debug('self.monitor_states[\'heartbeat_last\']: ' + str(self.monitor_states['heartbeat_last']))433            except multiprocessing.ProcessError as e:434                logger.exception('multiprocessing.ProcessError raised in heartbeat monitor.')435                logger.exception(e)436                #raise437            except Exception as e:438                logger.exception('Exception raised in heartbeat monitor.')439                logger.exception(e)440                #raise441            except KeyboardInterrupt:442                logger.debug('KeyboardInterrupt in heartbeat monitor.')443                #raise444            """445            finally:446                #self.monitor_isrunning = False447                #logger.debug('self.monitor_isrunning: ' + str(self.monitor_isrunning))448                self.monitor_states['isrunning'] = False449                logger.debug('self.monitor_states[\'isrunning\']: ' + str(self.monitor_states['isrunning']))450                self.monitor_states['heartbeat_last'] = datetime.datetime.now()451                logger.debug('self.monitor_states[\'heartbeat_last\']: ' + str(self.monitor_states['heartbeat_last']))452                alert_message = 'Heartbeat monitor *_DEACTIVATED_* at ' + str(self.monitor_states['heartbeat_last']) + '.'453                if self.heartbeat_monitor == 'slack':454                    alert_result = HeartbeatMonitor.send_slack_alert(self, channel_id=self.slack_alert_channel_id_heartbeat,455                                                                     message=alert_message, submessage=alert_submessage, status_message=True)456                    logger.debug('alert_result: ' + str(alert_result))457                elif self.heartbeat_monitor == 'testing':458                    logger.info('Alert Message:    ' + alert_message)459                    logger.info('Alert Submessage: ' + alert_submessage)460            """461    def send_slack_alert(self, channel_id, message, submessage=None, flatline=False, status_message=False):462        alert_result = True463        try:464            if status_message == True:465                heartbeat_message = '*Heartbeat monitor status changed.*'466                fallback_message = 'Heartbeat monitor status changed.'467                heartbeat_color = '#FFFF00'468            elif flatline == False:469                heartbeat_message = '*Heartbeat detected.*'470                fallback_message = 'Heartbeat detected.'471                heartbeat_color = '#36A64F'     # Green472            else:473                heartbeat_message = '*WARNING: No heartbeat detected!*'474                fallback_message = 'WARNING: No heartbeat detected!'475                heartbeat_color = '#FF0000'     # Red476            attachment_array =  [{"fallback": fallback_message,477                                  "color": heartbeat_color,   # Green = #36A64F, Blue = #3AA3E3, Yellow = #FFFF00, Orange = #FFA500, Red = #FF0000478                                  "title": "Module: " + self.module_name,479                                  "pretext": message}]480            if submessage != None:481                attachment_array[0]['text'] = submessage482            attachments = json.dumps(attachment_array)483            self.slack_client.api_call(484                'chat.postMessage',485                channel=channel_id,486                text=heartbeat_message,487                username=self.slack_bot_user,488                icon_url=self.slack_bot_icon,489                attachments=attachments490            )491        except Exception as e:492            logger.exception('Exception in heartbeat function.')493            logger.exception(e)494            alert_result = False495        finally:496            return alert_result497if __name__ == '__main__':498    test_config_path = '../../TeslaBot/config/config.ini'499    test_timeout = 0.25500    test_flatline_timeout = 1501    hb = HeartbeatMonitor(module='Testing', monitor='slack', config_path=test_config_path,502                          timeout=test_timeout, flatline_timeout=test_flatline_timeout,503                          flatline_alerts_only=False, test_channel=True)504    try:505        hb.start_monitor()506        """507        while (True):508            #logger.debug('hb.monitor_states[\'isrunning\']: ' + str(hb.monitor_states['isrunning']))509            logger.debug('hb.monitor_isrunning: ' + str(hb.monitor_isrunning))510            #if hb.monitor_states['isrunning'] == True:511            if hb.monitor_isrunning == True:512                break513            time.sleep(1)514        """515        logger.info('Heartbeat monitor ready.')516        for x in range(0, 2):517            heartbeat_message = 'Heartbeat #' + str(x + 1)518            logger.debug('heartbeat_message: ' + heartbeat_message)519            hb.heartbeat(message=heartbeat_message)520            if x < 2:521                time.sleep(5)522        logger.debug('Sleeping for >' + str(test_timeout) + ' min to trigger heartbeat flatline alert.')523        test_delay = (test_timeout * 60) + 1524        time.sleep(test_delay)525        hb.stop_monitor()526        logger.debug('Done.')527    #except multiprocessing.ProcessError as e:528        #logger.exception('multiprocessing.ProcessError raised in main.')529        #logger.exception(e)530    except Exception as e:531        logger.exception('Exception raised.')532        logger.exception(e)533    except KeyboardInterrupt:...

test_heartbeat.py

Source:test_heartbeat.py

...24############25# Fixtures #26############27@pytest.fixture(scope='function', name='fxt_heartbeat')28def heartbeat():29    with patch('voltha.adapters.adtran_onu.heartbeat.structlog.get_logger'):30        return HeartBeat(MagicMock(spec=AdtranOnuHandler), DEVICE_ID)31##############32# Unit tests #33##############34# Basic test of HeartBeat() object creation35def test_heartbeat_init(fxt_heartbeat):36    assert fxt_heartbeat._device_id == DEVICE_ID37# Test Heartbeat.__str__() to ensure proper return value38def test_heartbeat___str__(fxt_heartbeat):39    assert str(fxt_heartbeat) == "HeartBeat: count: 0, miss: 0"40# Test static method constructor for HeartBeat()41def test_heartbeat_create():42    handler = MagicMock(spec=AdtranOnuHandler)...

device_views.py

Source:device_views.py

...24    device["last_heard"] = str(datetime.now())[:-3]25    set_device(device)26    return {}, 20027@app.route("/device/heartbeat", methods=["POST"])28def device_heartbeat():29    heartbeat_info = request.get_json()30    if not heartbeat_info:31        return "Must provide heartbeat information in JSON body", 40032    if "serial" not in heartbeat_info:33        return "Must provide 'serial' in heartbeat information", 40034    if "name" not in heartbeat_info:35        return "Must provide 'name' in heartbeat information", 40036    result, device = get_device(device_name=heartbeat_info["name"])37    if result != "success":38        return "Unknown device name in heartbeat information", 40039    if heartbeat_info["serial"] != device["serial"]:40        return "Serial number in heartbeat information does not match device serial", 40041    device["availability"] = True42    device["last_heard"] = str(datetime.now())[:-3]...

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.