How to use search_in_log method in lettuce-tools

Best Python code snippet using lettuce-tools_python

test.py

Source:test.py

...92        return result93    except BaseException as e:94        logging.error("Failed-finding-files: {}".format(e))95        sys.exit(1)96def search_in_log(filename, root_path, search_string, attachment_filename):97    global today_date98    found_files = find_all_files(filename, root_path)99    results = {}100    for file_name in found_files:101        # insert today date in the grep command102        grep_cmd = "egrep -Hn --color -A 5 -B 5 '{} .*{})'".format(today_date, "|".join(search_string))103        # grep_cmd = grep_cmd.format(today_date)104        # Append file path to the grep command105        grep_cmd = grep_cmd + " " + file_name + " >> " + attachment_filename106        try:107            run = subprocess.Popen(shlex.split(grep_cmd), stdout=subprocess.PIPE)108            stdout = run.stdout.read()109        except BaseException as e:110            logging.error("Failed-running-grep-command: {}".format(e))111            sys.exit(1)112        if stdout.decode("utf-8") != '':113            results[file_name] = stdout.decode("utf-8")114    ret_value = {}115    ret_value[filename] = {"error_detected_file_count": len(results)}116    logging.info("Error-detected-file-count: {}, filename: {}".format(len(results), filename))117    return ret_value118def execute_task(task):119    filenames = task["files"]120    root_path = task["path"]121    search_string = task["search_string"]122    attachment_filename = task["attachment_filename"]123    status = []124    for filename in filenames:125        ret_values = search_in_log(filename, root_path, search_string, attachment_filename)126        status.append(ret_values)127    return status128def main():129    global today_date130    # Get today's date131    today_date = date.today().strftime("%Y/%m/%d")132    # Configure log level133    logging.basicConfig(level=logging.INFO,134                        filename=monitoring_log_path,135                        format='%(asctime)s %(levelname)-8s %(message)s',136                        datefmt='%Y-%m-%d %H:%M:%S')137    # Get configurations138    # config = ConfigParser()139    # config.read(config_file)140    config = None141    with open(config_file) as fp:142        config = yaml.load(fp)143    mail_config = config.get('email')144    mail_sender = mail_config['sender']145    mail_recipients = mail_config['recipients']146    smtp_config = config.get('smtp')147    smtp_server_address = smtp_config['smtp-ip']148    smtp_port = smtp_config['smtp-port']149    tasks = config.get('tasks')150    # Initialize Mail utils151    mailer = MailUtils(smtp_server_address, smtp_port)152    # Search for error logs153    task_status = []154    for task in tasks:155        status = execute_task(task)156        task_status.append(status)157    #ta_config_search_results = search_in_log(ta_config_log_file)158    #ta_config_interface_search_results = search_in_log(ta_config_interface_log_file)159    #ta_config_files = pformat(list(ta_config_search_results.keys()), compact=True, width=1)160    #ta_config_interface_files = pformat(list(ta_config_interface_search_results.keys()), compact=True, width=1)161    formatted_task_status = pformat(task_status, compact=True, width=1)162    # Frame mail body based on the search results163    with open(mail_body_file_path, "w") as fp:164        fp.write("Below are the log files on the ta-core machine, detected with error logs\n")165        current_time = datetime.now().strftime("%H:%M:%S")166        fp.write("Date: {} Time: {}\n\n".format(today_date, current_time))167        fp.write("Error detected in below ta_configuration logs\n")168        fp.write(formatted_task_status)169        # fp.write("\n\nError detected in below ta_configure_interfaces logs\n")170        # fp.write(ta_config_interface_files)171    logging.info("Mail-body-is-written-into-temporary-file: {}".format(mail_body_file_path))172    # os.system("mailx -s '{}' '{}' < {}".format(mail_subject, mail_recepients, mail_body_file_path))...

log_monitor.py

Source:log_monitor.py

...73        return result74    except BaseException as e:75        logging.error("Failed-finding-files: {}".format(e))76        sys.exit(1)77def search_in_log(filename):78    global today_date79    found_files = find_all_files(filename, root_path)80    results = {}81    for file_name in found_files:82        grep_cmd = "egrep -Hn --color -B 5 '{} .*(An error occoured! Removing lockfile and reboot|Not Done)'"83        # insert today date in the grep command84        grep_cmd = grep_cmd.format(today_date)85        # Append file path to the grep command86        grep_cmd = grep_cmd + " " + file_name87        try:88            run = subprocess.Popen(shlex.split(grep_cmd), stdout=subprocess.PIPE)89            stdout = run.stdout.read()90        except BaseException as e:91            logging.error("Failed-running-grep-command: {}".format(e))92            sys.exit(1)93        if stdout.decode("utf-8") != '':94            results[file_name] = stdout.decode("utf-8")95    logging.info("Error-detected-file-count: {}, filename: {}".format(len(results), filename))96    return results97def main():98    global today_date99    # Get today's date100    today_date = date.today().strftime("%Y/%m/%d")101    # Configure log level102    logging.basicConfig(level=logging.INFO,103                        filename=monitoring_log_path,104                        format='%(asctime)s %(levelname)-8s %(message)s',105                        datefmt='%Y-%m-%d %H:%M:%S')106    # Get configurations107    config = ConfigParser()108    config.read(config_file)109    mail_sender = config.get('email', 'sender')110    mail_recipients = config.get('email', 'recipients').split(" ")111    smtp_server_address = config.get('smtp', 'server_address')112    smtp_port = config.get('smtp', 'port')113    # Initialize Mail utils114    mailer = MailUtils(smtp_server_address, smtp_port)115    # Search for error logs116    ta_config_search_results = search_in_log(ta_config_log_file)117    ta_config_interface_search_results = search_in_log(ta_config_interface_log_file)118    ta_config_files = pformat(list(ta_config_search_results.keys()), compact=True, width=1)119    ta_config_interface_files = pformat(list(ta_config_interface_search_results.keys()), compact=True, width=1)120    # Frame mail body based on the search results121    with open(mail_body_file_path, "w") as fp:122        fp.write("Below are the log files on the ta-core machine, detected with error logs\n")123        current_time = datetime.now().strftime("%H:%M:%S")124        fp.write("Date: {} Time: {}\n\n".format(today_date, current_time))125        fp.write("Error detected in below ta_configuration logs\n")126        fp.write(ta_config_files)127        fp.write("\n\nError detected in below ta_configure_interfaces logs\n")128        fp.write(ta_config_interface_files)129    logging.info("Mail-body-is-written-into-temporary-file: {}".format(mail_body_file_path))130    # os.system("mailx -s '{}' '{}' < {}".format(mail_subject, mail_recepients, mail_body_file_path))131    msg = ''...

test_usr1.py

Source:test_usr1.py

...27        assert (28            self.wait_for_record(r'"GET /usr1 HTTP/1.1" 200 0 "-" "-"', log)29            is not None30        ), 'reopen 2'31        assert self.search_in_log(r'/usr1', log_new) is None, 'rename new 2'32    def test_usr1_unit_log(self, temp_dir, unit_pid):33        self.load('log_body')34        log_new = 'new.log'35        log_path = temp_dir + '/unit.log'36        log_path_new = temp_dir + '/' + log_new37        os.rename(log_path, log_path_new)38        Log.swap(log_new)39        try:40            body = 'body_for_a_log_new\n'41            assert self.post(body=body)['status'] == 20042            assert (43                self.wait_for_record(body, log_new) is not None44            ), 'rename new'45            assert not os.path.isfile(log_path), 'rename old'46            os.kill(unit_pid, signal.SIGUSR1)47            assert waitforfiles(log_path), 'reopen'48            body = 'body_for_a_log_unit\n'49            assert self.post(body=body)['status'] == 20050            assert self.wait_for_record(body) is not None, 'rename new'51            assert self.search_in_log(body, log_new) is None, 'rename new 2'52        finally:53            # merge two log files into unit.log to check alerts54            with open(log_path, 'r', errors='ignore') as unit_log:55                log = unit_log.read()56            with open(log_path, 'w') as unit_log, open(57                log_path_new, 'r', errors='ignore'58            ) as unit_log_new:59                unit_log.write(unit_log_new.read())60                unit_log.write(log)...

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.