# -*- coding: utf-8 -*-
# Advice: use repr(our_file.read()) to print the full output of tqdm
# (else '\r' will replace the previous lines and you'll see only the latest.
import csv
import os
import re
import sys
from contextlib import contextmanager
from functools import wraps
from warnings import catch_warnings, simplefilter
from pytest import importorskip, mark, raises, skip
from tqdm import TqdmDeprecationWarning, TqdmWarning, tqdm, trange
from tqdm.contrib import DummyTqdmFile
from tqdm.std import EMA, Bar
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
from io import IOBase # to support unicode strings
from io import BytesIO
class DeprecationError(Exception):
pass
# Ensure we can use `with closing(...) as ... :` syntax
if getattr(StringIO, '__exit__', False) and getattr(StringIO, '__enter__', False):
def closing(arg):
return arg
else:
from contextlib import closing
nt_and_no_colorama = False
if os.name == 'nt':
try:
import colorama # NOQA
except ImportError:
nt_and_no_colorama = True
# Regex definitions
# List of control characters
CTRLCHR = [r'\r', r'\n', r'\x1b\[A'] # Need to escape [ for regex
# Regular expressions compilation
RE_rate = re.compile(r'[^\d](\d[.\d]+)it/s')
RE_ctrlchr = re.compile("(%s)" % '|'.join(CTRLCHR)) # Match control chars
RE_ctrlchr_excl = re.compile('|'.join(CTRLCHR)) # Match and exclude ctrl chars
RE_pos = re.compile(r'([\r\n]+((pos\d+) bar:\s+\d+%|\s{3,6})?[^\r\n]*)')
def pos_line_diff(res_list, expected_list, raise_nonempty=True):
"""
Return differences between two bar output lists.
To be used with `RE_pos`
"""
res = [(r, e) for r, e in zip(res_list, expected_list)
for pos in [len(e) - len(e.lstrip('\n'))] # bar position
if r != e # simple comparison
if not r.startswith(e) # start matches
or not (
# move up at end (maybe less due to closing bars)
any(r.endswith(end + i * '\x1b[A') for i in range(pos + 1)
for end in [
']', # bar
' ']) # cleared
or '100%' in r # completed bar
or r == '\n') # final bar
or r[(-1 - pos) * len('\x1b[A'):] == '\x1b[A'] # too many moves up
if raise_nonempty and (res or len(res_list) != len(expected_list)):
if len(res_list) < len(expected_list):
res.extend([(None, e) for e in expected_list[len(res_list):]])
elif len(res_list) > len(expected_list):
res.extend([(r, None) for r in res_list[len(expected_list):]])
raise AssertionError(
"Got => Expected\n" + '\n'.join('%r => %r' % i for i in res))
return res
class DiscreteTimer(object):
"""Virtual discrete time manager, to precisely control time for tests"""
def __init__(self):
self.t = 0.0
def sleep(self, t):
"""Sleep = increment the time counter (almost no CPU used)"""
self.t += t
def time(self):
"""Get the current time"""
return self.t
def cpu_timify(t, timer=None):
"""Force tqdm to use the specified timer instead of system-wide time()"""
if timer is None:
timer = DiscreteTimer()
t._time = timer.time
t._sleep = timer.sleep
t.start_t = t.last_print_t = t._time()
return timer
class UnicodeIO(IOBase):
"""Unicode version of StringIO"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.encoding = 'U8' # io.StringIO supports unicode, but no encoding
self.text = ''
self.cursor = 0
def __len__(self):
return len(self.text)
def seek(self, offset):
self.cursor = offset
def tell(self):
return self.cursor
def write(self, s):
self.text = self.text[:self.cursor] + s + self.text[self.cursor + len(s):]
self.cursor += len(s)
def read(self, n=-1):
_cur = self.cursor
self.cursor = len(self) if n < 0 else min(_cur + n, len(self))
return self.text[_cur:self.cursor]
def getvalue(self):
return self.text
def get_bar(all_bars, i=None):
"""Get a specific update from a whole bar traceback"""
# Split according to any used control characters
bars_split = RE_ctrlchr_excl.split(all_bars)
bars_split = list(filter(None, bars_split)) # filter out empty splits
return bars_split if i is None else bars_split[i]
def progressbar_rate(bar_str):
return float(RE_rate.search(bar_str).group(1))
def squash_ctrlchars(s):
"""Apply control characters in a string just like a terminal display"""
curline = 0
lines = [''] # state of fake terminal
for nextctrl in filter(None, RE_ctrlchr.split(s)):
# apply control chars
if nextctrl == '\r':
# go to line beginning (simplified here: just empty the string)
lines[curline] = ''
elif nextctrl == '\n':
if curline >= len(lines) - 1:
# wrap-around creates newline
lines.append('')
# move cursor down
curline += 1
elif nextctrl == '\x1b[A':
# move cursor up
if curline > 0:
curline -= 1
else:
raise ValueError("Cannot go further up")
else:
# print message on current line
lines[curline] += nextctrl
return lines
def test_format_interval():
"""Test time interval format"""
format_interval = tqdm.format_interval
assert format_interval(60) == '01:00'
assert format_interval(6160) == '1:42:40'
assert format_interval(238113) == '66:08:33'
def test_format_num():
"""Test number format"""
format_num = tqdm.format_num
assert float(format_num(1337)) == 1337
assert format_num(int(1e6)) == '1e+6'
assert format_num(1239876) == '1' '239' '876'
assert format_num(0.00001234) == '1.23e-5'
assert format_num(-0.1234) == '-0.123'
def test_format_meter():
"""Test statistics and progress bar formatting"""
try:
unich = unichr
except NameError:
unich = chr
format_meter = tqdm.format_meter
assert format_meter(0, 1000, 13) == " 0%| | 0/1000 [00:13<?, ?it/s]"
# If not implementing any changes to _tqdm.py, set prefix='desc'
# or else ": : " will be in output, so assertion should change
assert format_meter(0, 1000, 13, ncols=68, prefix='desc: ') == (
"desc: 0%| | 0/1000 [00:13<?, ?it/s]")
assert format_meter(231, 1000, 392) == (" 23%|" + unich(0x2588) * 2 + unich(0x258e) +
" | 231/1000 [06:32<21:44, 1.70s/it]")
assert format_meter(10000, 1000, 13) == "10000it [00:13, 769.23it/s]"
assert format_meter(231, 1000, 392, ncols=56, ascii=True) == " 23%|" + '#' * 3 + '6' + (
" | 231/1000 [06:32<21:44, 1.70s/it]")
assert format_meter(100000, 1000, 13, unit_scale=True,
unit='iB') == "100kiB [00:13, 7.69kiB/s]"
assert format_meter(100, 1000, 12, ncols=0,
rate=7.33) == " 10% 100/1000 [00:12<02:02, 7.33it/s]"
# ncols is small, l_bar is too large
# l_bar gets chopped
# no bar
# no r_bar
# 10/12 stars since ncols is 10
assert format_meter(
0, 1000, 13, ncols=10,
bar_format="************{bar:10}$$$$$$$$$$") == "**********"
# n_cols allows for l_bar and some of bar
# l_bar displays
# bar gets chopped
# no r_bar
# all 12 stars and 8/10 bar parts
assert format_meter(
0, 1000, 13, ncols=20,
bar_format="************{bar:10}$$$$$$$$$$") == "************ "
# n_cols allows for l_bar, bar, and some of r_bar
# l_bar displays
# bar displays
# r_bar gets chopped
# all 12 stars and 10 bar parts, but only 8/10 dollar signs
assert format_meter(
0, 1000, 13, ncols=30,
bar_format="************{bar:10}$$$$$$$$$$") == "************ $$$$$$$$"
# trim left ANSI; escape is before trim zone
# we only know it has ANSI codes, so we append an END code anyway
assert format_meter(
0, 1000, 13, ncols=10, bar_format="*****\033[22m****\033[0m***{bar:10}$$$$$$$$$$"
) == "*****\033[22m****\033[0m*\033[0m"
# trim left ANSI; escape is at trim zone
assert format_meter(
0, 1000, 13, ncols=10,
bar_format="*****\033[22m*****\033[0m**{bar:10}$$$$$$$$$$") == "*****\033[22m*****\033[0m"
# trim left ANSI; escape is after trim zone
assert format_meter(
0, 1000, 13, ncols=10,
bar_format="*****\033[22m******\033[0m*{bar:10}$$$$$$$$$$") == "*****\033[22m*****\033[0m"
# Check that bar_format correctly adapts {bar} size to the rest
assert format_meter(
20, 100, 12, ncols=13, rate=8.1,
bar_format=r'{l_bar}{bar}|{n_fmt}/{total_fmt}') == " 20%|" + unich(0x258f) + "|20/100"
assert format_meter(
20, 100, 12, ncols=14, rate=8.1,
bar_format=r'{l_bar}{bar}|{n_fmt}/{total_fmt}') == " 20%|" + unich(0x258d) + " |20/100"
# Check wide characters
assert format_meter(0, 1000, 13, ncols=68, prefix='ｆｕｌｌｗｉｄｔｈ: ') == (
"ｆｕｌｌｗｉｄｔｈ: 0%| | 0/1000 [00:13<?, ?it/s]")
assert format_meter(0, 1000, 13, ncols=68, prefix='ニッポン [ﾆｯﾎﾟﾝ]: ') == (
"ニッポン [ﾆｯﾎﾟﾝ]: 0%| | 0/1000 [00:13<?, ?it/s]")
# Check that bar_format can print only {bar} or just one side
assert format_meter(20, 100, 12, ncols=2, rate=8.1,
bar_format=r'{bar}') == unich(0x258d) + " "
assert format_meter(20, 100, 12, ncols=7, rate=8.1,
bar_format=r'{l_bar}{bar}') == " 20%|" + unich(0x258d) + " "
assert format_meter(20, 100, 12, ncols=6, rate=8.1,
bar_format=r'{bar}|test') == unich(0x258f) + "|test"
def test_ansi_escape_codes():
"""Test stripping of ANSI escape codes"""
ansi = {'BOLD': '\033[1m', 'RED': '\033[91m', 'END': '\033[0m'}
desc_raw = '{BOLD}{RED}Colored{END} description'
ncols = 123
desc_stripped = desc_raw.format(BOLD='', RED='', END='')
meter = tqdm.format_meter(0, 100, 0, ncols=ncols, prefix=desc_stripped)
assert len(meter) == ncols
desc = desc_raw.format(**ansi)
meter = tqdm.format_meter(0, 100, 0, ncols=ncols, prefix=desc)
# `format_meter` inserts an extra END for safety
ansi_len = len(desc) - len(desc_stripped) + len(ansi['END'])
assert len(meter) == ncols + ansi_len
def test_si_format():
"""Test SI unit prefixes"""
format_meter = tqdm.format_meter
assert '9.00 ' in format_meter(1, 9, 1, unit_scale=True, unit='B')
assert '99.0 ' in format_meter(1, 99, 1, unit_scale=True)
assert '999 ' in format_meter(1, 999, 1, unit_scale=True)
assert '9.99k ' in format_meter(1, 9994, 1, unit_scale=True)
assert '10.0k ' in format_meter(1, 9999, 1, unit_scale=True)
assert '99.5k ' in format_meter(1, 99499, 1, unit_scale=True)
assert '100k ' in format_meter(1, 99999, 1, unit_scale=True)
assert '1.00M ' in format_meter(1, 999999, 1, unit_scale=True)
assert '1.00G ' in format_meter(1, 999999999, 1, unit_scale=True)
assert '1.00T ' in format_meter(1, 999999999999, 1, unit_scale=True)
assert '1.00P ' in format_meter(1, 999999999999999, 1, unit_scale=True)
assert '1.00E ' in format_meter(1, 999999999999999999, 1, unit_scale=True)
assert '1.00Z ' in format_meter(1, 999999999999999999999, 1, unit_scale=True)
assert '1.0Y ' in format_meter(1, 999999999999999999999999, 1, unit_scale=True)
assert '10.0Y ' in format_meter(1, 9999999999999999999999999, 1, unit_scale=True)
assert '100.0Y ' in format_meter(1, 99999999999999999999999999, 1, unit_scale=True)
assert '1000.0Y ' in format_meter(1, 999999999999999999999999999, 1,
unit_scale=True)
def test_bar_formatspec():
"""Test Bar.__format__ spec"""
assert f"{Bar(0.3):5a}" == "#5 "
assert f"{Bar(0.5, charset=' .oO0'):2}" == "0 "
assert f"{Bar(0.5, charset=' .oO0'):2a}" == "# "
assert f"{Bar(0.5, 10):-6a}" == '## '
assert f"{Bar(0.5, 10):2b}" == ' '
def test_all_defaults():
"""Test default kwargs"""
with closing(UnicodeIO()) as our_file:
with tqdm(range(10), file=our_file) as progressbar:
assert len(progressbar) == 10
for _ in progressbar:
pass
# restore stdout/stderr output for `nosetest` interface
# try:
# sys.stderr.write('\x1b[A')
# except:
# pass
sys.stderr.write('\rTest default kwargs ... ')
class WriteTypeChecker(BytesIO):
"""File-like to assert the expected type is written"""
def __init__(self, expected_type):
super().__init__()
self.expected_type = expected_type
def write(self, s):
assert isinstance(s, self.expected_type)
def test_native_string_io_for_default_file():
"""Native strings written to unspecified files"""
stderr = sys.stderr
try:
sys.stderr = WriteTypeChecker(expected_type=type(''))
for _ in tqdm(range(3)):
pass
sys.stderr.encoding = None # py2 behaviour
for _ in tqdm(range(3)):
pass
finally:
sys.stderr = stderr
def test_unicode_string_io_for_specified_file():
"""Unicode strings written to specified files"""
for _ in tqdm(range(3), file=WriteTypeChecker(expected_type=type(u''))):
pass
def test_write_bytes():
"""Test write_bytes argument with and without `file`"""
# specified file (and bytes)
for _ in tqdm(range(3), file=WriteTypeChecker(expected_type=type(b'')),
write_bytes=True):
pass
# unspecified file (and unicode)
stderr = sys.stderr
try:
sys.stderr = WriteTypeChecker(expected_type=type(u''))
for _ in tqdm(range(3), write_bytes=False):
pass
finally:
sys.stderr = stderr
def test_iterate_over_csv_rows():
"""Test csv iterator"""
# Create a test csv pseudo file
with closing(StringIO()) as test_csv_file:
writer = csv.writer(test_csv_file)
for _ in range(3):
writer.writerow(['test'] * 3)
test_csv_file.seek(0)
# Test that nothing fails if we iterate over rows
reader = csv.DictReader(test_csv_file, fieldnames=('row1', 'row2', 'row3'))
with closing(StringIO()) as our_file:
for _ in tqdm(reader, file=our_file):
pass
def test_file_output():
"""Test output to arbitrary file-like objects"""
with closing(StringIO()) as our_file:
for i in tqdm(range(3), file=our_file):
if i == 1:
our_file.seek(0)
assert '0/3' in our_file.read()
def test_leave_option():
"""Test `leave=True` always prints info about the last iteration"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, leave=True):
pass
res = our_file.getvalue()
assert '| 3/3 ' in res
assert '\n' == res[-1] # not '\r'
with closing(StringIO()) as our_file2:
for _ in tqdm(range(3), file=our_file2, leave=False):
pass
assert '| 3/3 ' not in our_file2.getvalue()
def test_trange():
"""Test trange"""
with closing(StringIO()) as our_file:
for _ in trange(3, file=our_file, leave=True):
pass
assert '| 3/3 ' in our_file.getvalue()
with closing(StringIO()) as our_file2:
for _ in trange(3, file=our_file2, leave=False):
pass
assert '| 3/3 ' not in our_file2.getvalue()
def test_min_interval():
"""Test mininterval"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, mininterval=1e-10):
pass
assert " 0%| | 0/3 [00:00<" in our_file.getvalue()
def test_max_interval():
"""Test maxinterval"""
total = 100
bigstep = 10
smallstep = 5
# Test without maxinterval
timer = DiscreteTimer()
with closing(StringIO()) as our_file:
with closing(StringIO()) as our_file2:
# with maxinterval but higher than loop sleep time
t = tqdm(total=total, file=our_file, miniters=None, mininterval=0,
smoothing=1, maxinterval=1e-2)
cpu_timify(t, timer)
# without maxinterval
t2 = tqdm(total=total, file=our_file2, miniters=None, mininterval=0,
smoothing=1, maxinterval=None)
cpu_timify(t2, timer)
assert t.dynamic_miniters
assert t2.dynamic_miniters
# Increase 10 iterations at once
t.update(bigstep)
t2.update(bigstep)
# The next iterations should not trigger maxinterval (step 10)
for _ in range(4):
t.update(smallstep)
t2.update(smallstep)
timer.sleep(1e-5)
t.close() # because PyPy doesn't gc immediately
t2.close() # as above
assert "25%" not in our_file2.getvalue()
assert "25%" not in our_file.getvalue()
# Test with maxinterval effect
timer = DiscreteTimer()
with closing(StringIO()) as our_file:
with tqdm(total=total, file=our_file, miniters=None, mininterval=0,
smoothing=1, maxinterval=1e-4) as t:
cpu_timify(t, timer)
# Increase 10 iterations at once
t.update(bigstep)
# The next iterations should trigger maxinterval (step 5)
for _ in range(4):
t.update(smallstep)
timer.sleep(1e-2)
assert "25%" in our_file.getvalue()
# Test iteration based tqdm with maxinterval effect
timer = DiscreteTimer()
with closing(StringIO()) as our_file:
with tqdm(range(total), file=our_file, miniters=None,
mininterval=1e-5, smoothing=1, maxinterval=1e-4) as t2:
cpu_timify(t2, timer)
for i in t2:
if i >= (bigstep - 1) and ((i - (bigstep - 1)) % smallstep) == 0:
timer.sleep(1e-2)
if i >= 3 * bigstep:
break
assert "15%" in our_file.getvalue()
# Test different behavior with and without mininterval
timer = DiscreteTimer()
total = 1000
mininterval = 0.1
maxinterval = 10
with closing(StringIO()) as our_file:
with tqdm(total=total, file=our_file, miniters=None, smoothing=1,
mininterval=mininterval, maxinterval=maxinterval) as tm1:
with tqdm(total=total, file=our_file, miniters=None, smoothing=1,
mininterval=0, maxinterval=maxinterval) as tm2:
cpu_timify(tm1, timer)
cpu_timify(tm2, timer)
# Fast iterations, check if dynamic_miniters triggers
timer.sleep(mininterval) # to force update for t1
tm1.update(total / 2)
tm2.update(total / 2)
assert int(tm1.miniters) == tm2.miniters == total / 2
# Slow iterations, check different miniters if mininterval
timer.sleep(maxinterval * 2)
tm1.update(total / 2)
tm2.update(total / 2)
res = [tm1.miniters, tm2.miniters]
assert res == [(total / 2) * mininterval / (maxinterval * 2),
(total / 2) * maxinterval / (maxinterval * 2)]
# Same with iterable based tqdm
timer1 = DiscreteTimer() # need 2 timers for each bar because zip not work
timer2 = DiscreteTimer()
total = 100
mininterval = 0.1
maxinterval = 10
with closing(StringIO()) as our_file:
t1 = tqdm(range(total), file=our_file, miniters=None, smoothing=1,
mininterval=mininterval, maxinterval=maxinterval)
t2 = tqdm(range(total), file=our_file, miniters=None, smoothing=1,
mininterval=0, maxinterval=maxinterval)
cpu_timify(t1, timer1)
cpu_timify(t2, timer2)
for i in t1:
if i == ((total / 2) - 2):
timer1.sleep(mininterval)
if i == (total - 1):
timer1.sleep(maxinterval * 2)
for i in t2:
if i == ((total / 2) - 2):
timer2.sleep(mininterval)
if i == (total - 1):
timer2.sleep(maxinterval * 2)
assert t1.miniters == 0.255
assert t2.miniters == 0.5
t1.close()
t2.close()
def test_delay():
"""Test delay"""
timer = DiscreteTimer()
with closing(StringIO()) as our_file:
t = tqdm(total=2, file=our_file, leave=True, delay=3)
cpu_timify(t, timer)
timer.sleep(2)
t.update(1)
assert not our_file.getvalue()
timer.sleep(2)
t.update(1)
assert our_file.getvalue()
t.close()
def test_min_iters():
"""Test miniters"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, leave=True, mininterval=0, miniters=2):
pass
out = our_file.getvalue()
assert '| 0/3 ' in out
assert '| 1/3 ' not in out
assert '| 2/3 ' in out
assert '| 3/3 ' in out
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, leave=True, mininterval=0, miniters=1):
pass
out = our_file.getvalue()
assert '| 0/3 ' in out
assert '| 1/3 ' in out
assert '| 2/3 ' in out
assert '| 3/3 ' in out
def test_dynamic_min_iters():
"""Test purely dynamic miniters (and manual updates and __del__)"""
with closing(StringIO()) as our_file:
total = 10
t = tqdm(total=total, file=our_file, miniters=None, mininterval=0, smoothing=1)
t.update()
# Increase 3 iterations
t.update(3)
# The next two iterations should be skipped because of dynamic_miniters
t.update()
t.update()
# The third iteration should be displayed
t.update()
out = our_file.getvalue()
assert t.dynamic_miniters
t.__del__() # simulate immediate del gc
assert ' 0%| | 0/10 [00:00<' in out
assert '40%' in out
assert '50%' not in out
assert '60%' not in out
assert '70%' in out
# Check with smoothing=0, miniters should be set to max update seen so far
with closing(StringIO()) as our_file:
total = 10
t = tqdm(total=total, file=our_file, miniters=None, mininterval=0, smoothing=0)
t.update()
t.update(2)
t.update(5) # this should be stored as miniters
t.update(1)
out = our_file.getvalue()
assert all(i in out for i in ("0/10", "1/10", "3/10"))
assert "2/10" not in out
assert t.dynamic_miniters and not t.smoothing
assert t.miniters == 5
t.close()
# Check iterable based tqdm
with closing(StringIO()) as our_file:
t = tqdm(range(10), file=our_file, miniters=None, mininterval=None,
smoothing=0.5)
for _ in t:
pass
assert t.dynamic_miniters
# No smoothing
with closing(StringIO()) as our_file:
t = tqdm(range(10), file=our_file, miniters=None, mininterval=None,
smoothing=0)
for _ in t:
pass
assert t.dynamic_miniters
# No dynamic_miniters (miniters is fixed manually)
with closing(StringIO()) as our_file:
t = tqdm(range(10), file=our_file, miniters=1, mininterval=None)
for _ in t:
pass
assert not t.dynamic_miniters
def test_big_min_interval():
"""Test large mininterval"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(2), file=our_file, mininterval=1E10):
pass
assert '50%' not in our_file.getvalue()
with closing(StringIO()) as our_file:
with tqdm(range(2), file=our_file, mininterval=1E10) as t:
t.update()
t.update()
assert '50%' not in our_file.getvalue()
def test_smoothed_dynamic_min_iters():
"""Test smoothed dynamic miniters"""
timer = DiscreteTimer()
with closing(StringIO()) as our_file:
with tqdm(total=100, file=our_file, miniters=None, mininterval=1,
smoothing=0.5, maxinterval=0) as t:
cpu_timify(t, timer)
# Increase 10 iterations at once
timer.sleep(1)
t.update(10)
# The next iterations should be partially skipped
for _ in range(2):
timer.sleep(1)
t.update(4)
for _ in range(20):
timer.sleep(1)
t.update()
assert t.dynamic_miniters
out = our_file.getvalue()
assert ' 0%| | 0/100 [00:00<' in out
assert '20%' in out
assert '23%' not in out
assert '25%' in out
assert '26%' not in out
assert '28%' in out
def test_smoothed_dynamic_min_iters_with_min_interval():
"""Test smoothed dynamic miniters with mininterval"""
timer = DiscreteTimer()
# In this test, `miniters` should gradually decline
total = 100
with closing(StringIO()) as our_file:
# Test manual updating tqdm
with tqdm(total=total, file=our_file, miniters=None, mininterval=1e-3,
smoothing=1, maxinterval=0) as t:
cpu_timify(t, timer)
t.update(10)
timer.sleep(1e-2)
for _ in range(4):
t.update()
timer.sleep(1e-2)
out = our_file.getvalue()
assert t.dynamic_miniters
with closing(StringIO()) as our_file:
# Test iteration-based tqdm
with tqdm(range(total), file=our_file, miniters=None,
mininterval=0.01, smoothing=1, maxinterval=0) as t2:
cpu_timify(t2, timer)
for i in t2:
if i >= 10:
timer.sleep(0.1)
if i >= 14:
break
out2 = our_file.getvalue()
assert t.dynamic_miniters
assert ' 0%| | 0/100 [00:00<' in out
assert '11%' in out and '11%' in out2
# assert '12%' not in out and '12%' in out2
assert '13%' in out and '13%' in out2
assert '14%' in out and '14%' in out2
@mark.slow
def test_rlock_creation():
"""Test that importing tqdm does not create multiprocessing objects."""
mp = importorskip('multiprocessing')
if not hasattr(mp, 'get_context'):
skip("missing multiprocessing.get_context")
# Use 'spawn' instead of 'fork' so that the process does not inherit any
# globals that have been constructed by running other tests
ctx = mp.get_context('spawn')
with ctx.Pool(1) as pool:
# The pool will propagate the error if the target method fails
pool.apply(_rlock_creation_target)
def _rlock_creation_target():
"""Check that the RLock has not been constructed."""
import multiprocessing as mp
patch = importorskip('unittest.mock').patch
# Patch the RLock class/method but use the original implementation
with patch('multiprocessing.RLock', wraps=mp.RLock) as rlock_mock:
# Importing the module should not create a lock
from tqdm import tqdm
assert rlock_mock.call_count == 0
# Creating a progress bar should initialize the lock
with closing(StringIO()) as our_file:
with tqdm(file=our_file) as _: # NOQA
pass
assert rlock_mock.call_count == 1
# Creating a progress bar again should reuse the lock
with closing(StringIO()) as our_file:
with tqdm(file=our_file) as _: # NOQA
pass
assert rlock_mock.call_count == 1
def test_disable():
"""Test disable"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, disable=True):
pass
assert our_file.getvalue() == ''
with closing(StringIO()) as our_file:
progressbar = tqdm(total=3, file=our_file, miniters=1, disable=True)
progressbar.update(3)
progressbar.close()
assert our_file.getvalue() == ''
def test_infinite_total():
"""Test treatment of infinite total"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, total=float("inf")):
pass
def test_nototal():
"""Test unknown total length"""
with closing(StringIO()) as our_file:
for _ in tqdm(iter(range(10)), file=our_file, unit_scale=10):
pass
assert "100it" in our_file.getvalue()
with closing(StringIO()) as our_file:
for _ in tqdm(iter(range(10)), file=our_file,
bar_format="{l_bar}{bar}{r_bar}"):
pass
assert "10/?" in our_file.getvalue()
def test_unit():
"""Test SI unit prefix"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), file=our_file, miniters=1, unit="bytes"):
pass
assert 'bytes/s' in our_file.getvalue()
def test_ascii():
"""Test ascii/unicode bar"""
# Test ascii autodetection
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file, ascii=None) as t:
assert t.ascii # TODO: this may fail in the future
# Test ascii bar
with closing(StringIO()) as our_file:
for _ in tqdm(range(3), total=15, file=our_file, miniters=1,
mininterval=0, ascii=True):
pass
res = our_file.getvalue().strip("\r").split("\r")
assert '7%|6' in res[1]
assert '13%|#3' in res[2]
assert '20%|##' in res[3]
# Test unicode bar
with closing(UnicodeIO()) as our_file:
with tqdm(total=15, file=our_file, ascii=False, mininterval=0) as t:
for _ in range(3):
t.update()
res = our_file.getvalue().strip("\r").split("\r")
assert u"7%|\u258b" in res[1]
assert u"13%|\u2588\u258e" in res[2]
assert u"20%|\u2588\u2588" in res[3]
# Test custom bar
for bars in [" .oO0", " #"]:
with closing(StringIO()) as our_file:
for _ in tqdm(range(len(bars) - 1), file=our_file, miniters=1,
mininterval=0, ascii=bars, ncols=27):
pass
res = our_file.getvalue().strip("\r").split("\r")
for b, line in zip(bars, res):
assert '|' + b + '|' in line
def test_update():
"""Test manual creation and updates"""
res = None
with closing(StringIO()) as our_file:
with tqdm(total=2, file=our_file, miniters=1, mininterval=0) as progressbar:
assert len(progressbar) == 2
progressbar.update(2)
assert '| 2/2' in our_file.getvalue()
progressbar.desc = 'dynamically notify of 4 increments in total'
progressbar.total = 4
progressbar.update(-1)
progressbar.update(2)
res = our_file.getvalue()
assert '| 3/4 ' in res
assert 'dynamically notify of 4 increments in total' in res
def test_close():
"""Test manual creation and closure and n_instances"""
# With `leave` option
with closing(StringIO()) as our_file:
progressbar = tqdm(total=3, file=our_file, miniters=10)
progressbar.update(3)
assert '| 3/3 ' not in our_file.getvalue() # Should be blank
assert len(tqdm._instances) == 1
progressbar.close()
assert len(tqdm._instances) == 0
assert '| 3/3 ' in our_file.getvalue()
# Without `leave` option
with closing(StringIO()) as our_file:
progressbar = tqdm(total=3, file=our_file, miniters=10, leave=False)
progressbar.update(3)
progressbar.close()
assert '| 3/3 ' not in our_file.getvalue() # Should be blank
# With all updates
with closing(StringIO()) as our_file:
assert len(tqdm._instances) == 0
with tqdm(total=3, file=our_file, miniters=0, mininterval=0,
leave=True) as progressbar:
assert len(tqdm._instances) == 1
progressbar.update(3)
res = our_file.getvalue()
assert '| 3/3 ' in res # Should be blank
assert '\n' not in res
# close() called
assert len(tqdm._instances) == 0
exres = res.rsplit(', ', 1)[0]
res = our_file.getvalue()
assert res[-1] == '\n'
if not res.startswith(exres):
raise AssertionError(f"\n<<< Expected:\n{exres}, ...it/s]\n>>> Got:\n{res}\n===")
# Closing after the output stream has closed
with closing(StringIO()) as our_file:
t = tqdm(total=2, file=our_file)
t.update()
t.update()
t.close()
def test_ema():
"""Test exponential weighted average"""
ema = EMA(0.01)
assert round(ema(10), 2) == 10
assert round(ema(1), 2) == 5.48
assert round(ema(), 2) == 5.48
assert round(ema(1), 2) == 3.97
assert round(ema(1), 2) == 3.22
def test_smoothing():
"""Test exponential weighted average smoothing"""
timer = DiscreteTimer()
# -- Test disabling smoothing
with closing(StringIO()) as our_file:
with tqdm(range(3), file=our_file, smoothing=None, leave=True) as t:
cpu_timify(t, timer)
for _ in t:
pass
assert '| 3/3 ' in our_file.getvalue()
# -- Test smoothing
# 1st case: no smoothing (only use average)
with closing(StringIO()) as our_file2:
with closing(StringIO()) as our_file:
t = tqdm(range(3), file=our_file2, smoothing=None, leave=True,
miniters=1, mininterval=0)
cpu_timify(t, timer)
with tqdm(range(3), file=our_file, smoothing=None, leave=True,
miniters=1, mininterval=0) as t2:
cpu_timify(t2, timer)
for i in t2:
# Sleep more for first iteration and
# see how quickly rate is updated
if i == 0:
timer.sleep(0.01)
else:
# Need to sleep in all iterations
# to calculate smoothed rate
# (else delta_t is 0!)
timer.sleep(0.001)
t.update()
n_old = len(tqdm._instances)
t.close()
assert len(tqdm._instances) == n_old - 1
# Get result for iter-based bar
a = progressbar_rate(get_bar(our_file.getvalue(), 3))
# Get result for manually updated bar
a2 = progressbar_rate(get_bar(our_file2.getvalue(), 3))
# 2nd case: use max smoothing (= instant rate)
with closing(StringIO()) as our_file2:
with closing(StringIO()) as our_file:
t = tqdm(range(3), file=our_file2, smoothing=1, leave=True,
miniters=1, mininterval=0)
cpu_timify(t, timer)
with tqdm(range(3), file=our_file, smoothing=1, leave=True,
miniters=1, mininterval=0) as t2:
cpu_timify(t2, timer)
for i in t2:
if i == 0:
timer.sleep(0.01)
else:
timer.sleep(0.001)
t.update()
t.close()
# Get result for iter-based bar
b = progressbar_rate(get_bar(our_file.getvalue(), 3))
# Get result for manually updated bar
b2 = progressbar_rate(get_bar(our_file2.getvalue(), 3))
# 3rd case: use medium smoothing
with closing(StringIO()) as our_file2:
with closing(StringIO()) as our_file:
t = tqdm(range(3), file=our_file2, smoothing=0.5, leave=True,
miniters=1, mininterval=0)
cpu_timify(t, timer)
t2 = tqdm(range(3), file=our_file, smoothing=0.5, leave=True,
miniters=1, mininterval=0)
cpu_timify(t2, timer)
for i in t2:
if i == 0:
timer.sleep(0.01)
else:
timer.sleep(0.001)
t.update()
t2.close()
t.close()
# Get result for iter-based bar
c = progressbar_rate(get_bar(our_file.getvalue(), 3))
# Get result for manually updated bar
c2 = progressbar_rate(get_bar(our_file2.getvalue(), 3))
# Check that medium smoothing's rate is between no and max smoothing rates
assert a <= c <= b
assert a2 <= c2 <= b2
@mark.skipif(nt_and_no_colorama, reason="Windows without colorama")
def test_deprecated_nested():
"""Test nested progress bars"""
# TODO: test degradation on windows without colorama?
# Artificially test nested loop printing
# Without leave
our_file = StringIO()
try:
tqdm(total=2, file=our_file, nested=True)
except TqdmDeprecationWarning:
if """`nested` is deprecated and automated.
Use `position` instead for manual control.""" not in our_file.getvalue():
raise
else:
raise DeprecationError("Should not allow nested kwarg")
def test_bar_format():
"""Test custom bar formatting"""
with closing(StringIO()) as our_file:
bar_format = ('{l_bar}{bar}|{n_fmt}/{total_fmt}-{n}/{total}'
'{percentage}{rate}{rate_fmt}{elapsed}{remaining}')
for _ in trange(2, file=our_file, leave=True, bar_format=bar_format):
pass
out = our_file.getvalue()
assert "\r 0%| |0/2-0/20.0None?it/s00:00?\r" in out
# Test unicode string auto conversion
with closing(StringIO()) as our_file:
bar_format = r'hello world'
with tqdm(ascii=False, bar_format=bar_format, file=our_file) as t:
assert isinstance(t.bar_format, str)
def test_custom_format():
"""Test adding additional derived format arguments"""
class TqdmExtraFormat(tqdm):
"""Provides a `total_time` format parameter"""
@property
def format_dict(self):
d = super().format_dict
total_time = d["elapsed"] * (d["total"] or 0) / max(d["n"], 1)
d.update(total_time=self.format_interval(total_time) + " in total")
return d
with closing(StringIO()) as our_file:
for _ in TqdmExtraFormat(
range(10), file=our_file,
bar_format="{total_time}: {percentage:.0f}%|{bar}{r_bar}"):
pass
assert "00:00 in total" in our_file.getvalue()
def test_eta(capsys):
"""Test eta bar_format"""
from datetime import datetime as dt
for _ in trange(999, miniters=1, mininterval=0, leave=True,
bar_format='{l_bar}{eta:%Y-%m-%d}'):
pass
_, err = capsys.readouterr()
assert f"\r100%|{dt.now():%Y-%m-%d}\n" in err
def test_unpause():
"""Test unpause"""
timer = DiscreteTimer()
with closing(StringIO()) as our_file:
t = trange(10, file=our_file, leave=True, mininterval=0)
cpu_timify(t, timer)
timer.sleep(0.01)
t.update()
timer.sleep(0.01)
t.update()
timer.sleep(0.1) # longer wait time
t.unpause()
timer.sleep(0.01)
t.update()
timer.sleep(0.01)
t.update()
t.close()
r_before = progressbar_rate(get_bar(our_file.getvalue(), 2))
r_after = progressbar_rate(get_bar(our_file.getvalue(), 3))
assert r_before == r_after
def test_disabled_unpause(capsys):
"""Test disabled unpause"""
with tqdm(total=10, disable=True) as t:
t.update()
t.unpause()
t.update()
print(t)
out, err = capsys.readouterr()
assert not err
assert out == ' 0%| | 0/10 [00:00<?, ?it/s]\n'
def test_reset():
"""Test resetting a bar for re-use"""
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file,
miniters=1, mininterval=0, maxinterval=0) as t:
t.update(9)
t.reset()
t.update()
t.reset(total=12)
t.update(10)
assert '| 1/10' in our_file.getvalue()
assert '| 10/12' in our_file.getvalue()
def test_disabled_reset(capsys):
"""Test disabled reset"""
with tqdm(total=10, disable=True) as t:
t.update(9)
t.reset()
t.update()
t.reset(total=12)
t.update(10)
print(t)
out, err = capsys.readouterr()
assert not err
assert out == ' 0%| | 0/12 [00:00<?, ?it/s]\n'
@mark.skipif(nt_and_no_colorama, reason="Windows without colorama")
def test_position():
"""Test positioned progress bars"""
# Artificially test nested loop printing
# Without leave
our_file = StringIO()
kwargs = {'file': our_file, 'miniters': 1, 'mininterval': 0, 'maxinterval': 0}
t = tqdm(total=2, desc='pos2 bar', leave=False, position=2, **kwargs)
t.update()
t.close()
out = our_file.getvalue()
res = [m[0] for m in RE_pos.findall(out)]
exres = ['\n\n\rpos2 bar: 0%',
'\n\n\rpos2 bar: 50%',
'\n\n\r ']
pos_line_diff(res, exres)
# Test iteration-based tqdm positioning
our_file = StringIO()
kwargs["file"] = our_file
for _ in trange(2, desc='pos0 bar', position=0, **kwargs):
for _ in trange(2, desc='pos1 bar', position=1, **kwargs):
for _ in trange(2, desc='pos2 bar', position=2, **kwargs):
pass
out = our_file.getvalue()
res = [m[0] for m in RE_pos.findall(out)]
exres = ['\rpos0 bar: 0%',
'\n\rpos1 bar: 0%',
'\n\n\rpos2 bar: 0%',
'\n\n\rpos2 bar: 50%',
'\n\n\rpos2 bar: 100%',
'\rpos2 bar: 100%',
'\n\n\rpos1 bar: 50%',
'\n\n\rpos2 bar: 0%',
'\n\n\rpos2 bar: 50%',
'\n\n\rpos2 bar: 100%',
'\rpos2 bar: 100%',
'\n\n\rpos1 bar: 100%',
'\rpos1 bar: 100%',
'\n\rpos0 bar: 50%',
'\n\rpos1 bar: 0%',
'\n\n\rpos2 bar: 0%',
'\n\n\rpos2 bar: 50%',
'\n\n\rpos2 bar: 100%',
'\rpos2 bar: 100%',
'\n\n\rpos1 bar: 50%',
'\n\n\rpos2 bar: 0%',
'\n\n\rpos2 bar: 50%',
'\n\n\rpos2 bar: 100%',
'\rpos2 bar: 100%',
'\n\n\rpos1 bar: 100%',
'\rpos1 bar: 100%',
'\n\rpos0 bar: 100%',
'\rpos0 bar: 100%',
'\n']
pos_line_diff(res, exres)
# Test manual tqdm positioning
our_file = StringIO()
kwargs["file"] = our_file
kwargs["total"] = 2
t1 = tqdm(desc='pos0 bar', position=0, **kwargs)
t2 = tqdm(desc='pos1 bar', position=1, **kwargs)
t3 = tqdm(desc='pos2 bar', position=2, **kwargs)
for _ in range(2):
t1.update()
t3.update()
t2.update()
out = our_file.getvalue()
res = [m[0] for m in RE_pos.findall(out)]
exres = ['\rpos0 bar: 0%',
'\n\rpos1 bar: 0%',
'\n\n\rpos2 bar: 0%',
'\rpos0 bar: 50%',
'\n\n\rpos2 bar: 50%',
'\n\rpos1 bar: 50%',
'\rpos0 bar: 100%',
'\n\n\rpos2 bar: 100%',
'\n\rpos1 bar: 100%']
pos_line_diff(res, exres)
t1.close()
t2.close()
t3.close()
# Test auto repositioning of bars when a bar is prematurely closed
# tqdm._instances.clear() # reset number of instances
with closing(StringIO()) as our_file:
t1 = tqdm(total=10, file=our_file, desc='1.pos0 bar', mininterval=0)
t2 = tqdm(total=10, file=our_file, desc='2.pos1 bar', mininterval=0)
t3 = tqdm(total=10, file=our_file, desc='3.pos2 bar', mininterval=0)
res = [m[0] for m in RE_pos.findall(our_file.getvalue())]
exres = ['\r1.pos0 bar: 0%',
'\n\r2.pos1 bar: 0%',
'\n\n\r3.pos2 bar: 0%']
pos_line_diff(res, exres)
t2.close()
t4 = tqdm(total=10, file=our_file, desc='4.pos2 bar', mininterval=0)
t1.update(1)
t3.update(1)
t4.update(1)
res = [m[0] for m in RE_pos.findall(our_file.getvalue())]
exres = ['\r1.pos0 bar: 0%',
'\n\r2.pos1 bar: 0%',
'\n\n\r3.pos2 bar: 0%',
'\r2.pos1 bar: 0%',
'\n\n\r4.pos2 bar: 0%',
'\r1.pos0 bar: 10%',
'\n\n\r3.pos2 bar: 10%',
'\n\r4.pos2 bar: 10%']
pos_line_diff(res, exres)
t4.close()
t3.close()
t1.close()
def test_set_description():
"""Test set description"""
with closing(StringIO()) as our_file:
with tqdm(desc='Hello', file=our_file) as t:
assert t.desc == 'Hello'
t.set_description_str('World')
assert t.desc == 'World'
t.set_description()
assert t.desc == ''
t.set_description('Bye')
assert t.desc == 'Bye: '
assert "World" in our_file.getvalue()
# without refresh
with closing(StringIO()) as our_file:
with tqdm(desc='Hello', file=our_file) as t:
assert t.desc == 'Hello'
t.set_description_str('World', False)
assert t.desc == 'World'
t.set_description(None, False)
assert t.desc == ''
assert "World" not in our_file.getvalue()
# unicode
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file) as t:
t.set_description(u"\xe1\xe9\xed\xf3\xfa")
def test_deprecated_gui():
"""Test internal GUI properties"""
# Check: StatusPrinter iff gui is disabled
with closing(StringIO()) as our_file:
t = tqdm(total=2, gui=True, file=our_file, miniters=1, mininterval=0)
assert not hasattr(t, "sp")
try:
t.update(1)
except TqdmDeprecationWarning as e:
if (
'Please use `tqdm.gui.tqdm(...)` instead of `tqdm(..., gui=True)`'
not in our_file.getvalue()
):
raise e
else:
raise DeprecationError('Should not allow manual gui=True without'
' overriding __iter__() and update()')
finally:
t._instances.clear()
# t.close()
# len(tqdm._instances) += 1 # undo the close() decrement
t = tqdm(range(3), gui=True, file=our_file, miniters=1, mininterval=0)
try:
for _ in t:
pass
except TqdmDeprecationWarning as e:
if (
'Please use `tqdm.gui.tqdm(...)` instead of `tqdm(..., gui=True)`'
not in our_file.getvalue()
):
raise e
else:
raise DeprecationError('Should not allow manual gui=True without'
' overriding __iter__() and update()')
finally:
t._instances.clear()
# t.close()
# len(tqdm._instances) += 1 # undo the close() decrement
with tqdm(total=1, gui=False, file=our_file) as t:
assert hasattr(t, "sp")
def test_cmp():
"""Test comparison functions"""
with closing(StringIO()) as our_file:
t0 = tqdm(total=10, file=our_file)
t1 = tqdm(total=10, file=our_file)
t2 = tqdm(total=10, file=our_file)
assert t0 < t1
assert t2 >= t0
assert t0 <= t2
t3 = tqdm(total=10, file=our_file)
t4 = tqdm(total=10, file=our_file)
t5 = tqdm(total=10, file=our_file)
t5.close()
t6 = tqdm(total=10, file=our_file)
assert t3 != t4
assert t3 > t2
assert t5 == t6
t6.close()
t4.close()
t3.close()
t2.close()
t1.close()
t0.close()
def test_repr():
"""Test representation"""
with closing(StringIO()) as our_file:
with tqdm(total=10, ascii=True, file=our_file) as t:
assert str(t) == ' 0%| | 0/10 [00:00<?, ?it/s]'
def test_clear():
"""Test clearing bar display"""
with closing(StringIO()) as our_file:
t1 = tqdm(total=10, file=our_file, desc='pos0 bar', bar_format='{l_bar}')
t2 = trange(10, file=our_file, desc='pos1 bar', bar_format='{l_bar}')
before = squash_ctrlchars(our_file.getvalue())
t2.clear()
t1.clear()
after = squash_ctrlchars(our_file.getvalue())
t1.close()
t2.close()
assert before == ['pos0 bar: 0%|', 'pos1 bar: 0%|']
assert after == ['', '']
def test_clear_disabled():
"""Test disabled clear"""
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file, desc='pos0 bar', disable=True,
bar_format='{l_bar}') as t:
t.clear()
assert our_file.getvalue() == ''
def test_refresh():
"""Test refresh bar display"""
with closing(StringIO()) as our_file:
t1 = tqdm(total=10, file=our_file, desc='pos0 bar',
bar_format='{l_bar}', mininterval=999, miniters=999)
t2 = tqdm(total=10, file=our_file, desc='pos1 bar',
bar_format='{l_bar}', mininterval=999, miniters=999)
t1.update()
t2.update()
before = squash_ctrlchars(our_file.getvalue())
t1.refresh()
t2.refresh()
after = squash_ctrlchars(our_file.getvalue())
t1.close()
t2.close()
# Check that refreshing indeed forced the display to use realtime state
assert before == [u'pos0 bar: 0%|', u'pos1 bar: 0%|']
assert after == [u'pos0 bar: 10%|', u'pos1 bar: 10%|']
def test_disabled_repr(capsys):
"""Test disabled repr"""
with tqdm(total=10, disable=True) as t:
str(t)
t.update()
print(t)
out, err = capsys.readouterr()
assert not err
assert out == ' 0%| | 0/10 [00:00<?, ?it/s]\n'
def test_disabled_refresh():
"""Test disabled refresh"""
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file, desc='pos0 bar', disable=True,
bar_format='{l_bar}', mininterval=999, miniters=999) as t:
t.update()
t.refresh()
assert our_file.getvalue() == ''
def test_write():
"""Test write messages"""
s = "Hello world"
with closing(StringIO()) as our_file:
# Change format to keep only left part w/o bar and it/s rate
t1 = tqdm(total=10, file=our_file, desc='pos0 bar',
bar_format='{l_bar}', mininterval=0, miniters=1)
t2 = trange(10, file=our_file, desc='pos1 bar', bar_format='{l_bar}',
mininterval=0, miniters=1)
t3 = tqdm(total=10, file=our_file, desc='pos2 bar',
bar_format='{l_bar}', mininterval=0, miniters=1)
t1.update()
t2.update()
t3.update()
before = our_file.getvalue()
# Write msg and see if bars are correctly redrawn below the msg
t1.write(s, file=our_file) # call as an instance method
tqdm.write(s, file=our_file) # call as a class method
after = our_file.getvalue()
t1.close()
t2.close()
t3.close()
before_squashed = squash_ctrlchars(before)
after_squashed = squash_ctrlchars(after)
assert after_squashed == [s, s] + before_squashed
# Check that no bar clearing if different file
with closing(StringIO()) as our_file_bar:
with closing(StringIO()) as our_file_write:
t1 = tqdm(total=10, file=our_file_bar, desc='pos0 bar',
bar_format='{l_bar}', mininterval=0, miniters=1)
t1.update()
before_bar = our_file_bar.getvalue()
tqdm.write(s, file=our_file_write)
after_bar = our_file_bar.getvalue()
t1.close()
assert before_bar == after_bar
# Test stdout/stderr anti-mixup strategy
# Backup stdout/stderr
stde = sys.stderr
stdo = sys.stdout
# Mock stdout/stderr
with closing(StringIO()) as our_stderr:
with closing(StringIO()) as our_stdout:
sys.stderr = our_stderr
sys.stdout = our_stdout
t1 = tqdm(total=10, file=sys.stderr, desc='pos0 bar',
bar_format='{l_bar}', mininterval=0, miniters=1)
t1.update()
before_err = sys.stderr.getvalue()
before_out = sys.stdout.getvalue()
tqdm.write(s, file=sys.stdout)
after_err = sys.stderr.getvalue()
after_out = sys.stdout.getvalue()
t1.close()
assert before_err == '\rpos0 bar: 0%|\rpos0 bar: 10%|'
assert before_out == ''
after_err_res = [m[0] for m in RE_pos.findall(after_err)]
exres = ['\rpos0 bar: 0%|',
'\rpos0 bar: 10%|',
'\r ',
'\r\rpos0 bar: 10%|']
pos_line_diff(after_err_res, exres)
assert after_out == s + '\n'
# Restore stdout and stderr
sys.stderr = stde
sys.stdout = stdo
def test_len():
"""Test advance len (numpy array shape)"""
np = importorskip('numpy')
with closing(StringIO()) as f:
with tqdm(np.zeros((3, 4)), file=f) as t:
assert len(t) == 3
def test_autodisable_disable():
"""Test autodisable will disable on non-TTY"""
with closing(StringIO()) as our_file:
with tqdm(total=10, disable=None, file=our_file) as t:
t.update(3)
assert our_file.getvalue() == ''
def test_autodisable_enable():
"""Test autodisable will not disable on TTY"""
with closing(StringIO()) as our_file:
our_file.isatty = lambda: True
with tqdm(total=10, disable=None, file=our_file) as t:
t.update()
assert our_file.getvalue() != ''
def test_deprecation_exception():
def test_TqdmDeprecationWarning():
with closing(StringIO()) as our_file:
raise (TqdmDeprecationWarning('Test!', fp_write=getattr(
our_file, 'write', sys.stderr.write)))
def test_TqdmDeprecationWarning_nofpwrite():
raise TqdmDeprecationWarning('Test!', fp_write=None)
raises(TqdmDeprecationWarning, test_TqdmDeprecationWarning)
raises(Exception, test_TqdmDeprecationWarning_nofpwrite)
def test_postfix():
"""Test postfix"""
postfix = {'float': 0.321034, 'gen': 543, 'str': 'h', 'lst': [2]}
postfix_order = (('w', 'w'), ('a', 0)) # no need for OrderedDict
expected = ['float=0.321', 'gen=543', 'lst=[2]', 'str=h']
expected_order = ['w=w', 'a=0', 'float=0.321', 'gen=543', 'lst=[2]', 'str=h']
# Test postfix set at init
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file, desc='pos0 bar',
bar_format='{r_bar}', postfix=postfix) as t1:
t1.refresh()
out = our_file.getvalue()
# Test postfix set after init
with closing(StringIO()) as our_file:
with trange(10, file=our_file, desc='pos1 bar', bar_format='{r_bar}',
postfix=None) as t2:
t2.set_postfix(**postfix)
t2.refresh()
out2 = our_file.getvalue()
# Order of items in dict may change, so need a loop to check per item
for res in expected:
assert res in out
assert res in out2
# Test postfix (with ordered dict and no refresh) set after init
with closing(StringIO()) as our_file:
with trange(10, file=our_file, desc='pos2 bar', bar_format='{r_bar}',
postfix=None) as t3:
t3.set_postfix(postfix_order, False, **postfix)
t3.refresh() # explicit external refresh
out3 = our_file.getvalue()
out3 = out3[1:-1].split(', ')[3:]
assert out3 == expected_order
# Test postfix (with ordered dict and refresh) set after init
with closing(StringIO()) as our_file:
with trange(10, file=our_file, desc='pos2 bar',
bar_format='{r_bar}', postfix=None) as t4:
t4.set_postfix(postfix_order, True, **postfix)
t4.refresh() # double refresh
out4 = our_file.getvalue()
assert out4.count('\r') > out3.count('\r')
assert out4.count(", ".join(expected_order)) == 2
# Test setting postfix string directly
with closing(StringIO()) as our_file:
with trange(10, file=our_file, desc='pos2 bar', bar_format='{r_bar}',
postfix=None) as t5:
t5.set_postfix_str("Hello", False)
t5.set_postfix_str("World")
out5 = our_file.getvalue()
assert "Hello" not in out5
out5 = out5[1:-1].split(', ')[3:]
assert out5 == ["World"]
def test_postfix_direct():
"""Test directly assigning non-str objects to postfix"""
with closing(StringIO()) as our_file:
with tqdm(total=10, file=our_file, miniters=1, mininterval=0,
bar_format="{postfix[0][name]} {postfix[1]:>5.2f}",
postfix=[{'name': "foo"}, 42]) as t:
for i in range(10):
if i % 2:
t.postfix[0]["name"] = "abcdefghij"[i]
else:
t.postfix[1] = i
t.update()
res = our_file.getvalue()
assert "f 6.00" in res
assert "h 6.00" in res
assert "h 8.00" in res
assert "j 8.00" in res
@contextmanager
def std_out_err_redirect_tqdm(tqdm_file=sys.stderr):
orig_out_err = sys.stdout, sys.stderr
try:
sys.stdout = sys.stderr = DummyTqdmFile(tqdm_file)
yield orig_out_err[0]
# Relay exceptions
except Exception as exc:
raise exc
# Always restore sys.stdout/err if necessary
finally:
sys.stdout, sys.stderr = orig_out_err
def test_file_redirection():
"""Test redirection of output"""
with closing(StringIO()) as our_file:
# Redirect stdout to tqdm.write()
with std_out_err_redirect_tqdm(tqdm_file=our_file):
with tqdm(total=3) as pbar:
print("Such fun")
pbar.update(1)
print("Such", "fun")
pbar.update(1)
print("Such ", end="")
print("fun")
pbar.update(1)
res = our_file.getvalue()
assert res.count("Such fun\n") == 3
assert "0/3" in res
assert "3/3" in res
def test_external_write():
"""Test external write mode"""
with closing(StringIO()) as our_file:
# Redirect stdout to tqdm.write()
for _ in trange(3, file=our_file):
del tqdm._lock # classmethod should be able to recreate lock
with tqdm.external_write_mode(file=our_file):
our_file.write("Such fun\n")
res = our_file.getvalue()
assert res.count("Such fun\n") == 3
assert "0/3" in res
assert "3/3" in res
def test_unit_scale():
"""Test numeric `unit_scale`"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(9), unit_scale=9, file=our_file,
miniters=1, mininterval=0):
pass
out = our_file.getvalue()
assert '81/81' in out
def patch_lock(thread=True):
"""decorator replacing tqdm's lock with vanilla threading/multiprocessing"""
try:
if thread:
from threading import RLock
else:
from multiprocessing import RLock
lock = RLock()
except (ImportError, OSError) as err:
skip(str(err))
def outer(func):
"""actual decorator"""
@wraps(func)
def inner(*args, **kwargs):
"""set & reset lock even if exceptions occur"""
default_lock = tqdm.get_lock()
try:
tqdm.set_lock(lock)
return func(*args, **kwargs)
finally:
tqdm.set_lock(default_lock)
return inner
return outer
@patch_lock(thread=False)
def test_threading():
"""Test multiprocess/thread-realted features"""
pass # TODO: test interleaved output #445
def test_bool():
"""Test boolean cast"""
def internal(our_file, disable):
kwargs = {'file': our_file, 'disable': disable}
with trange(10, **kwargs) as t:
assert t
with trange(0, **kwargs) as t:
assert not t
with tqdm(total=10, **kwargs) as t:
assert bool(t)
with tqdm(total=0, **kwargs) as t:
assert not bool(t)
with tqdm([], **kwargs) as t:
assert not t
with tqdm([0], **kwargs) as t:
assert t
with tqdm(iter([]), **kwargs) as t:
assert t
with tqdm(iter([1, 2, 3]), **kwargs) as t:
assert t
with tqdm(**kwargs) as t:
try:
print(bool(t))
except TypeError:
pass
else:
raise TypeError("Expected bool(tqdm()) to fail")
# test with and without disable
with closing(StringIO()) as our_file:
internal(our_file, False)
internal(our_file, True)
def backendCheck(module):
"""Test tqdm-like module fallback"""
tn = module.tqdm
tr = module.trange
with closing(StringIO()) as our_file:
with tn(total=10, file=our_file) as t:
assert len(t) == 10
with tr(1337) as t:
assert len(t) == 1337
def test_auto():
"""Test auto fallback"""
from tqdm import auto, autonotebook
backendCheck(autonotebook)
backendCheck(auto)
def test_wrapattr():
"""Test wrapping file-like objects"""
data = "a twenty-char string"
with closing(StringIO()) as our_file:
with closing(StringIO()) as writer:
with tqdm.wrapattr(writer, "write", file=our_file, bytes=True) as wrap:
wrap.write(data)
res = writer.getvalue()
assert data == res
res = our_file.getvalue()
assert '%.1fB [' % len(data) in res
with closing(StringIO()) as our_file:
with closing(StringIO()) as writer:
with tqdm.wrapattr(writer, "write", file=our_file, bytes=False) as wrap:
wrap.write(data)
res = our_file.getvalue()
assert '%dit [' % len(data) in res
def test_float_progress():
"""Test float totals"""
with closing(StringIO()) as our_file:
with trange(10, total=9.6, file=our_file) as t:
with catch_warnings(record=True) as w:
simplefilter("always", category=TqdmWarning)
for i in t:
if i < 9:
assert not w
assert w
assert "clamping frac" in str(w[-1].message)
def test_screen_shape():
"""Test screen shape"""
# ncols
with closing(StringIO()) as our_file:
with trange(10, file=our_file, ncols=50) as t:
list(t)
res = our_file.getvalue()
assert all(len(i) == 50 for i in get_bar(res))
# no second/third bar, leave=False
with closing(StringIO()) as our_file:
kwargs = {'file': our_file, 'ncols': 50, 'nrows': 2, 'miniters': 0,
'mininterval': 0, 'leave': False}
with trange(10, desc="one", **kwargs) as t1:
with trange(10, desc="two", **kwargs) as t2:
with trange(10, desc="three", **kwargs) as t3:
list(t3)
list(t2)
list(t1)
res = our_file.getvalue()
assert "one" in res
assert "two" not in res
assert "three" not in res
assert "\n\n" not in res
assert "more hidden" in res
# double-check ncols
assert all(len(i) == 50 for i in get_bar(res)
if i.strip() and "more hidden" not in i)
# all bars, leave=True
with closing(StringIO()) as our_file:
kwargs = {'file': our_file, 'ncols': 50, 'nrows': 2,
'miniters': 0, 'mininterval': 0}
with trange(10, desc="one", **kwargs) as t1:
with trange(10, desc="two", **kwargs) as t2:
assert "two" not in our_file.getvalue()
with trange(10, desc="three", **kwargs) as t3:
assert "three" not in our_file.getvalue()
list(t3)
list(t2)
list(t1)
res = our_file.getvalue()
assert "one" in res
assert "two" in res
assert "three" in res
assert "\n\n" not in res
assert "more hidden" in res
# double-check ncols
assert all(len(i) == 50 for i in get_bar(res)
if i.strip() and "more hidden" not in i)
# second bar becomes first, leave=False
with closing(StringIO()) as our_file:
kwargs = {'file': our_file, 'ncols': 50, 'nrows': 2, 'miniters': 0,
'mininterval': 0, 'leave': False}
t1 = tqdm(total=10, desc="one", **kwargs)
with tqdm(total=10, desc="two", **kwargs) as t2:
t1.update()
t2.update()
t1.close()
res = our_file.getvalue()
assert "one" in res
assert "two" not in res
assert "more hidden" in res
t2.update()
res = our_file.getvalue()
assert "two" in res
def test_initial():
"""Test `initial`"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(9), initial=10, total=19, file=our_file,
miniters=1, mininterval=0):
pass
out = our_file.getvalue()
assert '10/19' in out
assert '19/19' in out
def test_colour():
"""Test `colour`"""
with closing(StringIO()) as our_file:
for _ in tqdm(range(9), file=our_file, colour="#beefed"):
pass
out = our_file.getvalue()
assert '\x1b[38;2;%d;%d;%dm' % (0xbe, 0xef, 0xed) in out
with catch_warnings(record=True) as w:
simplefilter("always", category=TqdmWarning)
with tqdm(total=1, file=our_file, colour="charm") as t:
assert w
t.update()
assert "Unknown colour" in str(w[-1].message)
with closing(StringIO()) as our_file2:
for _ in tqdm(range(9), file=our_file2, colour="blue"):
pass
out = our_file2.getvalue()
assert '\x1b[34m' in out
def test_closed():
"""Test writing to closed file"""
with closing(StringIO()) as our_file:
for i in trange(9, file=our_file, miniters=1, mininterval=0):
if i == 5:
our_file.close()
def test_reversed(capsys):
"""Test reversed()"""
for _ in reversed(tqdm(range(9))):
pass
out, err = capsys.readouterr()
assert not out
assert ' 0%' in err
assert '100%' in err
def test_contains(capsys):
"""Test __contains__ doesn't iterate"""
with tqdm(list(range(9))) as t:
assert 9 not in t
assert all(i in t for i in range(9))
out, err = capsys.readouterr()
assert not out
assert ' 0%' in err
assert '100%' not in err