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_backports.py (1175B)

import mmap

from joblib.backports import make_memmap, concurrency_safe_rename
from joblib.test.common import with_numpy
from joblib.testing import parametrize
from joblib import Parallel, delayed

@with_numpy
def test_memmap(tmpdir):
    fname = tmpdir.join('test.mmap').strpath
    size = 5 * mmap.ALLOCATIONGRANULARITY
    offset = mmap.ALLOCATIONGRANULARITY + 1
    memmap_obj = make_memmap(fname, shape=size, mode='w+', offset=offset)
    assert memmap_obj.offset == offset

@parametrize('dst_content', [None, 'dst content'])
@parametrize('backend', [None, 'threading'])
def test_concurrency_safe_rename(tmpdir, dst_content, backend):
    src_paths = [tmpdir.join('src_%d' % i) for i in range(4)]
    for src_path in src_paths:
        src_path.write('src content')
    dst_path = tmpdir.join('dst')
    if dst_content is not None:
        dst_path.write(dst_content)

Parallel(n_jobs=4, backend=backend)(
        delayed(concurrency_safe_rename)(src_path.strpath, dst_path.strpath)
        for src_path in src_paths
    )
    assert dst_path.exists()
    assert dst_path.read() == 'src content'
    for src_path in src_paths:
        assert not src_path.exists()

import mmap from joblib.backports import make_memmap, concurrency_safe_rename from joblib.test.common import with_numpy from joblib.testing import parametrize from joblib import Parallel, delayed @with_numpy def test_memmap(tmpdir): fname = tmpdir.join('test.mmap').strpath size = 5 * mmap.ALLOCATIONGRANULARITY offset = mmap.ALLOCATIONGRANULARITY + 1 memmap_obj = make_memmap(fname, shape=size, mode='w+', offset=offset) assert memmap_obj.offset == offset @parametrize('dst_content', [None, 'dst content']) @parametrize('backend', [None, 'threading']) def test_concurrency_safe_rename(tmpdir, dst_content, backend): src_paths = [tmpdir.join('src_%d' % i) for i in range(4)] for src_path in src_paths: src_path.write('src content') dst_path = tmpdir.join('dst') if dst_content is not None: dst_path.write(dst_content) Parallel(n_jobs=4, backend=backend)( delayed(concurrency_safe_rename)(src_path.strpath, dst_path.strpath) for src_path in src_paths ) assert dst_path.exists() assert dst_path.read() == 'src content' for src_path in src_paths: assert not src_path.exists()