Rev 3028: Split note output from StreamTaskDisplay to a separate stream as the old Progress code does. in http://people.ubuntu.com/~robertc/baz2.0/nested-pb
Robert Collins
robertc at robertcollins.net
Tue Nov 20 09:41:47 GMT 2007
At http://people.ubuntu.com/~robertc/baz2.0/nested-pb
------------------------------------------------------------
revno: 3028
revision-id:robertc at robertcollins.net-20071120094136-gh6uaxitdm0psyzd
parent: robertc at robertcollins.net-20071120091730-wutke5gauy1a72cv
committer: Robert Collins <robertc at robertcollins.net>
branch nick: pb.simplify
timestamp: Tue 2007-11-20 20:41:36 +1100
message:
Split note output from StreamTaskDisplay to a separate stream as the old Progress code does.
modified:
bzrlib/progress.py progress.py-20050610070202-df9faaab791964c0
bzrlib/tests/test_progress.py test_progress.py-20060308160359-978c397bc79b7fda
=== modified file 'bzrlib/progress.py'
--- a/bzrlib/progress.py 2007-11-20 09:04:54 +0000
+++ b/bzrlib/progress.py 2007-11-20 09:41:36 +0000
@@ -331,7 +331,14 @@
raise NotImplementedError(self.clear)
def note(self, fmt, *args, **kwargs):
- """Display a note without disrupting the display."""
+ """Display a note without disrupting the display.
+
+ Notes are typically things that the user needs to see rather than
+ transient status messages.
+
+ :param fmt: A format string to show.
+ :param *args: The arguments to pass to the format string.
+ """
raise NotImplementedError(self.note)
def task_changed(self):
@@ -345,9 +352,19 @@
This is a base class for common logic to the dots and tty versions.
"""
- def __init__(self, task, stream):
+ def __init__(self, task, stream, message_stream=None):
+ """Create a StreamTaskDisplay.
+
+ :param task: The task to display.
+ :param stream: The stream to put progress information on.
+ :param message_stream: An optional separate stream for notes to be
+ output on. A common use for this is to set message_stream to
+ stdout, and stream to stderr, so that if the user redirects stderr
+ to /dev/null the notes are still displayed.
+ """
TaskDisplay.__init__(self, task)
- self._stream = stream
+ self.stream = stream
+ self.message_stream = message_stream or stream
self._init_state()
def _init_state(self):
@@ -357,7 +374,7 @@
def note(self, fmt, *args, **kwargs):
self.clear()
- self._stream.write((fmt + '\n') % args)
+ self.message_stream.write((fmt + '\n') % args)
class DotsTaskDisplay(StreamTaskDisplay):
@@ -365,7 +382,7 @@
def clear(self):
if self._last_message is not None:
- self._stream.write('\n')
+ self.stream.write('\n')
self._init_state()
def task_changed(self):
@@ -374,11 +391,11 @@
# at the beginning of a line.
message = self._task.get_message()
if message != self._last_message:
- self._stream.write(message + ':')
+ self.stream.write(message + ':')
self._last_message = message
self._last_current = self._task.current
elif self._task.current != self._last_current:
- self._stream.write('.')
+ self.stream.write('.')
self._last_current = self._task.current
@@ -398,8 +415,9 @@
class TTYTaskDisplay(StreamTaskDisplay):
"""Display the task on a TTY file, using \r's to update a single line."""
- def __init__(self, task, stream, terminal_width=osutils.terminal_width):
- StreamTaskDisplay.__init__(self, task, stream)
+ def __init__(self, task, stream, message_stream=None,
+ terminal_width=osutils.terminal_width):
+ StreamTaskDisplay.__init__(self, task, stream, message_stream)
self.terminal_width = terminal_width()
def clear(self):
=== modified file 'bzrlib/tests/test_progress.py'
--- a/bzrlib/tests/test_progress.py 2007-11-20 09:17:30 +0000
+++ b/bzrlib/tests/test_progress.py 2007-11-20 09:41:36 +0000
@@ -430,77 +430,93 @@
task = CountedTask('message')
output = StringIO()
display = DotsTaskDisplay(task, output)
+ # When not supplied, messages will go to the main output stream.
+ self.assertEqual(display.stream, display.message_stream)
+ self.assertEqual(output, display.stream)
+
+ def test_construct_specific_message_stream(self):
+ task = CountedTask('message')
+ output = StringIO()
+ messages = StringIO()
+ display = DotsTaskDisplay(task, output, messages)
+ self.assertEqual(output, display.stream)
+ self.assertEqual(messages, display.message_stream)
def get_display(self):
"""Get a task, output and display to test with."""
task = CountedTask('message')
output = StringIO()
- display = DotsTaskDisplay(task, output)
- return task, output, display
+ messages = StringIO()
+ display = DotsTaskDisplay(task, output, messages)
+ return task, output, messages, display
def test_clear_after_clear_does_nothing(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.task_changed()
display.clear()
display.clear()
self.assertEqual('message:\n', output.getvalue())
def test_clear_no_output_does_nothing(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.clear()
self.assertEqual('', output.getvalue())
def test_clear_after_output_emits_nl(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.task_changed()
display.clear()
self.assertEqual('message:\n', output.getvalue())
def test_note_before_changes_outputs_note(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.note('Humans read this.')
- self.assertEqual('Humans read this.\n', output.getvalue())
+ self.assertEqual('', output.getvalue())
+ self.assertEqual('Humans read this.\n', messages.getvalue())
def test_note_after_clear_is_contiguous(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.task_changed()
display.clear()
display.note('Humans read this.')
- self.assertEqual('message:\nHumans read this.\n', output.getvalue())
+ self.assertEqual('message:\n', output.getvalue())
+ self.assertEqual('Humans read this.\n', messages.getvalue())
def test_note_after_message_auto_clears(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.task_changed()
display.note('Humans read this.')
- self.assertEqual('message:\nHumans read this.\n', output.getvalue())
+ self.assertEqual('message:\n', output.getvalue())
+ self.assertEqual('Humans read this.\n', messages.getvalue())
def test_note_with_format_variables(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.note('Something %s%d.', "to format", 1)
- self.assertEqual('Something to format1.\n', output.getvalue())
+ self.assertEqual('Something to format1.\n', messages.getvalue())
def test_task_changed_after_clear_emits_message_again(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.task_changed()
display.clear()
display.task_changed()
self.assertEqual('message:\nmessage:', output.getvalue())
def test_task_changed_after_note_emits_message_again(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.task_changed()
display.note('For the user')
display.task_changed()
- self.assertEqual('message:\nFor the user\nmessage:', output.getvalue())
+ self.assertEqual('message:\nmessage:', output.getvalue())
+ self.assertEqual('For the user\n', messages.getvalue())
def test_task_changed_first_call(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
# The first change will display the message
display.task_changed()
self.assertEqual('message:', output.getvalue())
def test_task_changed_no_change(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
# The first change will display the message
display.task_changed()
self.assertEqual('message:', output.getvalue())
@@ -510,7 +526,7 @@
self.assertEqual('message:', output.getvalue())
def test_task_changed_task_has_ticked(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
task.tick()
# The first change will display the message only, even if the current
# position is non-zero.
@@ -530,14 +546,27 @@
# Override terminal width detection for testing
display = TTYTaskDisplay(task, output, terminal_width=lambda:10)
self.assertEqual(10, display.terminal_width)
+ # When not supplied, messages will go to the main output stream.
+ self.assertEqual(display.stream, display.message_stream)
+ self.assertEqual(output, display.stream)
+
+ def test_construct_specific_message_stream(self):
+ task = KnownLengthTask('message', 1)
+ output = _TTYStringIO()
+ messages = StringIO()
+ display = TTYTaskDisplay(task, output, messages)
+ self.assertEqual(output, display.stream)
+ self.assertEqual(messages, display.message_stream)
def get_display(self):
"""Get a task, output and display to test with."""
task = KnownLengthTask('message', 1)
output = _TTYStringIO()
+ messages = StringIO()
# Override terminal width detection for testing
- display = TTYTaskDisplay(task, output, terminal_width=lambda:40)
- return task, output, display
+ display = TTYTaskDisplay(task, output, messages,
+ terminal_width=lambda:40)
+ return task, output, messages, display
def test_usable(self):
"""TTYTaskDisplay's are usable on TTY's with non-dumb terminals."""
@@ -549,14 +578,14 @@
self.assertFalse(TTYTaskDisplay.usable(_TTYStringIO()))
def test_note_before_changes_outputs_note(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.note('Humans read this.')
- self.assertEqual('Humans read this.\n', output.getvalue())
+ self.assertEqual('Humans read this.\n', messages.getvalue())
def test_note_with_format_variables(self):
- task, output, display = self.get_display()
+ task, output, messages, display = self.get_display()
display.note('Something %s%d.', "to format", 1)
- self.assertEqual('Something to format1.\n', output.getvalue())
+ self.assertEqual('Something to format1.\n', messages.getvalue())
class TestDisplayTypeSelection(TestCase):
More information about the bazaar-commits
mailing list