e0b4c2ee04
This utility receives input from stdin or a file, buffers up to "n" lines of it, and prints those lines either when it reaches the end of the file/stdin, or when it receives a SIGUSR1 which can be sent to it using the "kill" utility. When it receives a SIGUSR1, it keeps running and tracking the input. Change-Id: I9eca4514378cf24a31002d37e10e58cc3ee63b5d Reviewed-on: https://gem5-review.googlesource.com/c/public/gem5/+/43266 Maintainer: Gabe Black <gabe.black@gmail.com> Tested-by: kokoro <noreply+kokoro@google.com> Reviewed-by: Giacomo Travaglini <giacomo.travaglini@arm.com>
246 lines
8.9 KiB
Python
Executable File
246 lines
8.9 KiB
Python
Executable File
#! /usr/bin/env python3
|
|
#
|
|
# Copyright 2021 Google Inc.
|
|
#
|
|
# Redistribution and use in source and binary forms, with or without
|
|
# modification, are permitted provided that the following conditions are
|
|
# met: redistributions of source code must retain the above copyright
|
|
# notice, this list of conditions and the following disclaimer;
|
|
# redistributions in binary form must reproduce the above copyright
|
|
# notice, this list of conditions and the following disclaimer in the
|
|
# documentation and/or other materials provided with the distribution;
|
|
# neither the name of the copyright holders nor the names of its
|
|
# contributors may be used to endorse or promote products derived from
|
|
# this software without specific prior written permission.
|
|
#
|
|
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
|
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
|
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
|
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
|
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
|
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
|
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
|
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
|
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
|
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
|
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
|
|
|
import argparse
|
|
import copy
|
|
import signal
|
|
import sys
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description='''Circular buffer for text output.
|
|
|
|
To capture the rolling last 25 lines of output from command "command":
|
|
command | logroll.py -n 25
|
|
|
|
While that's running, to see the most recent 25 lines of output without
|
|
interrupting "command", send SIGUSR1 to the logroll.py process:
|
|
kill -s USR1 ${PID of logroll.py}''')
|
|
parser.add_argument('-n', '--lines', default=10, type=int,
|
|
help='Maximum number of lines to buffer at a time.')
|
|
parser.add_argument('file', nargs='?', default=sys.stdin,
|
|
type=argparse.FileType('r', encoding='UTF-8'),
|
|
help='File to read from, default is stdin')
|
|
|
|
args = parser.parse_args()
|
|
|
|
def dump_lines(lines, idx):
|
|
for line in lines[idx:]:
|
|
print(line, end='')
|
|
for line in lines[:idx]:
|
|
print(line, end='')
|
|
|
|
def dump_and_exit(lines, idx):
|
|
dump_lines(lines, idx)
|
|
sys.exit(0)
|
|
|
|
def main(target, incoming):
|
|
idx = 0
|
|
lines = []
|
|
last_idx = target - 1
|
|
|
|
signal.signal(signal.SIGUSR1, lambda num, frame: dump_lines(lines, idx))
|
|
signal.signal(signal.SIGINT, lambda num, frame: dump_and_exit(lines, idx))
|
|
|
|
for line in incoming:
|
|
lines.append(line)
|
|
if idx == last_idx:
|
|
idx = 0
|
|
break
|
|
else:
|
|
idx += 1
|
|
|
|
for lines[idx] in incoming:
|
|
idx = 0 if idx == last_idx else idx + 1
|
|
|
|
dump_lines(lines, idx)
|
|
|
|
if __name__ == '__main__':
|
|
main(target=args.lines, incoming=args.file)
|
|
|
|
|
|
|
|
# Unit tests #
|
|
|
|
class CopyingMock(unittest.mock.MagicMock):
|
|
def __call__(self, *args, **kwargs):
|
|
args = copy.deepcopy(args)
|
|
kwargs = copy.deepcopy(kwargs)
|
|
return super(CopyingMock, self).__call__(*args, **kwargs)
|
|
|
|
class TestLogroll(unittest.TestCase):
|
|
# Test data.
|
|
lines2 = ['First line',
|
|
'Second line']
|
|
lines3 = ['First line',
|
|
'Second line',
|
|
'Third line']
|
|
lines8 = ['First line',
|
|
'Second line',
|
|
'Third line',
|
|
'Fourth line',
|
|
'Fifth line',
|
|
'Sixth line',
|
|
'Seventh line',
|
|
'Eigth line']
|
|
|
|
# Generator which returns lines like a file object would.
|
|
def line_gen(self, lines):
|
|
for line in lines:
|
|
yield line
|
|
|
|
# Generator like above, but which simulates a signal midway through.
|
|
def signal_line_gen(self, lines, pos, sig_dict, signal):
|
|
# Return the first few lines.
|
|
for line in lines[:pos]:
|
|
yield line
|
|
|
|
# Simulate receiving the signal.
|
|
self.assertIn(signal, sig_dict)
|
|
if signal in sig_dict:
|
|
# Pas in junk for the num and frame arguments.
|
|
sig_dict[signal](None, None)
|
|
|
|
# Return the remaining lines.
|
|
for line in lines[pos:]:
|
|
yield line
|
|
|
|
# Set up a mock of signal.signal to record handlers in a dict.
|
|
def mock_signal_dict(self, mock):
|
|
signal_dict = {}
|
|
def signal_signal(num, action):
|
|
signal_dict[num] = action
|
|
mock.side_effect = signal_signal
|
|
return signal_dict
|
|
|
|
# Actual test methods.
|
|
def test_filling_dump_lines(self):
|
|
with unittest.mock.patch('builtins.print') as mock_print:
|
|
dump_lines(self.lines2, len(self.lines2))
|
|
calls = list([ unittest.mock.call(line, end='') for
|
|
line in self.lines2 ])
|
|
mock_print.assert_has_calls(calls)
|
|
|
|
def test_full_dump_lines(self):
|
|
with unittest.mock.patch('builtins.print') as mock_print:
|
|
dump_lines(self.lines2, 0)
|
|
calls = list([ unittest.mock.call(line, end='') for
|
|
line in self.lines2 ])
|
|
mock_print.assert_has_calls(calls)
|
|
|
|
def test_offset_dump_lines(self):
|
|
with unittest.mock.patch('builtins.print') as mock_print:
|
|
dump_lines(self.lines3, 1)
|
|
calls = [ unittest.mock.call(self.lines3[1], end=''),
|
|
unittest.mock.call(self.lines3[2], end=''),
|
|
unittest.mock.call(self.lines3[0], end='') ]
|
|
mock_print.assert_has_calls(calls)
|
|
|
|
def test_dump_and_exit(self):
|
|
with unittest.mock.patch('sys.exit') as mock_sys_exit, \
|
|
unittest.mock.patch(__name__ + '.dump_lines',
|
|
new_callable=CopyingMock) as mock_dump_lines:
|
|
idx = 1
|
|
dump_and_exit(self.lines3, idx)
|
|
mock_dump_lines.assert_called_with(self.lines3, idx)
|
|
mock_sys_exit.assert_called_with(0)
|
|
|
|
def test_filling_main(self):
|
|
with unittest.mock.patch('builtins.print') as mock_print:
|
|
main(5, self.line_gen(self.lines3))
|
|
calls = list([ unittest.mock.call(line, end='') for
|
|
line in self.lines3 ])
|
|
mock_print.assert_has_calls(calls)
|
|
|
|
def test_full_main(self):
|
|
with unittest.mock.patch('builtins.print') as mock_print:
|
|
main(5, self.line_gen(self.lines8))
|
|
calls = list([ unittest.mock.call(line, end='') for
|
|
line in self.lines8[-5:] ])
|
|
mock_print.assert_has_calls(calls)
|
|
|
|
def test_sigusr1_filling_main(self):
|
|
with unittest.mock.patch('signal.signal') as mock_signal, \
|
|
unittest.mock.patch(__name__ + '.dump_lines',
|
|
new_callable=CopyingMock) as mock_dump_lines:
|
|
|
|
signal_dict = self.mock_signal_dict(mock_signal)
|
|
|
|
main(4, self.signal_line_gen(
|
|
self.lines8, 3, signal_dict, signal.SIGUSR1))
|
|
|
|
mock_dump_lines.assert_has_calls([
|
|
unittest.mock.call(self.lines8[0:3], 3 % 4),
|
|
unittest.mock.call(self.lines8[-4:], len(self.lines8) % 4)
|
|
])
|
|
|
|
def test_sigint_filling_main(self):
|
|
with unittest.mock.patch('signal.signal') as mock_signal, \
|
|
unittest.mock.patch(__name__ + '.dump_lines',
|
|
new_callable=CopyingMock) as mock_dump_lines:
|
|
|
|
signal_dict = self.mock_signal_dict(mock_signal)
|
|
|
|
with self.assertRaises(SystemExit):
|
|
main(4, self.signal_line_gen(
|
|
self.lines8, 3, signal_dict, signal.SIGINT))
|
|
|
|
mock_dump_lines.assert_has_calls([
|
|
unittest.mock.call(self.lines8[0:3], 3 % 4),
|
|
])
|
|
|
|
def test_sigusr1_full_main(self):
|
|
with unittest.mock.patch('signal.signal') as mock_signal, \
|
|
unittest.mock.patch(__name__ + '.dump_lines',
|
|
new_callable=CopyingMock) as mock_dump_lines:
|
|
|
|
signal_dict = self.mock_signal_dict(mock_signal)
|
|
|
|
main(4, self.signal_line_gen(
|
|
self.lines8, 5, signal_dict, signal.SIGUSR1))
|
|
|
|
mock_dump_lines.assert_has_calls([
|
|
unittest.mock.call(self.lines8[4:5] + self.lines8[1:4], 5 % 4),
|
|
unittest.mock.call(self.lines8[-4:], len(self.lines8) % 4)
|
|
])
|
|
|
|
def test_sigint_full_main(self):
|
|
with unittest.mock.patch('signal.signal') as mock_signal, \
|
|
unittest.mock.patch(__name__ + '.dump_lines',
|
|
new_callable=CopyingMock) as mock_dump_lines:
|
|
|
|
signal_dict = self.mock_signal_dict(mock_signal)
|
|
|
|
with self.assertRaises(SystemExit):
|
|
main(4, self.signal_line_gen(
|
|
self.lines8, 5, signal_dict, signal.SIGINT))
|
|
|
|
mock_dump_lines.assert_has_calls([
|
|
unittest.mock.call(self.lines8[4:5] + self.lines8[1:4], 5 % 4),
|
|
])
|