Test ProgressBar with parallel computing, advanced version.
(capsys)
| 105 | |
| 106 | |
| 107 | def test_progressbar_parallel_more(capsys): |
| 108 | """Test ProgressBar with parallel computing, advanced version.""" |
| 109 | assert capsys.readouterr().out == "" |
| 110 | # This must be "1" because "capsys" won't get stdout properly otherwise |
| 111 | parallel, p_fun, _ = parallel_func(_identity_block_wide, n_jobs=1, verbose=False) |
| 112 | arr = np.arange(10) |
| 113 | with use_log_level(True): |
| 114 | with ProgressBar(len(arr) * 2) as pb: |
| 115 | out = parallel( |
| 116 | p_fun(x, pb.subset(pb_idx)) |
| 117 | for pb_idx, x in array_split_idx(arr, 2, n_per_split=2) |
| 118 | ) |
| 119 | idxs = np.concatenate([o[1] for o in out]) |
| 120 | assert_array_equal(idxs, np.arange(len(arr) * 2)) |
| 121 | out = np.concatenate([o[0] for o in out]) |
| 122 | assert Path(pb._mmap_fname).is_file() |
| 123 | sum_ = np.memmap( |
| 124 | pb._mmap_fname, dtype="bool", mode="r", shape=len(arr) * 2 |
| 125 | ).sum() |
| 126 | assert sum_ == len(arr) * 2 |
| 127 | assert not Path(pb._mmap_fname).is_file(), "__exit__ not called?" |
| 128 | cap = capsys.readouterr() |
| 129 | out = cap.err |
| 130 | assert "100%" in out |
nothing calls this directly
no test coverage detected