Best Python code snippet using autotest_python
crashcollect.py
Source:crashcollect.py
...15 get_crashdumps(host, test_start_time)16 if wait_for_machine_to_recover(host):17 # run any site-specific collection18 get_site_crashinfo(host, test_start_time)19 crashinfo_dir = get_crashinfo_dir(host)20 collect_messages(host)21 collect_log_file(host, "/var/log/monitor-ssh-reboots", crashinfo_dir)22 collect_command(host, "dmesg", os.path.join(crashinfo_dir, "dmesg"))23 collect_uncollected_logs(host)24def wait_for_machine_to_recover(host, hours_to_wait=4.0):25 """Wait for a machine (possibly down) to become accessible again.26 @param host: A RemoteHost instance to wait on27 @param hours_to_wait: Number of hours to wait before giving up28 @returns: True if the machine comes back up, False otherwise29 """30 current_time = time.strftime("%b %d %H:%M:%S", time.localtime())31 if host.is_up():32 logging.info("%s already up, collecting crash info", host.hostname)33 return True34 logging.info("Waiting four hours for %s to come up (%s)",35 host.hostname, current_time)36 if not host.wait_up(timeout=hours_to_wait * 3600):37 logging.warning("%s down, unable to collect crash info",38 host.hostname)39 return False40 else:41 logging.info("%s is back up, collecting crash info", host.hostname)42 return True43def get_crashinfo_dir(host):44 """Find and if necessary create a directory to store crashinfo in.45 @param host: The RemoteHost object that crashinfo will be collected from46 @returns: The path to an existing directory for writing crashinfo into47 """48 host_resultdir = getattr(getattr(host, "job", None), "resultdir", None)49 if host_resultdir:50 infodir = host_resultdir51 else:52 infodir = os.path.abspath(os.getcwd())53 infodir = os.path.join(infodir, "crashinfo.%s" % host.hostname)54 if not os.path.exists(infodir):55 os.mkdir(infodir)56 return infodir57def collect_log_file(host, log_path, dest_path):58 """Collects a log file from the remote machine.59 Log files are collected from the remote machine and written into the60 destination path. If dest_path is a directory, the log file will be named61 using the basename of the remote log path.62 @param host: The RemoteHost to collect logs from63 @param log_path: The remote path to collect the log file from64 @param dest_path: A path (file or directory) to write the copies logs into65 """66 logging.info("Collecting %s...", log_path)67 try:68 host.get_file(log_path, dest_path, preserve_perm=False)69 except Exception:70 logging.warning("Collection of %s failed", log_path)71def collect_command(host, command, dest_path):72 """Collects the result of a command on the remote machine.73 The standard output of the command will be collected and written into the74 desitionation path. The destination path is assumed to be filename and75 not a directory.76 @param host: The RemoteHost to collect from77 @param command: A shell command to run on the remote machine and capture78 the output from.79 @param dest_path: A file path to write the results of the log into80 """81 logging.info("Collecting '%s' ...", command)82 devnull = open("/dev/null", "w")83 try:84 try:85 result = host.run(command, stdout_tee=devnull).stdout86 utils.open_write_close(dest_path, result)87 except Exception, e:88 logging.warning("Collection of '%s' failed:\n%s", command, e)89 finally:90 devnull.close()91def collect_uncollected_logs(host):92 """Collects any leftover uncollected logs from the client.93 @param host: The RemoteHost to collect from94 """95 if host.job:96 try:97 logs = host.job.get_client_logs()98 for hostname, remote_path, local_path in logs:99 if hostname == host.hostname:100 logging.info("Retrieving logs from %s:%s into %s",101 hostname, remote_path, local_path)102 host.get_file(remote_path + "/", local_path + "/")103 except Exception, e:104 logging.warning("Error while trying to collect stranded "105 "Autotest client logs: %s", e)106def collect_messages(host):107 """Collects the 'new' contents of /var/log/messages.108 If host.VAR_LOG_MESSAGE_COPY_PATH is on the remote machine, collects109 the contents of /var/log/messages excluding whatever initial contents110 are already present in host.VAR_LOG_MESSAGE_COPY_PATH. If it is not111 present, simply collects the entire contents of /var/log/messages.112 @param host: The RemoteHost to collect from113 """114 crashinfo_dir = get_crashinfo_dir(host)115 try:116 # paths to the messages files117 messages = os.path.join(crashinfo_dir, "messages")118 messages_raw = os.path.join(crashinfo_dir, "messages.raw")119 messages_at_start = os.path.join(crashinfo_dir, "messages.at_start")120 # grab the files from the remote host121 collect_log_file(host, host.VAR_LOG_MESSAGES_COPY_PATH,122 messages_at_start)123 collect_log_file(host, "/var/log/messages", messages_raw)124 # figure out how much of messages.raw to skip125 if os.path.exists(messages_at_start):126 # if the first lines of the messages at start should match the127 # first lines of the current messages; if they don't then messages128 # has been erase or rotated and we just grab all of it...
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!!