mirror of
https://github.com/borgmatic-collective/borgmatic.git
synced 2026-07-22 02:03:01 +02:00
Fix "spot" check hang (#1242).
Reviewed-on: https://projects.torsion.org/borgmatic-collective/borgmatic/pulls/1247
This commit is contained in:
Dan Helfman
@@ -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!
|
||||
|
||||
+240
-124
@@ -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'}
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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',
|
||||
|
||||
Reference in New Issue
Block a user