Test Parallel wrapping.
(n_jobs)
| 24 | ], |
| 25 | ) |
| 26 | def test_parallel_func(n_jobs): |
| 27 | """Test Parallel wrapping.""" |
| 28 | joblib = pytest.importorskip("joblib") |
| 29 | if os.getenv("MNE_FORCE_SERIAL", "").lower() in ("true", "1"): |
| 30 | pytest.skip("MNE_FORCE_SERIAL is set") |
| 31 | |
| 32 | def fun(x): |
| 33 | return x * 2 |
| 34 | |
| 35 | if isinstance(n_jobs, str): |
| 36 | backend, n_jobs = n_jobs.split() |
| 37 | n_jobs = want_jobs = int(n_jobs) |
| 38 | try: |
| 39 | func = joblib.parallel_config |
| 40 | except AttributeError: |
| 41 | # joblib < 1.3 |
| 42 | func = joblib.parallel_backend |
| 43 | ctx = func(backend, n_jobs=n_jobs) |
| 44 | n_jobs = None |
| 45 | else: |
| 46 | ctx = nullcontext() |
| 47 | if n_jobs is not None and n_jobs < 0: |
| 48 | want_jobs = multiprocessing.cpu_count() + 1 + n_jobs |
| 49 | else: |
| 50 | want_jobs = 1 |
| 51 | with ctx: |
| 52 | parallel, p_fun, got_jobs = parallel_func(fun, n_jobs, verbose="debug") |
| 53 | assert got_jobs == want_jobs |
| 54 | |
| 55 | |
| 56 | def test_parallel_func_n_jobs_none(): |
nothing calls this directly
no test coverage detected