# -*- coding: utf-8 -*- """This module tests Halo spinners. """ import io import os import re import sys import time import unittest try: from cStringIO import StringIO except ImportError: from io import StringIO from spinners.spinners import Spinners from halo import Halo from halo._utils import get_terminal_columns, is_supported from tests._utils import strip_ansi, find_colors, encode_utf_8_text, decode_utf_8_text from termcolor import COLORS if sys.version_info.major == 2: get_coded_text = encode_utf_8_text else: get_coded_text = decode_utf_8_text if is_supported(): frames = [get_coded_text(frame) for frame in Spinners['dots'].value['frames']] default_spinner = Spinners['dots'].value else: frames = [get_coded_text(frame) for frame in Spinners['line'].value['frames']] default_spinner = Spinners['line'].value class SpecificException(Exception): """A unique exc class we know only our tests would raise""" class TestHalo(unittest.TestCase): """Test Halo enum for attribute values. """ TEST_FOLDER = os.path.dirname(os.path.abspath(__file__)) def setUp(self): """Set up things before beginning of each test. """ self._stream_file = os.path.join(self.TEST_FOLDER, 'test.txt') self._stream = io.open(self._stream_file, 'w+') self._stream_no_tty = StringIO() def _get_test_output(self, no_ansi=True): """Clean the output from stream and return it in list form. Returns ------- list Clean output from stream """ self._stream.seek(0) data = self._stream.readlines() output = {} output_text = [] output_colors = [] for line in data: clean_line = strip_ansi(line.strip('\n')) if no_ansi else line.strip('\n') if clean_line != '': output_text.append(get_coded_text(clean_line)) colors_found = find_colors(line.strip('\n')) if colors_found: tmp = [] for color in colors_found: tmp.append(re.sub(r'[^0-9]', '', color, flags=re.I)) output_colors.append(tmp) output['text'] = output_text output['colors'] = output_colors return output def test_basic_spinner(self): """Test the basic of basic spinners. """ spinner = Halo(text='foo', spinner='dots', stream=self._stream) spinner.start() time.sleep(1) spinner.stop() output = self._get_test_output()['text'] self.assertEqual(output[0], '{} foo'.format(frames[0])) self.assertEqual(output[1], '{} foo'.format(frames[1])) self.assertEqual(output[2], '{} foo'.format(frames[2])) def test_text_spinner_color(self): """Test basic spinner with available colors color (both spinner and text) """ for color, color_int in COLORS.items(): self._stream_file = os.path.join(self.TEST_FOLDER, 'test.txt') self._stream = io.open(self._stream_file, 'w+') spinner = Halo( text='foo', text_color=color, color=color, spinner='dots', stream=self._stream ) spinner.start() time.sleep(1) spinner.stop() output = self._get_test_output()['colors'] # check if spinner colors match self.assertEqual(color_int, int(output[0][0])) self.assertEqual(color_int, int(output[1][0])) self.assertEqual(color_int, int(output[2][0])) # check if text colors match self.assertEqual(color_int, int(output[0][1])) self.assertEqual(color_int, int(output[1][1])) self.assertEqual(color_int, int(output[2][1])) def test_spinner_getter(self): instance = Halo() if is_supported(): default_spinner_value = "dots" else: default_spinner_value = "line" instance.spinner = default_spinner_value self.assertEqual(default_spinner, instance.spinner) instance.spinner = "This_spinner_do_not_exist" self.assertEqual(default_spinner, instance.spinner) instance.spinner = -123 self.assertEqual(default_spinner, instance.spinner) def test_text_stripping(self): """Test the text being stripped before output. """ spinner = Halo(text='foo\n', spinner='dots', stream=self._stream) spinner.start() time.sleep(1) spinner.succeed('foo\n') output = self._get_test_output()['text'] self.assertEqual(output[0], '{} foo'.format(frames[0])) self.assertEqual(output[1], '{} foo'.format(frames[1])) self.assertEqual(output[2], '{} foo'.format(frames[2])) pattern = re.compile(r'(✔|v) foo', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) def test_text_ellipsing(self): """Test the text gets ellipsed if it's too long """ text = 'This is a text that it is too long. In fact, it exceeds the eighty column standard ' \ 'terminal width, which forces the text frame renderer to add an ellipse at the end of the ' \ 'text. ' * 6 spinner = Halo(text=text, spinner='dots', stream=self._stream) spinner.start() time.sleep(1) spinner.succeed('End!') output = self._get_test_output()['text'] terminal_width = get_terminal_columns() # -6 of the ' (...)' ellipsis, -2 of the spinner and space self.assertEqual(output[0], '{} {} (...)'.format(frames[0], text[:terminal_width - 6 - 2])) self.assertEqual(output[1], '{} {} (...)'.format(frames[1], text[:terminal_width - 6 - 2])) self.assertEqual(output[2], '{} {} (...)'.format(frames[2], text[:terminal_width - 6 - 2])) pattern = re.compile(r'(✔|v) End!', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) def test_text_animation(self): """Test the text gets animated when it is too long """ text = 'This is a text that it is too long. In fact, it exceeds the eighty column standard ' \ 'terminal width, which forces the text frame renderer to add an ellipse at the end of the ' \ 'text. ' * 6 spinner = Halo(text=text, spinner='dots', stream=self._stream, animation='marquee') spinner.start() time.sleep(1) spinner.succeed('End!') output = self._get_test_output()['text'] terminal_width = get_terminal_columns() self.assertEqual(output[0], '{} {}'.format(frames[0], text[:terminal_width - 2])) self.assertEqual(output[1], '{} {}'.format(frames[1], text[1:terminal_width - 1])) self.assertEqual(output[2], '{} {}'.format(frames[2], text[2:terminal_width])) pattern = re.compile(r'(✔|v) End!', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) def test_context_manager(self): """Test the basic of basic spinners used through the with statement. """ with Halo(text='foo', spinner='dots', stream=self._stream): time.sleep(1) output = self._get_test_output()['text'] self.assertEqual(output[0], '{} foo'.format(frames[0])) self.assertEqual(output[1], '{} foo'.format(frames[1])) self.assertEqual(output[2], '{} foo'.format(frames[2])) def test_context_manager_exceptions(self): """Test Halo context manager allows exceptions to bubble up """ with self.assertRaises(SpecificException): with Halo(text='foo', spinner='dots', stream=self._stream): raise SpecificException def test_decorator_spinner(self): """Test basic usage of spinners with the decorator syntax.""" @Halo(text="foo", spinner="dots", stream=self._stream) def decorated_function(): time.sleep(1) decorated_function() output = self._get_test_output()['text'] self.assertEqual(output[0], '{} foo'.format(frames[0])) self.assertEqual(output[1], '{} foo'.format(frames[1])) self.assertEqual(output[2], '{} foo'.format(frames[2])) def test_decorator_exceptions(self): """Test Halo decorator allows exceptions to bubble up""" @Halo(text="foo", spinner="dots", stream=self._stream) def decorated_function(): raise SpecificException with self.assertRaises(SpecificException): decorated_function() def test_initial_title_spinner(self): """Test Halo with initial title. """ spinner = Halo('bar', stream=self._stream) spinner.start() time.sleep(1) spinner.stop() output = self._get_test_output()['text'] self.assertEqual(output[0], '{} bar'.format(frames[0])) self.assertEqual(output[1], '{} bar'.format(frames[1])) self.assertEqual(output[2], '{} bar'.format(frames[2])) def test_id_not_created_before_start(self): """Test Spinner ID not created before start. """ spinner = Halo(stream=self._stream) self.assertEqual(spinner.spinner_id, None) def test_ignore_multiple_start_calls(self): """Test ignoring of multiple start calls. """ spinner = Halo(stream=self._stream) spinner.start() spinner_id = spinner.spinner_id spinner.start() self.assertEqual(spinner.spinner_id, spinner_id) spinner.stop() def test_chaining_start(self): """Test chaining start with constructor """ spinner = Halo(stream=self._stream).start() spinner_id = spinner.spinner_id self.assertIsNotNone(spinner_id) spinner.stop() def test_succeed(self): """Test succeed method """ spinner = Halo(stream=self._stream) spinner.start('foo') spinner.succeed('foo') output = self._get_test_output()['text'] pattern = re.compile(r'(✔|v) foo', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) spinner.stop() def test_succeed_with_new_text(self): """Test succeed method with new text """ spinner = Halo(stream=self._stream) spinner.start('foo') spinner.succeed('bar') output = self._get_test_output()['text'] pattern = re.compile(r'(✔|v) bar', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) spinner.stop() def test_info(self): """Test info method """ spinner = Halo(stream=self._stream) spinner.start('foo') spinner.info() output = self._get_test_output()['text'] pattern = re.compile(r'(ℹ|¡) foo', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) spinner.stop() def test_fail(self): """Test fail method """ spinner = Halo(stream=self._stream) spinner.start('foo') spinner.fail() output = self._get_test_output()['text'] pattern = re.compile(r'(✖|×) foo', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) spinner.stop() def test_warning(self): """Test warn method """ spinner = Halo(stream=self._stream) spinner.start('foo') spinner.warn('Warning!') output = self._get_test_output()['text'] pattern = re.compile(r'(⚠|!!) Warning!', re.UNICODE) self.assertRegexpMatches(output[-1], pattern) spinner.stop() def test_spinner_getters_setters(self): """Test spinner getters and setters. """ spinner = Halo() self.assertEqual(spinner.text, '') self.assertIsNone(spinner.text_color, None) self.assertEqual(spinner.color, 'cyan') self.assertIsNone(spinner.spinner_id) spinner.spinner = 'dots12' spinner.text = 'bar' spinner.text_color = 'red' spinner.color = 'red' self.assertEqual(spinner.text, 'bar') self.assertEqual(spinner.text_color, 'red') self.assertEqual(spinner.color, 'red') if is_supported(): self.assertEqual(spinner.spinner, Spinners['dots12'].value) else: self.assertEqual(spinner.spinner, default_spinner) spinner.spinner = 'dots11' if is_supported(): self.assertEqual(spinner.spinner, Spinners['dots11'].value) else: self.assertEqual(spinner.spinner, default_spinner) spinner.spinner = 'foo_bar' self.assertEqual(spinner.spinner, default_spinner) # Color is None spinner.text_color = None spinner.color = None spinner.start() spinner.stop() self.assertIsNone(spinner.text_color) self.assertIsNone(spinner.color) def test_unavailable_spinner_defaults(self): """Test unavailable spinner defaults. """ spinner = Halo('dot') self.assertEqual(spinner.text, 'dot') self.assertEqual(spinner.spinner, default_spinner) def test_if_enabled(self): """Test if spinner is enabled """ spinner = Halo(text='foo', enabled=False, stream=self._stream) spinner.start() time.sleep(1) spinner.fail() output = self._get_test_output()['text'] self.assertEqual(len(output), 0) self.assertEqual(output, []) def test_writing_disabled_on_closed_stream(self): """Test no I/O is performed on closed streams """ # BytesIO supports the writable() method, while StringIO does not, in # some versions of Python. We want to check whether the stream is # writable (e.g. for file streams which can be open but not writable), # but only if the stream supports it — otherwise we assume # open == writable. for io_class in (io.StringIO, io.BytesIO): stream = io_class() stream.close() # sanity checks self.assertTrue(stream.closed) self.assertRaises(ValueError, stream.isatty) self.assertRaises(ValueError, stream.write, u'') try: spinner = Halo(text='foo', stream=stream) spinner.start() time.sleep(0.5) spinner.stop() except ValueError as e: self.fail('Attempted to write to a closed stream: {}'.format(e)) def test_closing_stream_before_stopping(self): """Test no I/O is performed on streams closed before stop is called """ stream = io.StringIO() spinner = Halo(text='foo', stream=stream) spinner.start() time.sleep(0.5) # no exception raised after closing the stream means test was successful try: stream.close() time.sleep(0.5) spinner.stop() except ValueError as e: self.fail('Attempted to write to a closed stream: {}'.format(e)) def test_closing_stream_before_persistent(self): """Test no I/O is performed on streams closed before stop_and_persist is called """ stream = io.StringIO() spinner = Halo(text='foo', stream=stream) spinner.start() time.sleep(0.5) # no exception raised after closing the stream means test was successful try: stream.close() time.sleep(0.5) spinner.stop_and_persist('done') except ValueError as e: self.fail('Attempted to write to a closed stream: {}'.format(e)) def test_setting_enabled_property(self): """Test if spinner stops writing when enabled property set to False """ spinner = Halo(text='foo', stream=self._stream) spinner.start() time.sleep(0.5) spinner.enabled = False bytes_written = self._stream.tell() time.sleep(0.5) spinner.stop() total_bytes_written = self._stream.tell() self.assertEqual(total_bytes_written, bytes_written) def test_spinner_interval_default(self): """Test proper assignment of the default interval value. """ spinner = Halo() self.assertEqual(spinner._interval, default_spinner['interval']) def test_spinner_interval_argument(self): """Test proper assignment of the interval value from the constructor argument. """ spinner = Halo(interval=123) self.assertEqual(spinner._interval, 123) def test_spinner_interval_dict(self): """Test proper assignment of the interval value from a dictionary. """ spinner = Halo(spinner={'interval': 321, 'frames': ['+', '-']}) self.assertEqual(spinner._interval, 321) def test_invalid_placement(self): """Test invalid placement of spinner. """ with self.assertRaises(ValueError): Halo(placement='') Halo(placement='foo') Halo(placement=None) spinner = Halo(placement='left') with self.assertRaises(ValueError): spinner.placement = '' spinner.placement = 'foo' spinner.placement = None def test_default_placement(self): """Test default placement of spinner. """ spinner = Halo() self.assertEqual(spinner.placement, 'left') def test_right_placement(self): """Test right placement of spinner. """ spinner = Halo(text='foo', placement='right', stream=self._stream) spinner.start() time.sleep(1) output = self._get_test_output()['text'] (text, _) = output[-1].split(' ') self.assertEqual(text, 'foo') spinner.succeed() output = self._get_test_output()['text'] (text, symbol) = output[-1].split(' ') pattern = re.compile(r"(✔|v)", re.UNICODE) self.assertEqual(text, 'foo') self.assertRegexpMatches(symbol, pattern) spinner.stop() def test_bounce_animation(self): def filler_text(n_chars): return "_" * n_chars terminal_width = get_terminal_columns() text = "{}abc".format(filler_text(terminal_width)) expected_frames_without_appended_spinner = [ "{}".format(filler_text(terminal_width - 2)), "{}".format(filler_text(terminal_width - 2)), "{}".format(filler_text(terminal_width - 2)), "{}a".format(filler_text(terminal_width - 3)), "{}ab".format(filler_text(terminal_width - 4)), "{}abc".format(filler_text(terminal_width - 5)), "{}abc".format(filler_text(terminal_width - 5)), "{}ab".format(filler_text(terminal_width - 4)), "{}a".format(filler_text(terminal_width - 3)), "{}".format(filler_text(terminal_width - 2)), "{}".format(filler_text(terminal_width - 2)), "{}".format(filler_text(terminal_width - 2)), ] # Prepend the actual spinner expected_frames = [ "{} {}".format(frames[idx % frames.__len__()], frame) for idx, frame in enumerate(expected_frames_without_appended_spinner) ] spinner = Halo(text, animation="bounce", stream=self._stream) spinner.start() # Sleep a full bounce cycle time.sleep(1.2) spinner.stop() output = self._get_test_output()['text'] zipped_expected_and_actual_frame = zip(expected_frames, output) for multiple_frames in zipped_expected_and_actual_frame: expected_frame, actual_frame = multiple_frames self.assertEquals(expected_frame, actual_frame) def test_animation_setter(self): spinner = Halo("Asdf") spinner.animation = "bounce" self.assertEquals("bounce", spinner.animation) spinner.animation = "marquee" self.assertEquals("marquee", spinner.animation) def test_spinner_color(self): """Test ANSI escape characters are present """ for color, color_int in COLORS.items(): self._stream = io.open(self._stream_file, 'w+') # reset stream spinner = Halo(color=color, stream=self._stream) spinner.start() spinner.stop() output = self._get_test_output(no_ansi=False) output_merged = [arr for c in output['colors'] for arr in c] self.assertEquals(str(color_int) in output_merged, True) def test_redirect_stdout(self): """Test redirect stdout """ out = self._stream try: self._stream = self._stream_no_tty spinner = Halo(text='foo', spinner='dots', stream=self._stream) spinner.start() time.sleep(1) spinner.stop() output = self._get_test_output()['text'] finally: self._stream = out self.assertIn('foo', output[0]) def tearDown(self): pass if __name__ == '__main__': SUITE = unittest.TestLoader().loadTestsFromTestCase(TestHalo) unittest.TextTestRunner(verbosity=2).run(SUITE)