How to use write_lastlines_file method in autotest

Best Python code snippet using autotest_python

monitors_util.py

Source:monitors_util.py Github

copy

Full Screen

...148 # Then translate this into a reverse line number149 # (count newlines that occur afterward)150 reverse_lineno = target_data.count('\n', loc + len(lastlines))151 return reverse_lineno152def write_lastlines_file(lastlines_dirpath, path, data):153 """Write data to lastlines file for path.154 Args:155 lastlines_dirpath: str; Dirpath to store lastlines files to.156 path: str; Filepath to source file that data comes from.157 data: str;158 Returns:159 str; Filepath that lastline data was written to.160 """161 underscored = path.replace('/', '_')162 dest_path = os.path.join(lastlines_dirpath, underscored)163 open(dest_path, 'w').write(data)164 return dest_path165def nonblocking(pipe):166 """Set python file object to nonblocking mode.167 This allows us to take advantage of pipe.read()168 where we don't have to specify a buflen.169 Cuts down on a few lines we'd have to maintain.170 Args:171 pipe: file; File object to modify172 Returns: pipe173 """174 flags = fcntl.fcntl(pipe, fcntl.F_GETFL)175 fcntl.fcntl(pipe, fcntl.F_SETFL, flags| os.O_NONBLOCK)176 return pipe177def launch_tails(follow_paths, lastlines_dirpath=None):178 """Launch a tail process for each follow_path.179 Args:180 follow_paths: list;181 lastlines_dirpath: str;182 Returns:183 tuple; (procs, pipes) or184 ({path: subprocess.Popen, ...}, {file: path, ...})185 """186 if lastlines_dirpath and not os.path.exists(lastlines_dirpath):187 os.makedirs(lastlines_dirpath)188 tail_cmd = ('/usr/bin/tail', '--retry', '--follow=name')189 procs = {} # path -> tail_proc190 pipes = {} # tail_proc.stdout -> path191 for path in follow_paths:192 cmd = list(tail_cmd)193 if lastlines_dirpath:194 reverse_lineno = lookup_lastlines(lastlines_dirpath, path)195 if reverse_lineno is None:196 reverse_lineno = 1197 cmd.append('--lines=%d' % reverse_lineno)198 cmd.append(path)199 tail_proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)200 procs[path] = tail_proc201 pipes[nonblocking(tail_proc.stdout)] = path202 return procs, pipes203def poll_tail_pipes(pipes, lastlines_dirpath=None, waitsecs=5):204 """Wait on tail pipes for new data for waitsecs, return any new lines.205 Args:206 pipes: dict; {subprocess.Popen: follow_path, ...}207 lastlines_dirpath: str; Path to write lastlines to.208 waitsecs: int; Timeout to pass to select209 Returns:210 tuple; (lines, bad_pipes) or ([line, ...], [subprocess.Popen, ...])211 """212 lines = []213 bad_pipes = []214 # Block until at least one is ready to read or waitsecs elapses215 ready, _, _ = select.select(pipes.keys(), (), (), waitsecs)216 for fi in ready:217 path = pipes[fi]218 data = fi.read()219 if len(data) == 0:220 # If no data, process is probably dead, add to bad_pipes221 bad_pipes.append(fi)222 continue223 if lastlines_dirpath:224 # Overwrite the lastlines file for this source path225 # Probably just want to write the last 1-3 lines.226 write_lastlines_file(lastlines_dirpath, path, data)227 for line in data.splitlines():228 lines.append('[%s]\t%s\n' % (path, line))229 return lines, bad_pipes230def snuff(subprocs):231 """Helper for killing off remaining live subprocesses.232 Args:233 subprocs: list; [subprocess.Popen, ...]234 """235 for proc in subprocs:236 if proc.poll() is None:237 os.kill(proc.pid, signal.SIGKILL)238 proc.wait()239def follow_files(follow_paths, outstream, lastlines_dirpath=None, waitsecs=5):240 """Launch tail on a set of files and merge their output into outstream....

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful