FM / test

/usr/local/lib/python3.6/site-packages/joblib/test/

/usr/local/lib/python3.6/site-packages/joblib/test

Name	Size	Mode	Actions
data/	-	0755	rm
__pycache__/	-	0755	rm
common.py	3283	0644	edit dl rm
testutils.py	251	0644	edit dl rm
test_backports.py	1175	0644	edit dl rm
test_cloudpickle_wrapper.py	749	0644	edit dl rm
test_dask.py	17053	0644	edit dl rm
test_deprecated_objects.py	1249	0644	edit dl rm
test_disk.py	2205	0644	edit dl rm
test_format_stack.py	4250	0644	edit dl rm
test_func_inspect.py	8935	0644	edit dl rm
test_func_inspect_special_encoding.py	146	0644	edit dl rm
test_hashing.py	16078	0644	edit dl rm
test_init.py	422	0644	edit dl rm
test_logger.py	985	0644	edit dl rm
test_memmapping.py	42154	0644	edit dl rm
test_memory.py	44810	0644	edit dl rm
test_missing_multiprocessing.py	1123	0644	edit dl rm
test_module.py	1832	0644	edit dl rm
test_my_exceptions.py	2066	0644	edit dl rm
test_numpy_pickle.py	38700	0644	edit dl rm
test_numpy_pickle_compat.py	624	0644	edit dl rm
test_numpy_pickle_utils.py	421	0644	edit dl rm
test_parallel.py	59653	0644	edit dl rm
test_store_backends.py	1942	0644	edit dl rm
test_testing.py	2467	0644	edit dl rm
test_utils.py	584	0644	edit dl rm
__init__.py	73	0644	edit dl rm

Edit: /usr/local/lib/python3.6/site-packages/joblib/test/test_store_backends.py (1942B)

from joblib.testing import parametrize, timeout
from joblib.test.common import with_multiprocessing
from joblib.backports import concurrency_safe_rename
from joblib import Parallel, delayed
from joblib._store_backends import concurrency_safe_write

def write_func(output, filename):
    with open(filename, 'wb') as f:
        cpickle.dump(output, f)

def load_func(expected, filename):
    for i in range(10):
        try:
            with open(filename, 'rb') as f:
                reloaded = cpickle.load(f)
            break
        except (OSError, IOError):
            # On Windows you can have WindowsError ([Error 5] Access
            # is denied or [Error 13] Permission denied) when reading the file,
            # probably because a writer process has a lock on the file
            time.sleep(0.1)
    else:
        raise
    assert expected == reloaded

def concurrency_safe_write_rename(to_write, filename, write_func):
    temporary_filename = concurrency_safe_write(to_write,
                                                filename, write_func)
    concurrency_safe_rename(temporary_filename, filename)

@timeout(0)  # No timeout as this test can be long
@with_multiprocessing
@parametrize('backend', ['multiprocessing', 'loky', 'threading'])
def test_concurrency_safe_write(tmpdir, backend):
    # Add one item to cache
    filename = tmpdir.join('test.pkl').strpath

obj = {str(i): i for i in range(int(1e5))}
    funcs = [functools.partial(concurrency_safe_write_rename,
                               write_func=write_func)
             if i % 3 != 2 else load_func for i in range(12)]
    Parallel(n_jobs=2, backend=backend)(
        delayed(func)(obj, filename) for func in funcs)

try: # Python 2.7: use the C pickle to speed up # test_concurrency_safe_write which pickles big python objects import cPickle as cpickle except ImportError: import pickle as cpickle import functools import time from joblib.testing import parametrize, timeout from joblib.test.common import with_multiprocessing from joblib.backports import concurrency_safe_rename from joblib import Parallel, delayed from joblib._store_backends import concurrency_safe_write def write_func(output, filename): with open(filename, 'wb') as f: cpickle.dump(output, f) def load_func(expected, filename): for i in range(10): try: with open(filename, 'rb') as f: reloaded = cpickle.load(f) break except (OSError, IOError): # On Windows you can have WindowsError ([Error 5] Access # is denied or [Error 13] Permission denied) when reading the file, # probably because a writer process has a lock on the file time.sleep(0.1) else: raise assert expected == reloaded def concurrency_safe_write_rename(to_write, filename, write_func): temporary_filename = concurrency_safe_write(to_write, filename, write_func) concurrency_safe_rename(temporary_filename, filename) @timeout(0) # No timeout as this test can be long @with_multiprocessing @parametrize('backend', ['multiprocessing', 'loky', 'threading']) def test_concurrency_safe_write(tmpdir, backend): # Add one item to cache filename = tmpdir.join('test.pkl').strpath obj = {str(i): i for i in range(int(1e5))} funcs = [functools.partial(concurrency_safe_write_rename, write_func=write_func) if i % 3 != 2 else load_func for i in range(12)] Parallel(n_jobs=2, backend=backend)( delayed(func)(obj, filename) for func in funcs)