Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_disk_valueerror():
with pytest.raises(ValueError):
with dc.Cache(disk=dc.Disk('test')):
pass
assert cache1[count] == str(count)
for count in range(100, 200):
assert cache1[count] == str(count) * int(1e5)
for count in range(200, 300):
assert cache1[count] == str(count)
for count in range(300, 400):
assert cache1[count] == str(count) * int(1e5)
shutil.rmtree(cache_dir1, ignore_errors=True)
shutil.rmtree(cache_dir2, ignore_errors=True)
class SHA256FilenameDisk(dc.Disk):
def filename(self, key=dc.UNKNOWN, value=dc.UNKNOWN):
filename = hashlib.sha256(key).hexdigest()[:32]
full_path = op.join(self._directory, filename)
return filename, full_path
def test_custom_filename_disk():
with dc.FanoutCache(disk=SHA256FilenameDisk) as cache:
for count in range(100, 200):
key = str(count).encode('ascii')
cache[key] = str(count) * int(1e5)
disk = SHA256FilenameDisk(cache.directory)
for count in range(100, 200):
key = str(count).encode('ascii')
def test_custom_disk():
with dc.Cache(disk=dc.JSONDisk, disk_compress_level=6) as cache:
values = [None, True, 0, 1.23, {}, [None] * 10000]
for value in values:
cache[value] = value
for value in values:
assert cache[value] == value
shutil.rmtree(cache.directory, ignore_errors=True)
class SHA256FilenameDisk(diskcache.Disk):
def filename(self, key=dc.UNKNOWN, value=dc.UNKNOWN):
filename = hashlib.sha256(key).hexdigest()[:32]
full_path = op.join(self._directory, filename)
return filename, full_path
def test_custom_filename_disk():
with dc.Cache(disk=SHA256FilenameDisk) as cache:
for count in range(100, 200):
key = str(count).encode('ascii')
cache[key] = str(count) * int(1e5)
for count in range(100, 200):
key = str(count).encode('ascii')
filename = hashlib.sha256(key).hexdigest()[:32]
full_path = op.join(cache.directory, filename)
import gzip
from diskcache import FanoutCache, Disk
from diskcache.core import BytesType, MODE_BINARY, BytesIO
from util.logconf import logging
log = logging.getLogger(__name__)
# log.setLevel(logging.WARN)
log.setLevel(logging.INFO)
# log.setLevel(logging.DEBUG)
class GzipDisk(Disk):
def store(self, value, read, key=None):
"""
Override from base class diskcache.Disk.
Chunking is due to needing to work on pythons < 2.7.13:
- Issue #27130: In the "zlib" module, fix handling of large buffers
(typically 2 or 4 GiB). Previously, inputs were limited to 2 GiB, and
compression and decompression operations did not properly handle results of
2 or 4 GiB.
:param value: value to convert
:param bool read: True when value is file-like object
:return: (size, mode, filename, value) tuple for Cache table
"""
# pylint: disable=unidiomatic-typecheck
if type(value) is BytesType: