diff --git a/NEWS b/NEWS index 7c882f8e..d1bbbbb3 100644 --- a/NEWS +++ b/NEWS @@ -1,6 +1,7 @@ 2.1.1.dev0 * #1241: For the "recreate" action, actually pass the "--dry-run" flag through to Borg instead of just skipping the Borg call. + * #1242: Fix a regression in which the "spot" check hung while collecting archive contents. 2.1.0 * TL;DR: Many logging, memory, and performance improvements. Mind those breaking changes! diff --git a/borgmatic/execute.py b/borgmatic/execute.py index 34fdebbe..ae6309ee 100644 --- a/borgmatic/execute.py +++ b/borgmatic/execute.py @@ -3,6 +3,7 @@ import contextlib import enum import json import logging +import os import select import subprocess import textwrap @@ -41,7 +42,9 @@ def interpret_exit_code(command, exit_code, borg_local_path=None, borg_exit_code if exit_code == 0: return Exit_status.SUCCESS - if borg_local_path and command[0] == borg_local_path: + parsed_command = command.split(' ', 1) if isinstance(command, str) else command + + if borg_local_path and parsed_command[0] == borg_local_path: # First try looking for the exit code in the borg_exit_codes configuration. for entry in borg_exit_codes or (): if entry.get('code') == exit_code: @@ -163,7 +166,9 @@ def parse_log_line(line, log_level, elevate_stderr, borg_local_path, command): came from stderr and the string "warning:" appears at the start of the log line. In that case, just elevate the log level to a WARN. ''' - if borg_local_path and command[0] == borg_local_path: + parsed_command = command.split(' ', 1) if isinstance(command, str) else command + + if borg_local_path and parsed_command[0] == borg_local_path: log_record = borg_json_log_line_to_record(line, log_level) if log_record: @@ -177,18 +182,20 @@ def parse_log_line(line, log_level, elevate_stderr, borg_local_path, command): return log_line_to_record(line, log_level) -def handle_log_record(log_record, last_lines): +def handle_log_record(log_record, last_lines=None): ''' Given a log record to be logged and a rolling list of last lines, append the record's message to - the last lines. Then (if the log level is not None), log the record. + the last lines (if given). Then (if the log level is not None), log the record. Return the log record. ''' log_message = log_record.getMessage() - last_lines.append(log_message) - if len(last_lines) > ERROR_OUTPUT_MAX_LINE_COUNT: - last_lines.pop(0) + if last_lines is not None: + last_lines.append(log_message) + + if len(last_lines) > ERROR_OUTPUT_MAX_LINE_COUNT: + last_lines.pop(0) if log_record.levelno is not None: logger.handle(log_record) @@ -196,7 +203,205 @@ def handle_log_record(log_record, last_lines): return log_record -def log_outputs( # noqa: PLR0912 +READ_CHUNK_SIZE = 4096 + + +def read_lines(buffer, process, line_separator='\n'): + ''' + Given a Python buffer (like stdout) ready for reading, its process, and a line separator, + repeatedly yield a tuple of (decoded) lines from the buffer until the process has exited. + + It is assumed that this function's generator is used in conjunction with an external select() + call to know when to read more lines. Otherwise, the generator will busywait if it's called in a + tight loop. + ''' + data = '' + + while True: + chunk = os.read(buffer.fileno(), READ_CHUNK_SIZE).decode() + + if not chunk: # EOF + # The process is still running, so we keep running too. + if process.poll() is None: + continue + + break + + data += chunk + lines = [] + + # Split the data into lines, holding back anything leftover that might + # be a partial line. + while True: + separator_position = data.find(line_separator) + + if separator_position == -1: + break + + lines.append(data[:separator_position].rstrip()) + data = data[separator_position + 1 :] + + yield tuple(lines) + + # Yield any leftover data from the end of the buffer. + if data: + yield (data.rstrip(),) + + +Buffer_reader = collections.namedtuple( + 'Buffer_reader', + ('lines', 'process'), +) + + +Process_metadata = collections.namedtuple( + 'Process_metadata', + ('last_lines', 'capture'), +) + + +def log_buffer_lines( + buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr=False +): + ''' + Given a dict from buffer object to Buffer_reader, a dict from subprocess.Popen() instance to + Process_metadata instance, a requested output log level for stdout, Borg's local path, and + whether to capture stderr, read and log any ready output lines from the buffers. Additionally, + if the log level is None for any log record, then yield those log messages for capture. + + This function just does one "turn of the crank" of logging buffer output. It is intended to be + called repeatedly to continue to process buffers. + ''' + if not buffer_readers: + return + + (ready_buffers, _, _) = select.select(buffer_readers.keys(), [], []) + + for ready_buffer in ready_buffers: + reader = buffer_readers[ready_buffer] + + # The "ready" process has exited, but it might be a pipe destination with other + # processes (pipe sources) waiting to be read from. So as a measure to prevent + # hangs, vent all processes when one exits. + if reader.process and reader.process.poll() is not None: + for other_process in process_metadatas: + if ( + other_process.poll() is None + and other_process.stdout + and other_process.stdout not in buffer_readers + ): + # Add the process's output to buffer_readers to ensure it'll get read. + buffer_readers[other_process.stdout] = Buffer_reader( + read_lines(other_process.stdout, other_process), other_process + ) + + try: + lines = next(reader.lines) + except StopIteration: + continue + + for line in lines: + if not line or not reader.process: + continue + + # Keep the last few lines of output in case the process errors and we need the + # output for the exception below. + log_record = handle_log_record( + parse_log_line( + line=line, + log_level=output_log_level, + elevate_stderr=(ready_buffer == reader.process.stderr and not capture_stderr), + borg_local_path=borg_local_path, + command=reader.process.args, + ), + last_lines=process_metadatas[reader.process].last_lines, + ) + + if log_record.levelno is None and process_metadatas[reader.process].capture: + yield log_record.getMessage() + + +def raise_for_process_errors(buffer_readers, process_metadatas, borg_local_path, borg_exit_codes): + ''' + Given a dict from buffer object to Buffer_reader, a dict from subprocess.Popen() instance to + Process_metadata instance, Borg's local path, a sequence of exit code configuration dicts, check + the given processes for error or warning exit codes. If found, vent or kill any running + processes. In the case of an error exit code, raise. In the case of warning, return + Exit_status.WARNING. Otherwise, return None. + ''' + result_status = None + + for process in process_metadatas: + exit_code = process.poll() if buffer_readers else process.wait() + + if exit_code is None: + continue + + exit_status = interpret_exit_code(process.args, exit_code, borg_local_path, borg_exit_codes) + + if exit_status not in {Exit_status.ERROR, Exit_status.WARNING}: + continue + + # Something has gone wrong. So vent each process' output buffer to prevent it from + # hanging. And then kill the process. + for other_process in process_metadatas: + if other_process.poll() is None: + other_process.stdout.read(0) + other_process.kill() + + if exit_status == Exit_status.WARNING: + result_status = Exit_status.WARNING + continue + + last_lines = process_metadatas[process].last_lines + + # If an error occurs, include its output in the raised exception so that we don't + # inadvertently hide error output. + if len(last_lines) >= ERROR_OUTPUT_MAX_LINE_COUNT: + last_lines.insert(0, '...') + + raise subprocess.CalledProcessError( + exit_code, + command_for_process(process), + '\n'.join(last_lines), + ) + + return result_status + + +def log_remaining_buffer_lines( + buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr=False +): + ''' + Given a dict from buffer object to Buffer_reader, a dict from subprocess.Popen() instance to + Process_metadata instance, a requested output log level for stdout, Borg's local path, and + whether to capture stderr, drain and log any remaining output lines from the buffers until + they're empty. Additionally, if the log level is None for any log record, then yield those log + messages for capture. + ''' + for output_buffer, reader in buffer_readers.items(): + if not reader.process: + continue + + for lines in reader.lines: + for line in lines: + log_record = handle_log_record( + parse_log_line( + line=line.rstrip(), + log_level=output_log_level, + elevate_stderr=( + output_buffer == reader.process.stderr and not capture_stderr + ), + borg_local_path=borg_local_path, + command=reader.process.args, + ), + ) + + if log_record.levelno is None and process_metadatas[reader.process].capture: + yield log_record.getMessage() + + +def log_outputs( processes, exclude_stdouts, output_log_level, @@ -221,131 +426,42 @@ def log_outputs( # noqa: PLR0912 buffers. Also note that stdout for a process can be None if output is intentionally not captured, in which case it won't be logged. ''' - # Map from output buffer to sequence of last lines. - process_last_lines = collections.defaultdict(list) - process_for_output_buffer = { - buffer: process + # Map from output buffer to Process_metadata instance. By convention, the last process is the + # process to capture. + process_metadatas = { + process: Process_metadata(last_lines=[], capture=bool(process == processes[-1])) + for process in processes + } + + # Map from buffer to Buffer_reader instance. + buffer_readers = { + buffer: Buffer_reader(read_lines(buffer, process), process) for process in processes if process.stdout or process.stderr for buffer in output_buffers_for_process(process, exclude_stdouts) } - output_buffers = list(process_for_output_buffer.keys()) - process_to_capture = processes[-1] - still_running = True - # Log output for each process until they all exit. - while True: # noqa: PLR1702 - if output_buffers: - (ready_buffers, _, _) = select.select(output_buffers, [], []) + # Log output lines for each process until they all exit. + while True: + yield from log_buffer_lines( + buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr + ) - for ready_buffer in ready_buffers: - ready_process = process_for_output_buffer.get(ready_buffer) - - # The "ready" process has exited, but it might be a pipe destination with other - # processes (pipe sources) waiting to be read from. So as a measure to prevent - # hangs, vent all processes when one exits. - if ready_process and ready_process.poll() is not None: - for other_process in processes: - if ( - other_process.poll() is None - and other_process.stdout - and other_process.stdout not in output_buffers - ): - # Add the process's output to output_buffers to ensure it'll get read. - output_buffers.append(other_process.stdout) - - while True: - line = ready_buffer.readline().rstrip().decode() - if not line or not ready_process: - break - - command = ( - ready_process.args.split(' ') - if isinstance(ready_process.args, str) - else ready_process.args - ) - - # Keep the last few lines of output in case the process errors and we need the - # output for the exception below. - log_record = handle_log_record( - parse_log_line( - line=line, - log_level=output_log_level, - elevate_stderr=( - ready_buffer == ready_process.stderr and not capture_stderr - ), - borg_local_path=borg_local_path, - command=command, - ), - last_lines=process_last_lines[ready_process], - ) - - if log_record.levelno is None and ready_process == process_to_capture: - yield log_record.getMessage() - - if not still_running: + if ( + raise_for_process_errors( + buffer_readers, process_metadatas, borg_local_path, borg_exit_codes + ) + == Exit_status.WARNING + ): break - still_running = False + if all(process.poll() is not None for process in processes): + break - for process in processes: - exit_code = process.poll() if output_buffers else process.wait() - - if exit_code is None: - still_running = True - command = process.args.split(' ') if isinstance(process.args, str) else process.args - continue - - command = process.args.split(' ') if isinstance(process.args, str) else process.args - exit_status = interpret_exit_code(command, exit_code, borg_local_path, borg_exit_codes) - - if exit_status in {Exit_status.ERROR, Exit_status.WARNING}: - last_lines = process_last_lines[process] - - # If an error occurs, include its output in the raised exception so that we don't - # inadvertently hide error output. - for output_buffer in output_buffers_for_process(process, exclude_stdouts): - # Collect any straggling output lines that came in since we last gathered output. - while output_buffer: # pragma: no cover - line = output_buffer.readline().rstrip().decode() - if not line: - break - - log_record = handle_log_record( - parse_log_line( - line=line, - log_level=output_log_level, - elevate_stderr=( - output_buffer == process.stderr and not capture_stderr - ), - borg_local_path=borg_local_path, - command=command, - ), - last_lines=last_lines, - ) - - if log_record.levelno is None and process == process_to_capture: - yield log_record.getMessage() - - if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT: - last_lines.insert(0, '...') - - # Something has gone wrong. So vent each process' output buffer to prevent it from - # hanging. And then kill the process. - for other_process in processes: - if other_process.poll() is None: - other_process.stdout.read(0) - other_process.kill() - - if exit_status == Exit_status.ERROR: - raise subprocess.CalledProcessError( - exit_code, - command_for_process(process), - '\n'.join(last_lines), - ) - - still_running = False - break + # Now that all processes have exited, drain and consume any last output. + yield from log_remaining_buffer_lines( + buffer_readers, process_metadatas, output_log_level, borg_local_path, capture_stderr + ) SECRET_COMMAND_FLAG_NAMES = {'--password'} diff --git a/tests/integration/test_execute.py b/tests/integration/test_execute.py index 2d58bd76..4617fe5d 100644 --- a/tests/integration/test_execute.py +++ b/tests/integration/test_execute.py @@ -8,6 +8,55 @@ from flexmock import flexmock from borgmatic import execute as module +def test_read_lines_yields_single_line(): + process = subprocess.Popen(['echo', 'hi'], stdout=subprocess.PIPE) + + assert tuple(module.read_lines(process.stdout, process)) == (('hi',),) + + +def test_read_lines_yields_single_line_longer_than_chunk_size(): + process = subprocess.Popen( + ['echo', 'this line is longer than the chunk size'], stdout=subprocess.PIPE + ) + + assert tuple(flexmock(module, READ_CHUNK_SIZE=16).read_lines(process.stdout, process)) == ( + (), + (), + ('this line is longer than the chunk size',), + ) + + +def test_read_lines_yields_multiple_lines(): + process = subprocess.Popen(['echo', 'hi\nthere'], stdout=subprocess.PIPE) + + assert tuple(module.read_lines(process.stdout, process)) == (('hi', 'there'),) + + +def test_read_lines_yields_multiple_lines_plus_partial_line(): + process = subprocess.Popen(['echo', '-n', 'hi\nthere\npartial'], stdout=subprocess.PIPE) + + assert tuple(module.read_lines(process.stdout, process)) == (('hi', 'there'), ('partial',)) + + +def test_read_lines_with_longer_running_process_yields_many_lines(): + process = subprocess.Popen( + [ + sys.executable, + '-c', + "import random, string; print('\\n'.join(random.choice(string.ascii_letters) for _ in range(1000)))", + ], + stdout=subprocess.PIPE, + ) + + assert tuple(module.read_lines(process.stdout, process)) + + +def test_read_lines_yields_nothing(): + process = subprocess.Popen(['echo', '-n'], stdout=subprocess.PIPE) + + assert tuple(module.read_lines(process.stdout, process)) == () + + def test_log_outputs_logs_each_line_separately(): hi_record = flexmock( msg='hi', @@ -269,7 +318,7 @@ def test_log_outputs_kills_other_processes_and_raises_when_one_errors(): other_process, (), ).and_return((other_process.stdout,)) - flexmock(other_process).should_receive('kill').once() + flexmock(other_process).should_call('kill').once() with pytest.raises(subprocess.CalledProcessError) as error: tuple( @@ -291,12 +340,6 @@ def test_log_outputs_kills_other_processes_and_returns_when_one_exits_with_warni flexmock(module).should_receive('command_for_process').and_return('grep') process = subprocess.Popen(['grep'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) - flexmock(module).should_receive('interpret_exit_code').with_args( - ['grep'], - None, - 'borg', - None, - ).and_return(module.Exit_status.SUCCESS) flexmock(module).should_receive('interpret_exit_code').with_args( ['grep'], 2, @@ -313,7 +356,7 @@ def test_log_outputs_kills_other_processes_and_returns_when_one_exits_with_warni None, 'borg', None, - ).and_return(module.Exit_status.SUCCESS) + ).and_return(module.Exit_status.STILL_RUNNING) flexmock(module).should_receive('output_buffers_for_process').with_args(process, ()).and_return( (process.stdout,), ) @@ -321,7 +364,7 @@ def test_log_outputs_kills_other_processes_and_returns_when_one_exits_with_warni other_process, (), ).and_return((other_process.stdout,)) - flexmock(other_process).should_receive('kill').once() + flexmock(other_process).should_call('kill').once() assert ( tuple( @@ -370,7 +413,15 @@ def test_log_outputs_vents_other_processes_when_one_exits(): other_process, (process.stdout,), ).and_return((other_process.stdout,)) - flexmock(process.stdout).should_call('readline').at_least().once() + flexmock(module.os).should_call('read').with_args( + process.stderr.fileno(), int + ).at_least().once() + flexmock(module.os).should_call('read').with_args( + process.stdout.fileno(), int + ).at_least().once() + flexmock(module.os).should_call('read').with_args( + other_process.stdout.fileno(), int + ).at_least().once() assert ( tuple( @@ -433,12 +484,6 @@ def test_log_outputs_truncates_long_error_output(): flexmock(module).should_receive('command_for_process').and_return('grep') process = subprocess.Popen(['grep'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) - flexmock(module).should_receive('interpret_exit_code').with_args( - ['grep'], - None, - 'borg', - None, - ).and_return(module.Exit_status.SUCCESS) flexmock(module).should_receive('interpret_exit_code').with_args( ['grep'], 2, @@ -487,8 +532,8 @@ def test_log_outputs_with_unfinished_process_re_polls(): flexmock(module.logger).should_receive('log').never() flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.SUCCESS) - process = subprocess.Popen(['true'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) - flexmock(process).should_receive('poll').and_return(None).and_return(0).times(3) + process = subprocess.Popen(['sleep', '0.001'], stdout=subprocess.PIPE) + flexmock(process).should_call('poll').at_least().times(3) flexmock(module).should_receive('output_buffers_for_process').and_return((process.stdout,)) assert ( diff --git a/tests/unit/test_execute.py b/tests/unit/test_execute.py index 2b1b2b31..8eb727c9 100644 --- a/tests/unit/test_execute.py +++ b/tests/unit/test_execute.py @@ -19,12 +19,18 @@ from borgmatic import execute as module (['borg1'], 1, 'borg1', None, module.Exit_status.WARNING), (['grep'], 100, None, None, module.Exit_status.ERROR), (['grep'], 100, 'borg', None, module.Exit_status.ERROR), + ('grep', 2, None, None, module.Exit_status.ERROR), + ('borg', 2, 'borg', None, module.Exit_status.ERROR), (['borg'], 100, 'borg', None, module.Exit_status.WARNING), (['borg1'], 100, 'borg1', None, module.Exit_status.WARNING), + ('borg', 100, 'borg', None, module.Exit_status.WARNING), + ('borg1', 100, 'borg1', None, module.Exit_status.WARNING), (['grep'], 0, None, None, module.Exit_status.SUCCESS), (['grep'], 0, 'borg', None, module.Exit_status.SUCCESS), (['borg'], 0, 'borg', None, module.Exit_status.SUCCESS), (['borg1'], 0, 'borg1', None, module.Exit_status.SUCCESS), + ('grep', 0, None, None, module.Exit_status.SUCCESS), + ('grep', 0, 'borg', None, module.Exit_status.SUCCESS), # -9 exit code occurs when child process get SIGKILLed. (['grep'], -9, None, None, module.Exit_status.ERROR), (['grep'], -9, 'borg', None, module.Exit_status.ERROR), @@ -174,6 +180,23 @@ def test_parse_log_line_with_borg_command_parses_borg_log_line(): ) +def test_parse_log_line_with_borg_command_parses_borg_log_line_with_string_command(): + record = flexmock() + flexmock(module).should_receive('borg_json_log_line_to_record').and_return(record).once() + flexmock(module).should_receive('log_line_to_record').never() + + assert ( + module.parse_log_line( + 'All done', + module.logging.INFO, + elevate_stderr=False, + borg_local_path='borg', + command='borg do-stuff', + ) + == record + ) + + def test_parse_log_line_without_borg_command_parses_plain_log_line(): record = flexmock() flexmock(module).should_receive('borg_json_log_line_to_record').never() @@ -191,6 +214,23 @@ def test_parse_log_line_without_borg_command_parses_plain_log_line(): ) +def test_parse_log_line_without_borg_command_parses_plain_log_line_with_string_command(): + record = flexmock() + flexmock(module).should_receive('borg_json_log_line_to_record').never() + flexmock(module).should_receive('log_line_to_record').and_return(record).once() + + assert ( + module.parse_log_line( + 'All done', + module.logging.INFO, + elevate_stderr=False, + borg_local_path='borg', + command='totally-not-borg do-stuff', + ) + == record + ) + + def test_parse_log_line_with_elevate_stderr_makes_error_record(): record = flexmock() flexmock(module).should_receive('borg_json_log_line_to_record').never() @@ -262,6 +302,837 @@ def test_handle_log_record_over_max_line_count_trims_and_appends(): assert last_lines == [*original_last_lines[1:], 'line'] +def test_log_buffer_lines_without_buffer_readers_bails(): + flexmock(module.select).should_receive('select').never() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers={}, + process_metadatas={}, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_without_ready_buffers_bails(): + buffer_readers = {flexmock(): flexmock()} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return([], [], []).once() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas={}, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_buffer_and_running_process_handles_each_log_line(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_buffer_and_capture_process_yields_each_line(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return( + flexmock(levelno=None, getMessage=lambda: 'message') + ).twice() + + assert tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) == ('message', 'message') + + +def test_log_buffer_lines_with_ready_buffer_and_log_level_and_capture_process_does_not_yield_each_line(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return( + flexmock(levelno=10, getMessage=lambda: 'message') + ).twice() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_buffer_and_finished_process_vents_other_processes(): + process_stdout = flexmock() + process = flexmock(poll=lambda: 0, stdout=process_stdout, stderr=flexmock(), args=flexmock()) + other_process = flexmock( + poll=lambda: None, stdout=flexmock(), stderr=flexmock(), args=flexmock() + ) + buffer_readers = {process_stdout: module.Buffer_reader(lines=iter((('hi',),)), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=[], capture=False), + other_process: module.Process_metadata(last_lines=[], capture=False), + } + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('read_lines').and_return(iter((('there',),))).once() + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).once() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + # log_buffer_lines() vents other processes by adding them to buffer_readers, with the idea that + # subsequent calls will then read from them. + assert len(buffer_readers) == 2 + + # Assert that the process' buffer has been consumed, indicating that it hasn't been accidentally + # replaced. + assert tuple(buffer_readers[process_stdout].lines) == () + + +def test_log_buffer_lines_with_ready_buffer_and_finished_process_does_not_vent_other_finished_processes(): + process_stdout = flexmock() + process = flexmock(poll=lambda: 0, stdout=process_stdout, stderr=flexmock(), args=flexmock()) + other_process = flexmock(poll=lambda: 0, stdout=flexmock(), stderr=flexmock(), args=flexmock()) + buffer_readers = {process_stdout: module.Buffer_reader(lines=iter((('hi',),)), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=[], capture=False), + other_process: module.Process_metadata(last_lines=[], capture=False), + } + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('read_lines').never() + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).once() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + assert len(buffer_readers) == 1 + assert tuple(buffer_readers[process_stdout].lines) == () + + +def test_log_buffer_lines_with_ready_eof_buffer_and_running_process_skips_it(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=iter(()), process=process)} + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').never() + flexmock(module).should_receive('handle_log_record').never() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_buffer_with_empty_line_skips_it(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=iter((('',),)), process=process)} + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').never() + flexmock(module).should_receive('handle_log_record').never() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_multiple_ready_buffers_and_running_processes_handles_log_lines_from_each(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + other_process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process), + flexmock(): module.Buffer_reader(lines=iter((('foo', 'bar'),)), process=other_process), + } + process_metadatas = { + process: module.Process_metadata(last_lines=[], capture=False), + other_process: module.Process_metadata(last_lines=[], capture=False), + } + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4) + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_multiple_ready_buffers_from_same_running_process_handles_all_log_lines(): + process = flexmock(poll=lambda: None, stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process), + flexmock(): module.Buffer_reader(lines=iter((('foo', 'bar'),)), process=process), + } + process_metadatas = { + process: module.Process_metadata(last_lines=[], capture=False), + } + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').and_return(flexmock()) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4) + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_stderr_buffer_and_running_process_elevates_stderr(): + process_stderr = flexmock() + process = flexmock(poll=lambda: None, stderr=process_stderr, args=flexmock()) + buffer_readers = { + process_stderr: module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=True, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_stdout_buffer_and_running_process_does_not_elevate_stderr(): + process_stdout = flexmock() + process = flexmock(poll=lambda: None, stdout=process_stdout, stderr=flexmock(), args=flexmock()) + buffer_readers = { + process_stdout: module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_buffer_lines_with_ready_stderr_buffer_and_capture_stderr_does_not_elevate_stderr(): + process_stderr = flexmock() + process = flexmock(poll=lambda: None, stderr=process_stderr, args=flexmock()) + buffer_readers = { + process_stderr: module.Buffer_reader(lines=iter((('hi', 'there'),)), process=process) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module.select).should_receive('select').with_args( + buffer_readers.keys(), [], [] + ).and_return(list(buffer_readers.keys()), [], []) + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).twice() + + assert ( + tuple( + module.log_buffer_lines( + buffer_readers=buffer_readers, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + capture_stderr=True, + ) + ) + == () + ) + + +def test_raise_for_process_errors_with_no_processes_bails(): + process = flexmock() + process.should_receive('poll').never() + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + + assert ( + module.raise_for_process_errors( + buffer_readers, + process_metadatas={}, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + is None + ) + + +def test_raise_for_process_errors_with_running_process_bails(): + process = flexmock(poll=lambda: None) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + + assert ( + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + is None + ) + + +def test_raise_for_process_errors_with_running_process_and_no_buffer_readers_waits_and_bails(): + process = flexmock() + process.should_receive('poll').never() + process.should_receive('wait').and_return(None).once() + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + + assert ( + module.raise_for_process_errors( + buffer_readers={}, + process_metadatas=process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + is None + ) + + +def test_raise_for_process_errors_with_successful_process_bails(): + process = flexmock(poll=lambda: 0, args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.SUCCESS) + + assert ( + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + is None + ) + + +def test_raise_for_process_errors_with_warning_process_returns_warning_status(): + process = flexmock(poll=lambda: 1, args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.WARNING) + + assert ( + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + == module.Exit_status.WARNING + ) + + +def test_raise_for_process_errors_with_error_process_raises(): + process = flexmock(poll=lambda: 3, args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=['hi', 'there'], capture=False) + } + flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR) + command = flexmock() + flexmock(module).should_receive('command_for_process').and_return(command) + + with pytest.raises(module.subprocess.CalledProcessError) as error: + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + + assert error.value.returncode == 3 + assert error.value.cmd == command + assert error.value.output == 'hi\nthere' + + +def test_raise_for_process_errors_with_success_process_and_warning_process_returns_warning_status(): + process = flexmock(poll=lambda: 0, args=flexmock()) + other_process = flexmock(poll=lambda: 1, args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=['hi', 'there'], capture=False), + other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False), + } + flexmock(module).should_receive('interpret_exit_code').with_args( + object, 0, object, object + ).and_return(module.Exit_status.SUCCESS) + flexmock(module).should_receive('interpret_exit_code').with_args( + object, 1, object, object + ).and_return(module.Exit_status.WARNING) + flexmock(module).should_receive('command_for_process').never() + + assert ( + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + == module.Exit_status.WARNING + ) + + +def test_raise_for_process_errors_with_warning_process_and_error_process_raises(): + process = flexmock(poll=lambda: 1, args=flexmock()) + other_process = flexmock(poll=lambda: 3, args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=['hi', 'there'], capture=False), + other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False), + } + flexmock(module).should_receive('interpret_exit_code').with_args( + object, 1, object, object + ).and_return(module.Exit_status.WARNING) + flexmock(module).should_receive('interpret_exit_code').with_args( + object, 3, object, object + ).and_return(module.Exit_status.ERROR) + command = flexmock() + flexmock(module).should_receive('command_for_process').and_return(command) + + with pytest.raises(module.subprocess.CalledProcessError) as error: + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + + assert error.value.returncode == 3 + assert error.value.cmd == command + assert error.value.output == 'and\nstuff' + + +def test_raise_for_process_errors_with_warning_process_and_running_process_kills_and_returns_warning_status(): + process = flexmock(poll=lambda: 1, args=flexmock()) + other_process = flexmock( + poll=lambda: None, stdout=flexmock(read=lambda size: None), args=flexmock() + ) + other_process.should_receive('kill').once() + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=['hi', 'there'], capture=False), + other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False), + } + flexmock(module).should_receive('interpret_exit_code').with_args( + object, 1, object, object + ).and_return(module.Exit_status.WARNING) + command = flexmock() + flexmock(module).should_receive('command_for_process').and_return(command) + + assert ( + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + == module.Exit_status.WARNING + ) + + +def test_raise_for_process_errors_with_error_process_and_running_process_kills_and_raises(): + process = flexmock(poll=lambda: 3, args=flexmock()) + other_process = flexmock( + poll=lambda: None, stdout=flexmock(read=lambda size: None), args=flexmock() + ) + other_process.should_receive('kill').once() + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=['hi', 'there'], capture=False), + other_process: module.Process_metadata(last_lines=['and', 'stuff'], capture=False), + } + flexmock(module).should_receive('interpret_exit_code').with_args( + object, 3, object, object + ).and_return(module.Exit_status.ERROR) + command = flexmock() + flexmock(module).should_receive('command_for_process').and_return(command) + + with pytest.raises(module.subprocess.CalledProcessError) as error: + module.raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + + assert error.value.returncode == 3 + assert error.value.cmd == command + assert error.value.output == 'hi\nthere' + + +def test_raise_for_process_errors_with_warning_process_and_long_output_raises_with_truncated_output(): + process = flexmock(poll=lambda: 3, args=flexmock()) + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=process)} + process_metadatas = { + process: module.Process_metadata(last_lines=['hi', 'there'], capture=False) + } + flexmock(module).should_receive('interpret_exit_code').and_return(module.Exit_status.ERROR) + command = flexmock() + flexmock(module).should_receive('command_for_process').and_return(command) + + with pytest.raises(module.subprocess.CalledProcessError) as error: + flexmock(module, ERROR_OUTPUT_MAX_LINE_COUNT=2).raise_for_process_errors( + buffer_readers, + process_metadatas, + borg_local_path=flexmock(), + borg_exit_codes=flexmock(), + ) + + assert error.value.returncode == 3 + assert error.value.cmd == command + assert error.value.output == '...\nhi\nthere' + + +def test_log_remaining_buffer_lines_without_buffer_readers_bails(): + process_metadatas = {flexmock(): module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('parse_log_line').never() + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers={}, + process_metadatas=process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_remaining_buffer_lines_without_reader_process_bails(): + buffer_readers = {flexmock(): module.Buffer_reader(lines=flexmock(), process=None)} + process_metadatas = {flexmock(): module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('parse_log_line').never() + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_remaining_buffer_lines_logs_each_line(): + process = flexmock(stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader( + lines=(('hi', 'there'), ('and', 'stuff')), + process=process, + ) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).times(4) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4) + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_remaining_buffer_lines_with_multiple_buffers_logs_lines_from_each(): + process = flexmock(stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader( + lines=(('hi', 'there'),), + process=process, + ), + flexmock(): module.Buffer_reader( + lines=(('and', 'stuff'),), + process=process, + ), + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).times(4) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4) + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_remaining_buffer_lines_with_stderr_buffer_elevates_stderr(): + stderr = flexmock() + process = flexmock(stderr=stderr, args=flexmock()) + buffer_readers = { + stderr: module.Buffer_reader( + lines=(('hi', 'there'),), + process=process, + ), + flexmock(): module.Buffer_reader( + lines=(('and', 'stuff'),), + process=process, + ), + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=True, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4) + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + +def test_log_remaining_buffer_lines_with_stderr_buffer_and_capture_stderr_does_not_elevate_stderr(): + stderr = flexmock() + process = flexmock(stderr=stderr, args=flexmock()) + buffer_readers = { + stderr: module.Buffer_reader( + lines=(('hi', 'there'),), + process=process, + ), + flexmock(): module.Buffer_reader( + lines=(('and', 'stuff'),), + process=process, + ), + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=False)} + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).times(4) + flexmock(module).should_receive('handle_log_record').and_return(flexmock(levelno=10)).times(4) + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + capture_stderr=True, + ) + ) + == () + ) + + +def test_log_remaining_buffer_lines_with_capture_process_yields_each_line(): + process = flexmock(stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader( + lines=(('hi', 'there'),), + process=process, + ) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)} + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('handle_log_record').and_return( + flexmock(levelno=None, getMessage=lambda: 'message') + ).twice() + + assert tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) == ('message', 'message') + + +def test_log_remaining_buffer_lines_with_log_level_and_capture_process_does_not_yield_each_line(): + process = flexmock(stderr=flexmock(), args=flexmock()) + buffer_readers = { + flexmock(): module.Buffer_reader( + lines=(('hi', 'there'),), + process=process, + ) + } + process_metadatas = {process: module.Process_metadata(last_lines=[], capture=True)} + flexmock(module).should_receive('parse_log_line').with_args( + line=str, log_level=object, elevate_stderr=False, borg_local_path=object, command=object + ).and_return(flexmock()).twice() + flexmock(module).should_receive('handle_log_record').and_return( + flexmock(levelno=10, getMessage=lambda: 'message') + ).twice() + + assert ( + tuple( + module.log_remaining_buffer_lines( + buffer_readers, + process_metadatas, + output_log_level=flexmock(), + borg_local_path=flexmock(), + ) + ) + == () + ) + + def test_mask_command_secrets_masks_password_flag_value(): assert module.mask_command_secrets(('cooldb', '--username', 'bob', '--password', 'pass')) == ( 'cooldb',