| # -*- 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. |
| from __future__ import print_function |
| |
| 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 |
| |
| try: |
| _range = xrange |
| except NameError: |
| _range = range |
| |
| try: |
| _unicode = unicode |
| except NameError: |
| _unicode = str |
| |
| 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(UnicodeIO, self).__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' |
| |
| |
| 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 |
| if sys.version_info >= (3,): |
| 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 "{0:5a}".format(Bar(0.3)) == "#5 " |
| assert "{0:2}".format(Bar(0.5, charset=" .oO0")) == "0 " |
| assert "{0:2a}".format(Bar(0.5, charset=" .oO0")) == "# " |
| assert "{0:-6a}".format(Bar(0.5, 10)) == '## ' |
| assert "{0:2b}".format(Bar(0.5, 10)) == ' ' |
| |
| |
| 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(WriteTypeChecker, self).__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("\n<<< Expected:\n{0}\n>>> Got:\n{1}\n===".format( |
| exres + ', ...it/s]\n', our_file.getvalue())) |
| |
| # 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, _unicode) |
| |
| |
| 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(TqdmExtraFormat, self).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 "\r100%|{eta:%Y-%m-%d}\n".format(eta=dt.now()) 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 |
