How to use write_lastlines_file method in autotest

Best Python code snippet using autotest_python

monitors_util.py

Source:monitors_util.py

...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....

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.