mirror of
https://github.com/not-kennethreitz/blindspin.git
synced 2026-06-05 23:10:18 +00:00
Merge pull request #9 from andy-maier/andy/issue8-suppress-when-redirected
Fixes #8, #10: Added support for suppressing spinner when not on a TTY.
This commit is contained in:
@@ -8,6 +8,7 @@ notifications:
|
||||
- yoav@yoavram.com
|
||||
install:
|
||||
- pip install click
|
||||
- pip install six
|
||||
- python setup.py install
|
||||
script: py.test
|
||||
deploy:
|
||||
|
||||
@@ -7,18 +7,21 @@ import itertools
|
||||
class Spinner(object):
|
||||
spinner_cycle = itertools.cycle(['-', '/', '|', '\\'])
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self, force=False):
|
||||
self._force = force
|
||||
self.stop_running = None
|
||||
self.spin_thread = None
|
||||
|
||||
def start(self):
|
||||
self.stop_running = threading.Event()
|
||||
self.spin_thread = threading.Thread(target=self.init_spin)
|
||||
self.spin_thread.start()
|
||||
if sys.stdout.isatty() or self._force:
|
||||
self.stop_running = threading.Event()
|
||||
self.spin_thread = threading.Thread(target=self.init_spin)
|
||||
self.spin_thread.start()
|
||||
|
||||
def stop(self):
|
||||
self.stop_running.set()
|
||||
self.spin_thread.join()
|
||||
if self.spin_thread:
|
||||
self.stop_running.set()
|
||||
self.spin_thread.join()
|
||||
|
||||
def init_spin(self):
|
||||
while not self.stop_running.is_set():
|
||||
@@ -34,9 +37,16 @@ class Spinner(object):
|
||||
self.stop()
|
||||
|
||||
|
||||
def spinner():
|
||||
def spinner(force=False):
|
||||
"""This function creates a context manager that is used to display a
|
||||
spinner as long as the context has not exited.
|
||||
spinner on stdout as long as the context has not exited.
|
||||
|
||||
The spinner is created only if stdout is not redirected, or if the spinner
|
||||
is forced using the `force` parameter.
|
||||
|
||||
Parameters:
|
||||
|
||||
force (bool): Force creation of spinner even when stdout is redirected.
|
||||
|
||||
Example usage::
|
||||
|
||||
@@ -45,7 +55,7 @@ def spinner():
|
||||
do_something_else()
|
||||
|
||||
"""
|
||||
return Spinner()
|
||||
return Spinner(force)
|
||||
|
||||
|
||||
from ._version import get_versions
|
||||
|
||||
@@ -18,8 +18,9 @@ setup(
|
||||
description='Spinner for Click',
|
||||
extras_require={
|
||||
'test': [
|
||||
'click'
|
||||
'click',
|
||||
'pytest',
|
||||
'six',
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import tempfile
|
||||
from six import StringIO
|
||||
import click
|
||||
from click.testing import CliRunner
|
||||
|
||||
@@ -33,3 +38,43 @@ def test_spinner_resume():
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exception is None
|
||||
|
||||
|
||||
def test_spinner_redirect():
|
||||
@click.command()
|
||||
def cli():
|
||||
stdout_io = StringIO()
|
||||
saved_stdout = sys.stdout
|
||||
sys.stdout = stdout_io # redirect stdout to a string buffer
|
||||
spinner = click_spinner.Spinner()
|
||||
spinner.start()
|
||||
time.sleep(1) # allow time for a few spins
|
||||
spinner.stop()
|
||||
sys.stdout = saved_stdout
|
||||
stdout_io.flush()
|
||||
stdout_str = stdout_io.getvalue()
|
||||
assert len(stdout_str) == 0
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exception is None
|
||||
|
||||
|
||||
def test_spinner_redirect_force():
|
||||
@click.command()
|
||||
def cli():
|
||||
stdout_io = StringIO()
|
||||
saved_stdout = sys.stdout
|
||||
sys.stdout = stdout_io # redirect stdout to a string buffer
|
||||
spinner = click_spinner.Spinner(force=True)
|
||||
spinner.start()
|
||||
time.sleep(1) # allow time for a few spins
|
||||
spinner.stop()
|
||||
sys.stdout = saved_stdout
|
||||
stdout_io.flush()
|
||||
stdout_str = stdout_io.getvalue()
|
||||
assert len(stdout_str) > 0
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exception is None
|
||||
|
||||
|
||||
Reference in New Issue
Block a user