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

How to use close_log_file method in autotest

Best Python code snippet using autotest_python

logger.py

Source:logger.py

...21    pass222324def stop_logging():25    EventLogger().close_log_file()26    DataLogger().close_log_file()27    MsgLogger().close_log_file()282930class MsgLogger(Observer, metaclass=Singleton):3132    def __init__(self):33        Observer.__init__(self)34        self.observe(self.msg_rxd)35        self.fp = None3637    def msg_rxd(self, msg):38        print(39            f'{get_tstamp()},{msg.get_msg_name()},{msg.get_payload()[0]},{msg.f16.ack},{msg.f16.msg_type},{msg.f16.uniq_id:04X}',40            file=self.fp)4142    def open_log_file(self):43        self.fp = open(Path(log_folder_abspath) / f'msg_log_{dt.now().isoformat().replace(":", ".")}.csv', 'w')44        print(f'timestamp,msg_name,payload,ack,msg_type,uniq_id', file=self.fp)4546    def close_log_file(self):47        if self.fp: self.fp.close()48        self.fp = None4950    def __del__(self):51        self.close_log_file()525354class EventLogger(Observer, metaclass=Singleton):5556    def __init__(self):57        Observer.__init__(self)58        self.observe(self.msg_rxd)59        self.eventlog_uidata = []60        self.fp = None6162    def msg_rxd(self, msg):63        ts_now = get_tstamp()64        if not msg.is_eventlog(): return65        try:66            msg_name = msg.get_msg_name()67        except (KeyError, ValueError):68            return69        payloads = msg.get_payload()70        resp = 'positive' if msg.f16.ack >= 0 else 'negative'71        try:72            disp_texts = df_eventlog_type.loc[df_eventlog_type['msg_id'] == msg_name, resp].iloc[0].split(',')73            if len(disp_texts) != len(payloads):74                disp_texts += [disp_texts[-1] for _ in range(len(payloads) - len(disp_texts))]75        except TypeError:76            print(f'Type error with {msg_name}')77        except:78            disp_texts = msg_name7980        if msg.is_assertlog():81            self.eventlog_uidata.append((ts_now, f'id:{payloads[0]}, f:{payloads[1]}, l:{payloads[2]}', disp_texts[0], msg.f16.ack))82            return8384        for i, payload in enumerate(payloads):85            disp_text = disp_texts if type(disp_texts) is str else disp_texts[i]86            if not msg.is_payload_timestamp():87                self.eventlog_uidata.append((ts_now, payload, disp_text, msg.f16.ack))88            else:89                self.eventlog_uidata.append((ts_now, '-', disp_text, msg.f16.ack))90            print(f'{ts_now},{payload},{msg_name},{disp_text}', file=self.fp)9192    def open_log_file(self):93        self.fp = open(Path(log_folder_abspath) / f'event_log_{dt.now().isoformat().replace(":", ".")}.txt', 'w')9495    def close_log_file(self):96        if self.fp: self.fp.close()97        self.fp = None9899    def get_ui_data(self):100        while len(self.eventlog_uidata) != 0:101            yield self.eventlog_uidata.pop(0)102103    def __del__(self):104        self.close_log_file()105106107class DataLogger(Observer, metaclass=Singleton):108109    def __init__(self, datalog_interval_ms=30):110        Observer.__init__(self)111        self.fp = open(Path(log_folder_abspath) / f'data_log_{dt.now().isoformat().replace(":", ".")}.csv', 'w')112        self.observe(self.msg_rxd)113        self.datakeep = []114        self.indexkeep, index = {}, 0115        for name, unit, disp_id, en in zip(df_datalog_type.datalog_column_name.values,116                                           df_datalog_type.datalog_column_unit.values,117                                           df_datalog_type.dom_display_id.values, df_datalog_type.enable.values):118            if en:119                self.datakeep.append([name, '-', unit, disp_id])120                self.indexkeep[name] = index121                index += 1122        self.log_in_progress = False123        self.log_trigger_thread = threading.Thread(target=self.log_trigger_fn, args=(datalog_interval_ms,))124125    def msg_rxd(self, msg):126        if not msg.is_datalog(): return127        try:128            msg_name = msg.get_msg_name()129            name = df_datalog_type.loc[df_datalog_type['msg_id'] == msg_name, 'datalog_column_name'].iloc[0]130            if name is None:131                raise ValueError132            index = self.indexkeep[name]133            disp_payload = msg.get_payload_str()[0]134            self.datakeep[index][1] = float(f'{disp_payload:.04}') if type(disp_payload) is float else disp_payload135        except (IndexError, KeyError, ValueError):136            return137138    def log(self):139        try:140            print(get_tstamp(), end=',', file=self.fp)141            for data in self.datakeep:142                print(data[1], end=',', file=self.fp)143            print('', file=self.fp)144        except ValueError:145            pass146147    def open_log_file(self):148        print('system timestamp', end=',', file=self.fp)149        for data in self.datakeep:150            print(data[0], end=',', file=self.fp)151        print('', file=self.fp)152        print('time', end=',', file=self.fp)153        for unit, en in zip(df_datalog_type.datalog_column_unit.values, df_datalog_type.enable.values):154            if en: print(unit, end=',', file=self.fp)155        print('', file=self.fp)156        self.log_in_progress = True157        self.log_trigger_thread.start()158159    def close_log_file(self):160        if self.fp: self.fp.close()161        self.fp = None162        self.log_in_progress = False163        self.log_trigger_thread.join()164165    def log_trigger_fn(self, wait_time_ms):166        while self.log_in_progress:167            sleep(wait_time_ms / 1000)168            self.log()169170    def __del__(self):
...

copy_raw_data.py

Source:copy_raw_data.py

...31        if api_util.copy_remote_file(raw_data_login, remote_run_info_file, cegr_run_info_path, lh):32            cegr_run_info_file = cegr_run_info_path33        else:34            lh.write('Error copying file\n%s\nfrom host\n%s\nto local file\n%s\n\n' % (remote_run_info_file, raw_data_login, cegr_run_info_path))35            api_util.close_log_file(lh, SCRIPT_NAME)36            sys.exit(1)37        current_run_dir = api_util.get_current_run_directory(cegr_run_info_file)38        if current_run_dir is None:39            lh.write('\nRequired entry for current_run_dir missing from cegr_run_info.txt (this should be the first non-blank non-comment line in the file).\n')40            api_util.close_log_file(lh, SCRIPT_NAME)41            sys.exit(1)42        run_complete_file_path = os.path.join(current_run_dir, run_complete_file_name)43        lh.write('Current run directory on remote server: %s\n' % current_run_dir)44        if api_util.exists_remote(raw_data_login, run_complete_file_path, lh):45            rc = api_util.copy_remote_directory_of_files(raw_data_login, current_run_dir, raw_data_directory, lh)46            if rc == 0:47                # All files were successfully copied, so remove the remote cegr_run_info.txt file.48                rc = api_util.remove_remote_file(raw_data_login, remote_run_info_file, lh)49                if rc == 0:50                    # Let everyone know we've finished.51                    api_util.create_script_complete_file(log_dir, SCRIPT_NAME)52                    break53                else:54                    lh.write('Error removing file %s from remote server, return code: %d.\n' % (remote_run_info_file, rc))55                    api_util.close_log_file(lh, SCRIPT_NAME)56                    sys.exit(1)57            else:58                lh.write('Error copying raw data files from remote server, return code: %d.\n' % rc)59                api_util.close_log_file(lh, SCRIPT_NAME)60                sys.exit(1)61        else:62            lh.write('The file named\n%s\ndoes not exist on remote host %s, so the run must not be complete - will check again in 5 minutes.\n' % (run_complete_file_path, raw_data_login))63            time.sleep(300)64    else:65        lh.write('The file\n%s\ndoes not exist on remote host %s, so will check again in 5 minutes.\n' % (remote_run_info_file, raw_data_login))66        time.sleep(300)...

log_handler.py

Source:log_handler.py

...3def create_log_file(file_name):4        #Create txt file5        filePathLogFile = "files/logs/" + str(file_name) +"_log.txt"6        logFile = open(filePathLogFile, "w")7        close_log_file(logFile)8        logFile.close()9        10def open_log_file(filename):11        #Open txt file12        filePathLogFile = "files/logs/" + str(filename) +"_log.txt"13        logFile = open(filePathLogFile, "a")14        return logFile15def log(filename, content):16        logFile = open_log_file(filename)17        logFile.write(content + "\n")18        close_log_file(logFile)19        return logFile20def close_log_file(logFile):...

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.