from copy import deepcopy import os.path as op import os import sys import webbrowser import numpy as np import warnings import pytest from scipy import sparse from numpy.testing import assert_equal, assert_array_equal, assert_allclose from mne import read_evokeds, open_docs from mne.datasets import testing from mne.externals.six.moves import StringIO from mne.io import show_fiff, read_raw_fif from mne.epochs import _segment_raw from mne.parallel import parallel_func from mne.time_frequency import tfr_morlet from mne.utils import (set_log_level, set_log_file, _TempDir, get_config, set_config, deprecated, _fetch_file, sum_squared, estimate_rank, _reg_pinv, _url_to_local_path, sizeof_fmt, _check_subject, _check_type_picks, object_hash, object_diff, requires_good_network, run_tests_if_main, md5sum, ArgvSetter, _memory_usage, check_random_state, _check_mayavi_version, requires_mayavi, traits_test, set_memmap_min_size, _get_stim_channel, _check_fname, create_slices, _time_mask, random_permutation, _get_call_line, compute_corr, sys_info, verbose, check_fname, get_config_path, warn, object_size, buggy_mkl_svd, _get_inst_data, copy_doc, copy_function_doc_to_method_doc, ProgressBar, linkcode_resolve, array_split_idx, filter_out_warnings) base_dir = op.join(op.dirname(__file__), '..', 'io', 'tests', 'data') fname_evoked = op.join(base_dir, 'test-ave.fif') fname_raw = op.join(base_dir, 'test_raw.fif') fname_log = op.join(base_dir, 'test-ave.log') fname_log_2 = op.join(base_dir, 'test-ave-2.log') data_path = testing.data_path(download=False) fname_fsaverage_trans = op.join(data_path, 'subjects', 'fsaverage', 'bem', 'fsaverage-trans.fif') def clean_lines(lines=[]): """Scrub filenames for checking logging output (in test_logging).""" return [l if 'Reading ' not in l else 'Reading test file' for l in lines] def test_buggy_mkl(): """Test decorator for buggy MKL issues.""" from unittest import SkipTest @buggy_mkl_svd def foo(a, b): raise np.linalg.LinAlgError('SVD did not converge') with pytest.warns(RuntimeWarning, match='convergence error'): pytest.raises(SkipTest, foo, 1, 2) @buggy_mkl_svd def bar(c, d, e): raise RuntimeError('SVD did not converge') pytest.raises(RuntimeError, bar, 1, 2, 3) def test_sys_info(): """Test info-showing utility.""" out = StringIO() sys_info(fid=out) out = out.getvalue() assert ('numpy:' in out) def test_get_call_line(): """Test getting a call line.""" @verbose def foo(verbose=None): return _get_call_line(in_verbose=True) for v in (None, True): my_line = foo(verbose=v) # testing assert_equal(my_line, 'my_line = foo(verbose=v) # testing') def bar(): return _get_call_line(in_verbose=False) my_line = bar() # testing more assert_equal(my_line, 'my_line = bar() # testing more') def test_object_size(): """Test object size estimation.""" assert (object_size(np.ones(10, np.float32)) < object_size(np.ones(10, np.float64))) for lower, upper, obj in ((0, 60, ''), (0, 30, 1), (0, 30, 1.), (0, 70, 'foo'), (0, 150, np.ones(0)), (0, 150, np.int32(1)), (150, 500, np.ones(20)), (100, 400, dict()), (400, 1000, dict(a=np.ones(50))), (200, 900, sparse.eye(20, format='csc')), (200, 900, sparse.eye(20, format='csr'))): size = object_size(obj) assert lower < size < upper, \ '%s < %s < %s:\n%s' % (lower, size, upper, obj) def test_get_inst_data(): """Test _get_inst_data.""" raw = read_raw_fif(fname_raw) raw.crop(tmax=1.) assert_equal(_get_inst_data(raw), raw._data) raw.pick_channels(raw.ch_names[:2]) epochs = _segment_raw(raw, 0.5) assert_equal(_get_inst_data(epochs), epochs._data) evoked = epochs.average() assert_equal(_get_inst_data(evoked), evoked.data) evoked.crop(tmax=0.1) picks = list(range(2)) freqs = np.array([50., 55.]) n_cycles = 3 tfr = tfr_morlet(evoked, freqs, n_cycles, return_itc=False, picks=picks) assert_equal(_get_inst_data(tfr), tfr.data) pytest.raises(TypeError, _get_inst_data, 'foo') def test_misc(): """Test misc utilities.""" assert_equal(_memory_usage(-1)[0], -1) assert_equal(_memory_usage((clean_lines, [], {}))[0], -1) assert_equal(_memory_usage(clean_lines)[0], -1) pytest.raises(ValueError, check_random_state, 'foo') pytest.raises(ValueError, set_memmap_min_size, 1) pytest.raises(ValueError, set_memmap_min_size, 'foo') pytest.raises(TypeError, get_config, 1) pytest.raises(TypeError, set_config, 1) pytest.raises(TypeError, set_config, 'foo', 1) pytest.raises(TypeError, _get_stim_channel, 1, None) pytest.raises(TypeError, _get_stim_channel, [1], None) pytest.raises(TypeError, _check_fname, 1) pytest.raises(IOError, check_fname, 'foo', 'tets-dip.x', (), ('.fif',)) pytest.raises(ValueError, _check_subject, None, None) pytest.raises(TypeError, _check_subject, None, 1) pytest.raises(TypeError, _check_subject, 1, None) @requires_mayavi @traits_test def test_check_mayavi(): """Test mayavi version check.""" pytest.raises(RuntimeError, _check_mayavi_version, '100.0.0') def test_run_tests_if_main(): """Test run_tests_if_main functionality.""" x = [] def test_a(): x.append(True) @pytest.mark.skipif(True) def test_b(): return try: __name__ = '__main__' run_tests_if_main(measure_mem=False) # dual meas causes problems def test_c(): raise RuntimeError try: __name__ = '__main__' run_tests_if_main(measure_mem=False) # dual meas causes problems except RuntimeError: pass else: raise RuntimeError('Error not raised') finally: del __name__ assert (len(x) == 2) assert (x[0] and x[1]) def test_hash(): """Test dictionary hashing and comparison functions.""" # does hashing all of these types work: # {dict, list, tuple, ndarray, str, float, int, None} d0 = dict(a=dict(a=0.1, b='fo', c=1), b=[1, 'b'], c=(), d=np.ones(3), e=None) d0[1] = None d0[2.] = b'123' d1 = deepcopy(d0) assert (len(object_diff(d0, d1)) == 0) assert (len(object_diff(d1, d0)) == 0) assert_equal(object_hash(d0), object_hash(d1)) # change values slightly d1['data'] = np.ones(3, int) d1['d'][0] = 0 assert object_hash(d0) != object_hash(d1) d1 = deepcopy(d0) assert_equal(object_hash(d0), object_hash(d1)) d1['a']['a'] = 0.11 assert (len(object_diff(d0, d1)) > 0) assert (len(object_diff(d1, d0)) > 0) assert object_hash(d0) != object_hash(d1) d1 = deepcopy(d0) assert_equal(object_hash(d0), object_hash(d1)) d1['a']['d'] = 0 # non-existent key assert (len(object_diff(d0, d1)) > 0) assert (len(object_diff(d1, d0)) > 0) assert object_hash(d0) != object_hash(d1) d1 = deepcopy(d0) assert_equal(object_hash(d0), object_hash(d1)) d1['b'].append(0) # different-length lists assert (len(object_diff(d0, d1)) > 0) assert (len(object_diff(d1, d0)) > 0) assert object_hash(d0) != object_hash(d1) d1 = deepcopy(d0) assert_equal(object_hash(d0), object_hash(d1)) d1['e'] = 'foo' # non-None assert (len(object_diff(d0, d1)) > 0) assert (len(object_diff(d1, d0)) > 0) assert object_hash(d0) != object_hash(d1) d1 = deepcopy(d0) d2 = deepcopy(d0) d1['e'] = StringIO() d2['e'] = StringIO() d2['e'].write('foo') assert (len(object_diff(d0, d1)) > 0) assert (len(object_diff(d1, d0)) > 0) d1 = deepcopy(d0) d1[1] = 2 assert (len(object_diff(d0, d1)) > 0) assert (len(object_diff(d1, d0)) > 0) assert object_hash(d0) != object_hash(d1) # generators (and other types) not supported d1 = deepcopy(d0) d2 = deepcopy(d0) d1[1] = (x for x in d0) d2[1] = (x for x in d0) pytest.raises(RuntimeError, object_diff, d1, d2) pytest.raises(RuntimeError, object_hash, d1) x = sparse.eye(2, 2, format='csc') y = sparse.eye(2, 2, format='csr') assert ('type mismatch' in object_diff(x, y)) y = sparse.eye(2, 2, format='csc') assert_equal(len(object_diff(x, y)), 0) y[1, 1] = 2 assert ('elements' in object_diff(x, y)) y = sparse.eye(3, 3, format='csc') assert ('shape' in object_diff(x, y)) y = 0 assert ('type mismatch' in object_diff(x, y)) # smoke test for gh-4796 assert object_hash(np.int64(1)) != 0 assert object_hash(np.bool_(True)) != 0 def test_md5sum(): """Test md5sum calculation.""" tempdir = _TempDir() fname1 = op.join(tempdir, 'foo') fname2 = op.join(tempdir, 'bar') with open(fname1, 'wb') as fid: fid.write(b'abcd') with open(fname2, 'wb') as fid: fid.write(b'efgh') assert_equal(md5sum(fname1), md5sum(fname1, 1)) assert_equal(md5sum(fname2), md5sum(fname2, 1024)) assert (md5sum(fname1) != md5sum(fname2)) def test_tempdir(): """Test TempDir.""" tempdir2 = _TempDir() assert (op.isdir(tempdir2)) x = str(tempdir2) del tempdir2 assert (not op.isdir(x)) def test_estimate_rank(): """Test rank estimation.""" data = np.eye(10) assert_array_equal(estimate_rank(data, return_singular=True)[1], np.ones(10)) data[0, 0] = 0 assert_equal(estimate_rank(data), 9) pytest.raises(ValueError, estimate_rank, data, 'foo') def test_logging(): """Test logging (to file).""" pytest.raises(ValueError, set_log_level, 'foo') tempdir = _TempDir() test_name = op.join(tempdir, 'test.log') with open(fname_log, 'r') as old_log_file: # [:-1] used to strip an extra "No baseline correction applied" old_lines = clean_lines(old_log_file.readlines()) old_lines.pop(-1) with open(fname_log_2, 'r') as old_log_file_2: old_lines_2 = clean_lines(old_log_file_2.readlines()) old_lines_2.pop(14) old_lines_2.pop(-1) if op.isfile(test_name): os.remove(test_name) # test it one way (printing default off) set_log_file(test_name) set_log_level('WARNING') # should NOT print evoked = read_evokeds(fname_evoked, condition=1) with open(test_name) as fid: assert (fid.readlines() == []) # should NOT print evoked = read_evokeds(fname_evoked, condition=1, verbose=False) with open(test_name) as fid: assert (fid.readlines() == []) # should NOT print evoked = read_evokeds(fname_evoked, condition=1, verbose='WARNING') with open(test_name) as fid: assert (fid.readlines() == []) # SHOULD print evoked = read_evokeds(fname_evoked, condition=1, verbose=True) with open(test_name, 'r') as new_log_file: new_lines = clean_lines(new_log_file.readlines()) assert_equal(new_lines, old_lines) set_log_file(None) # Need to do this to close the old file os.remove(test_name) # now go the other way (printing default on) set_log_file(test_name) set_log_level('INFO') # should NOT print evoked = read_evokeds(fname_evoked, condition=1, verbose='WARNING') with open(test_name) as fid: assert (fid.readlines() == []) # should NOT print evoked = read_evokeds(fname_evoked, condition=1, verbose=False) with open(test_name) as fid: assert (fid.readlines() == []) # SHOULD print evoked = read_evokeds(fname_evoked, condition=1) with open(test_name, 'r') as new_log_file: new_lines = clean_lines(new_log_file.readlines()) assert_equal(new_lines, old_lines) # check to make sure appending works (and as default, raises a warning) set_log_file(test_name, overwrite=False) with pytest.warns(RuntimeWarning, match='appended to the file'): set_log_file(test_name) evoked = read_evokeds(fname_evoked, condition=1) with open(test_name, 'r') as new_log_file: new_lines = clean_lines(new_log_file.readlines()) assert_equal(new_lines, old_lines_2) # make sure overwriting works set_log_file(test_name, overwrite=True) # this line needs to be called to actually do some logging evoked = read_evokeds(fname_evoked, condition=1) del evoked with open(test_name, 'r') as new_log_file: new_lines = clean_lines(new_log_file.readlines()) assert_equal(new_lines, old_lines) @testing.requires_testing_data def test_datasets(): """Test dataset config.""" # gh-4192 data_path = testing.data_path(download=False) os.environ['MNE_DATASETS_TESTING_PATH'] = op.dirname(data_path) assert testing.data_path(download=False) == data_path def test_config(): """Test mne-python config file support.""" tempdir = _TempDir() key = '_MNE_PYTHON_CONFIG_TESTING' value = '123456' value2 = '123' old_val = os.getenv(key, None) os.environ[key] = value assert (get_config(key) == value) del os.environ[key] # catch the warning about it being a non-standard config key assert (len(set_config(None, None)) > 10) # tuple of valid keys with pytest.warns(RuntimeWarning, match='non-standard'): set_config(key, None, home_dir=tempdir, set_env=False) assert (get_config(key, home_dir=tempdir) is None) pytest.raises(KeyError, get_config, key, raise_error=True) assert (key not in os.environ) with pytest.warns(RuntimeWarning, match='non-standard'): set_config(key, value, home_dir=tempdir, set_env=True) assert (key in os.environ) assert (get_config(key, home_dir=tempdir) == value) with pytest.warns(RuntimeWarning, match='non-standard'): set_config(key, None, home_dir=tempdir, set_env=True) assert (key not in os.environ) with pytest.warns(RuntimeWarning, match='non-standard'): set_config(key, None, home_dir=tempdir, set_env=True) assert (key not in os.environ) if old_val is not None: os.environ[key] = old_val # Check if get_config with key=None returns all config key = 'MNE_PYTHON_TESTING_KEY' assert key not in get_config(home_dir=tempdir) with pytest.warns(RuntimeWarning, match='non-standard'): set_config(key, value, home_dir=tempdir) assert_equal(get_config(home_dir=tempdir)[key], value) old_val = os.environ.get(key) try: # os.environ should take precedence over config file os.environ[key] = value2 assert_equal(get_config(home_dir=tempdir)[key], value2) finally: # reset os.environ if old_val is None: os.environ.pop(key, None) else: os.environ[key] = old_val # Check what happens when we use a corrupted file json_fname = get_config_path(home_dir=tempdir) with open(json_fname, 'w') as fid: fid.write('foo{}') with pytest.warns(RuntimeWarning, match='not a valid JSON'): assert key not in get_config(home_dir=tempdir) with pytest.warns(RuntimeWarning, match='non-standard'): pytest.raises(RuntimeError, set_config, key, 'true', home_dir=tempdir) @testing.requires_testing_data def test_show_fiff(): """Test show_fiff.""" # this is not exhaustive, but hopefully bugs will be found in use info = show_fiff(fname_evoked) keys = ['FIFF_EPOCH', 'FIFFB_HPI_COIL', 'FIFFB_PROJ_ITEM', 'FIFFB_PROCESSED_DATA', 'FIFFB_EVOKED', 'FIFF_NAVE', 'FIFF_EPOCH'] assert (all(key in info for key in keys)) info = show_fiff(fname_raw, read_limit=1024) assert ('COORD_TRANS' in show_fiff(fname_fsaverage_trans)) @deprecated('message') def deprecated_func(): pass @deprecated('message') class deprecated_class(object): def __init__(self): pass def test_deprecated(): """Test deprecated function.""" pytest.deprecated_call(deprecated_func) pytest.deprecated_call(deprecated_class) def test_how_to_deal_with_warnings(): """Test filter some messages out of warning records.""" with pytest.warns(UserWarning, match='bb') as w: warnings.warn("aa warning", UserWarning) warnings.warn("bb warning", UserWarning) warnings.warn("bb warning", RuntimeWarning) warnings.warn("aa warning", UserWarning) filter_out_warnings(w, category=UserWarning, match='aa') filter_out_warnings(w, category=RuntimeWarning) assert len(w) == 1 def _test_fetch(url): """Test URL retrieval.""" tempdir = _TempDir() with ArgvSetter(disable_stderr=False): # to capture stdout archive_name = op.join(tempdir, "download_test") _fetch_file(url, archive_name, timeout=30., verbose=False, resume=False) pytest.raises(Exception, _fetch_file, 'NOT_AN_ADDRESS', op.join(tempdir, 'test'), verbose=False) resume_name = op.join(tempdir, "download_resume") # touch file with open(resume_name + '.part', 'w'): os.utime(resume_name + '.part', None) _fetch_file(url, resume_name, resume=True, timeout=30., verbose=False) pytest.raises(ValueError, _fetch_file, url, archive_name, hash_='a', verbose=False) pytest.raises(RuntimeError, _fetch_file, url, archive_name, hash_='a' * 32, verbose=False) @requires_good_network def test_fetch_file_html(): """Test file downloading over http.""" _test_fetch('http://google.com') def test_sum_squared(): """Test optimized sum of squares.""" X = np.random.RandomState(0).randint(0, 50, (3, 3)) assert_equal(np.sum(X ** 2), sum_squared(X)) def test_sizeof_fmt(): """Test sizeof_fmt.""" assert_equal(sizeof_fmt(0), '0 bytes') assert_equal(sizeof_fmt(1), '1 byte') assert_equal(sizeof_fmt(1000), '1000 bytes') def test_url_to_local_path(): """Test URL to local path.""" assert_equal(_url_to_local_path('http://google.com/home/why.html', '.'), op.join('.', 'home', 'why.html')) def test_check_type_picks(): """Test checking type integrity checks of picks.""" picks = np.arange(12) assert_array_equal(picks, _check_type_picks(picks)) picks = list(range(12)) assert_array_equal(np.array(picks), _check_type_picks(picks)) picks = None assert_array_equal(None, _check_type_picks(picks)) picks = ['a', 'b'] pytest.raises(TypeError, _check_type_picks, picks) picks = 'b' pytest.raises(TypeError, _check_type_picks, picks) def test_compute_corr(): """Test Anscombe's Quartett.""" x = np.array([10, 8, 13, 9, 11, 14, 6, 4, 12, 7, 5]) y = np.array([[8.04, 6.95, 7.58, 8.81, 8.33, 9.96, 7.24, 4.26, 10.84, 4.82, 5.68], [9.14, 8.14, 8.74, 8.77, 9.26, 8.10, 6.13, 3.10, 9.13, 7.26, 4.74], [7.46, 6.77, 12.74, 7.11, 7.81, 8.84, 6.08, 5.39, 8.15, 6.42, 5.73], [8, 8, 8, 8, 8, 8, 8, 19, 8, 8, 8], [6.58, 5.76, 7.71, 8.84, 8.47, 7.04, 5.25, 12.50, 5.56, 7.91, 6.89]]) r = compute_corr(x, y.T) r2 = np.array([np.corrcoef(x, y[i])[0, 1] for i in range(len(y))]) assert_allclose(r, r2) pytest.raises(ValueError, compute_corr, [1, 2], []) def test_create_slices(): """Test checking the create of time create_slices.""" # Test that create_slices default provide an empty list assert (create_slices(0, 0) == []) # Test that create_slice return correct number of slices assert (len(create_slices(0, 100)) == 100) # Test with non-zero start parameters assert (len(create_slices(50, 100)) == 50) # Test slices' length with non-zero start and window_width=2 assert (len(create_slices(0, 100, length=2)) == 50) # Test slices' length with manual slice separation assert (len(create_slices(0, 100, step=10)) == 10) # Test slices' within length for non-consecutive samples assert (len(create_slices(0, 500, length=50, step=10)) == 46) # Test that slices elements start, stop and step correctly slices = create_slices(0, 10) assert (slices[0].start == 0) assert (slices[0].step == 1) assert (slices[0].stop == 1) assert (slices[-1].stop == 10) # Same with larger window width slices = create_slices(0, 9, length=3) assert (slices[0].start == 0) assert (slices[0].step == 1) assert (slices[0].stop == 3) assert (slices[-1].stop == 9) # Same with manual slices' separation slices = create_slices(0, 9, length=3, step=1) assert (len(slices) == 7) assert (slices[0].step == 1) assert (slices[0].stop == 3) assert (slices[-1].start == 6) assert (slices[-1].stop == 9) def test_time_mask(): """Test safe time masking.""" N = 10 x = np.arange(N).astype(float) assert_equal(_time_mask(x, 0, N - 1).sum(), N) assert_equal(_time_mask(x - 1e-10, 0, N - 1, sfreq=1000.).sum(), N) assert_equal(_time_mask(x - 1e-10, None, N - 1, sfreq=1000.).sum(), N) assert_equal(_time_mask(x - 1e-10, None, None, sfreq=1000.).sum(), N) assert_equal(_time_mask(x - 1e-10, -np.inf, None, sfreq=1000.).sum(), N) assert_equal(_time_mask(x - 1e-10, None, np.inf, sfreq=1000.).sum(), N) # non-uniformly spaced inputs x = np.array([4, 10]) assert_equal(_time_mask(x[:1], tmin=10, sfreq=1, raise_error=False).sum(), 0) assert_equal(_time_mask(x[:1], tmin=11, tmax=12, sfreq=1, raise_error=False).sum(), 0) assert_equal(_time_mask(x, tmin=10, sfreq=1).sum(), 1) assert_equal(_time_mask(x, tmin=6, sfreq=1).sum(), 1) assert_equal(_time_mask(x, tmin=5, sfreq=1).sum(), 1) assert_equal(_time_mask(x, tmin=4.5001, sfreq=1).sum(), 1) assert_equal(_time_mask(x, tmin=4.4999, sfreq=1).sum(), 2) assert_equal(_time_mask(x, tmin=4, sfreq=1).sum(), 2) # degenerate cases pytest.raises(ValueError, _time_mask, x[:1], tmin=11, tmax=12) pytest.raises(ValueError, _time_mask, x[:1], tmin=10, sfreq=1) def test_random_permutation(): """Test random permutation function.""" n_samples = 10 random_state = 42 python_randperm = random_permutation(n_samples, random_state) # matlab output when we execute rng(42), randperm(10) matlab_randperm = np.array([7, 6, 5, 1, 4, 9, 10, 3, 8, 2]) assert_array_equal(python_randperm, matlab_randperm - 1) def test_copy_doc(): """Test decorator for copying docstrings.""" class A: def m1(): """Docstring for m1.""" pass class B: def m1(): pass class C (A): @copy_doc(A.m1) def m1(): pass assert_equal(C.m1.__doc__, 'Docstring for m1.') pytest.raises(ValueError, copy_doc(B.m1), C.m1) def test_copy_function_doc_to_method_doc(): """Test decorator for re-using function docstring as method docstrings.""" def f1(object, a, b, c): """Docstring for f1. Parameters ---------- object : object Some object. This description also has blank lines in it. a : int Parameter a b : int Parameter b """ pass def f2(object): """Docstring for f2. Parameters ---------- object : object Only one parameter Returns ------- nothing. """ pass def f3(object): """Docstring for f3. Parameters ---------- object : object Only one parameter """ pass def f4(object): """Docstring for f4.""" pass def f5(object): # noqa: D410, D411, D414 """Docstring for f5. Parameters ---------- Returns ------- nothing. """ pass class A: @copy_function_doc_to_method_doc(f1) def method_f1(self, a, b, c): pass @copy_function_doc_to_method_doc(f2) def method_f2(self): "method_f3 own docstring" pass @copy_function_doc_to_method_doc(f3) def method_f3(self): pass assert_equal( A.method_f1.__doc__, """Docstring for f1. Parameters ---------- a : int Parameter a b : int Parameter b """ ) assert_equal( A.method_f2.__doc__, """Docstring for f2. Returns ------- nothing. method_f3 own docstring""" ) assert_equal(A.method_f3.__doc__, 'Docstring for f3.\n\n ') pytest.raises(ValueError, copy_function_doc_to_method_doc(f4), A.method_f1) pytest.raises(ValueError, copy_function_doc_to_method_doc(f5), A.method_f1) def test_progressbar(): """Test progressbar class.""" a = np.arange(10) pbar = ProgressBar(a) assert_equal(a, pbar.iterable) assert_equal(10, pbar.max_value) pbar = ProgressBar(10) assert_equal(10, pbar.max_value) assert (pbar.iterable is None) # Make sure that non-iterable input raises an error def iter_func(a): for ii in a: pass pytest.raises(ValueError, iter_func, ProgressBar(20)) def myfun(x): """Check url.""" assert 'martinos' in x def _identity(x): return x def test_progressbar_parallel_basic(capsys): """Test ProgressBar with parallel computing, basic version.""" assert capsys.readouterr().out == '' parallel, p_fun, _ = parallel_func(_identity, total=10, n_jobs=1, verbose=True) out = parallel(p_fun(x) for x in range(10)) assert out == list(range(10)) assert '100.00%' in capsys.readouterr().out def _identity_block(x, pb): for ii in range(len(x)): pb.update(ii + 1) return x def test_progressbar_parallel_advanced(capsys): """Test ProgressBar with parallel computing, advanced version.""" assert capsys.readouterr().out == '' # This must be "1" because "capsys" won't get stdout properly otherwise parallel, p_fun, _ = parallel_func(_identity_block, n_jobs=1, verbose=False) arr = np.arange(10) with ProgressBar(len(arr), verbose_bool=True) as pb: out = parallel(p_fun(x, pb.subset(pb_idx)) for pb_idx, x in array_split_idx(arr, 2)) assert op.isfile(pb._mmap_fname) sum_ = np.memmap(pb._mmap_fname, dtype='bool', mode='r', shape=10).sum() assert sum_ == len(arr) assert not op.isfile(pb._mmap_fname), '__exit__ not called?' out = np.concatenate(out) assert_array_equal(out, arr) assert '100.00%' in capsys.readouterr().out def _identity_block_wide(x, pb): for ii in range(len(x)): for jj in range(2): pb.update(ii * 2 + jj + 1) return x, pb.idx def test_progressbar_parallel_more(capsys): """Test ProgressBar with parallel computing, advanced version.""" assert capsys.readouterr().out == '' # This must be "1" because "capsys" won't get stdout properly otherwise parallel, p_fun, _ = parallel_func(_identity_block_wide, n_jobs=1, verbose=False) arr = np.arange(10) with ProgressBar(len(arr) * 2, verbose_bool=True) as pb: out = parallel(p_fun(x, pb.subset(pb_idx)) for pb_idx, x in array_split_idx(arr, 2, n_per_split=2)) idxs = np.concatenate([o[1] for o in out]) assert_array_equal(idxs, np.arange(len(arr) * 2)) out = np.concatenate([o[0] for o in out]) assert op.isfile(pb._mmap_fname) sum_ = np.memmap(pb._mmap_fname, dtype='bool', mode='r', shape=len(arr) * 2).sum() assert sum_ == len(arr) * 2 assert not op.isfile(pb._mmap_fname), '__exit__ not called?' assert '100.00%' in capsys.readouterr().out def test_open_docs(): """Test doc launching.""" old_tab = webbrowser.open_new_tab try: # monkey patch temporarily to prevent tabs from actually spawning webbrowser.open_new_tab = myfun open_docs() open_docs('tutorials', 'dev') open_docs('examples', 'stable') pytest.raises(ValueError, open_docs, 'foo') pytest.raises(ValueError, open_docs, 'api', 'foo') finally: webbrowser.open_new_tab = old_tab def test_linkcode_resolve(): """Test linkcode resolving.""" ex = '' if sys.version[0] == '2' else '#L' url = linkcode_resolve('py', dict(module='mne', fullname='Epochs')) assert '/mne/epochs.py' + ex in url url = linkcode_resolve('py', dict(module='mne', fullname='compute_covariance')) assert '/mne/cov.py' + ex in url url = linkcode_resolve('py', dict(module='mne', fullname='convert_forward_solution')) assert '/mne/forward/forward.py' + ex in url url = linkcode_resolve('py', dict(module='mne', fullname='datasets.sample.data_path')) assert '/mne/datasets/sample/sample.py' + ex in url def test_warn(capsys): """Test the smart warn() function.""" with pytest.warns(RuntimeWarning, match='foo'): warn('foo') captured = capsys.readouterr() assert captured.out == '' # gh-5592 assert captured.err == '' # this is because pytest.warns took it already def test_reg_pinv(): """Test regularization and inversion of covariance matrix.""" # create rank-deficient array a = np.array([[1., 0., 1.], [0., 1., 0.], [1., 0., 1.]]) # Test if rank-deficient matrix without regularization throws # specific warning with pytest.warns(RuntimeWarning, match='deficient'): _reg_pinv(a, reg=0.) # Test inversion with explicit rank a_inv_np = np.linalg.pinv(a) a_inv_mne, loading_factor, rank = _reg_pinv(a, rank=2) assert loading_factor == 0 assert rank == 2 assert_array_equal(a_inv_np, a_inv_mne) # Test inversion with automatic rank detection a_inv_mne, _, estimated_rank = _reg_pinv(a, rank=None) assert_array_equal(a_inv_np, a_inv_mne) assert estimated_rank == 2 # Test adding regularization a_inv_mne, loading_factor, estimated_rank = _reg_pinv(a, reg=2) # Since A has a diagonal of all ones, loading_factor should equal the # regularization parameter assert loading_factor == 2 # The estimated rank should be that of the non-regularized matrix assert estimated_rank == 2 # Test result against the NumPy version a_inv_np = np.linalg.pinv(a + loading_factor * np.eye(3)) assert_array_equal(a_inv_np, a_inv_mne) # Test setting rcond a_inv_np = np.linalg.pinv(a, rcond=0.5) a_inv_mne, _, estimated_rank = _reg_pinv(a, rcond=0.5) assert_array_equal(a_inv_np, a_inv_mne) assert estimated_rank == 1 # Test inverting an all zero cov a_inv, loading_factor, estimated_rank = _reg_pinv(np.zeros((3, 3)), reg=2) assert_array_equal(a_inv, 0) assert loading_factor == 0 assert estimated_rank == 0 run_tests_if_main()