forked from pyodide/pyodide
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_joblib.py
42 lines (31 loc) · 1.01 KB
/
test_joblib.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
import pytest
def test_joblib_numpy_pickle(selenium, request):
selenium.load_package(["numpy", "joblib"])
selenium.run(
"""
import joblib
import numpy as np
from numpy.testing import assert_array_equal
X = np.ones(10)
file_path = "./X.pkl"
joblib.dump(X, file_path)
X2 = joblib.load(file_path)
assert_array_equal(X, X2)"""
)
def test_joblib_parallel(selenium):
selenium.load_package(["numpy", "joblib"])
selenium.clean_logs()
selenium.run(
"""
from math import sqrt
from joblib import Parallel, delayed
# check that the default multiprocessing backend
# gracefully reduces to n_jobs=1
res = Parallel(n_jobs=2)(delayed(sqrt)(i ** 2)
for i in range(10))
assert res == [sqrt(i ** 2) for i in range(10)]
# check threading backend
Parallel(n_jobs=2, backend='threading')(
delayed(sqrt)(i ** 2) for i in range(10))
"""
)