Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
res1 = res_queue.get()
assert isinstance(res1, float)
res2 = res_queue.get()
assert (res2 is None) or isinstance(res2, KeyError)
@cachier()
def _delete_cache(arg_1, arg_2):
"""Some function."""
sleep(1)
return random() + arg_1 + arg_2
# _DEL_CACHE_FNAME = '.__main__._delete_cache'
_DEL_CACHE_FNAME = '.tests.test_pickle_core._delete_cache'
_DEL_CACHE_FPATH = os.path.join(EXPANDED_CACHIER_DIR, _DEL_CACHE_FNAME)
def _calls_delete_cache(res_queue, del_cache):
try:
# print('in')
res = _delete_cache(0.13, 0.02)
# print('out with {}'.format(res))
if del_cache:
# print('deleteing!')
os.remove(_DEL_CACHE_FPATH)
# print(os.path.isfile(_DEL_CACHE_FPATH))
res_queue.put(res)
except Exception as exc:
# print('found')
res_queue.put(exc)
for sleeptime in sleeptimes:
if _helper_bad_cache_file(sleeptime):
return
assert False
@cachier()
def _delete_cache(arg_1, arg_2):
"""Some function."""
sleep(1)
return random() + arg_1 + arg_2
# _DEL_CACHE_FNAME = '.__main__._delete_cache'
_DEL_CACHE_FNAME = '.tests.test_pickle_core._delete_cache'
_DEL_CACHE_FPATH = os.path.join(EXPANDED_CACHIER_DIR, _DEL_CACHE_FNAME)
def _calls_delete_cache(res_queue, del_cache):
try:
# print('in')
res = _delete_cache(0.13, 0.02)
# print('out with {}'.format(res))
if del_cache:
# print('deleteing!')
os.remove(_DEL_CACHE_FPATH)
# print(os.path.isfile(_DEL_CACHE_FPATH))
res_queue.put(res)
except Exception as exc:
# print('found')
res_queue.put(exc)
assert res_queue.qsize() == 2
res1 = res_queue.get()
res2 = res_queue.get()
assert res1 == res2
@cachier()
def _bad_cache(arg_1, arg_2):
"""Some function."""
sleep(1)
return random() + arg_1 + arg_2
# _BAD_CACHE_FNAME = '.__main__._bad_cache'
_BAD_CACHE_FNAME = '.tests.test_pickle_core._bad_cache'
_BAD_CACHE_FPATH = os.path.join(EXPANDED_CACHIER_DIR, _BAD_CACHE_FNAME)
def _calls_bad_cache(res_queue, trash_cache):
try:
res = _bad_cache(0.13, 0.02)
if trash_cache:
with open(_BAD_CACHE_FPATH, 'w') as cache_file:
cache_file.seek(0)
cache_file.truncate()
res_queue.put(res)
except Exception as exc:
res_queue.put(exc)
def _helper_bad_cache_file(sleeptime):
"""Test pickle core handling of bad cache files."""
assert res_queue.qsize() == 2
res1 = res_queue.get()
res2 = res_queue.get()
assert res1 == res2
@cachier()
def _bad_cache(arg_1, arg_2):
"""Some function."""
sleep(1)
return random() + arg_1 + arg_2
# _BAD_CACHE_FNAME = '.__main__._bad_cache'
_BAD_CACHE_FNAME = '.tests.test_pickle_core._bad_cache'
_BAD_CACHE_FPATH = os.path.join(EXPANDED_CACHIER_DIR, _BAD_CACHE_FNAME)
def _calls_bad_cache(res_queue, trash_cache):
try:
res = _bad_cache(0.13, 0.02)
if trash_cache:
with open(_BAD_CACHE_FPATH, 'w') as cache_file:
cache_file.seek(0)
cache_file.truncate()
res_queue.put(res)
except Exception as exc:
res_queue.put(exc)
def test_bad_cache_file():
"""Test pickle core handling of bad cache files."""