pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use FormCache everywhere

Browse Source
This commit is contained in:
richard-dds 2018-11-16 11:42:33 -05:00
parent 6177f46587
commit 5447be8b52
2 changed files with 14 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
atst/utils/form_cache.py
View File

@ -6,18 +6,6 @@ import pickle
DEFAULT_CACHE_NAME = "formcache"
def cache_form_data(redis, formdata, expiry_seconds=3600, key_prefix=DEFAULT_CACHE_NAME):
value = pickle.dumps(formdata)
key = hashlib.sha1(os.urandom(64)).hexdigest()
redis.setex(name="{}:{}".format(key_prefix, key), value=value, time=expiry_seconds)
return key
def retrieve_form_data(redis, formdata_key, key_prefix="formcache"):
data = redis.get("{}:{}".format(key_prefix, formdata_key))
return pickle.loads(data)
class FormCache(object):
def __init__(self, redis):
@ -28,13 +16,13 @@ class FormCache(object):
if cache_key:
return self.read(cache_key)
def write(self, formdata, expiry_seconds=3600, key_prefix="formcache"):
def write(self, formdata, expiry_seconds=3600, key_prefix=DEFAULT_CACHE_NAME):
value = pickle.dumps(formdata)
hash_ = hashlib.sha1(os.urandom(64)).hexdigest()
self.redis.setex(name=self._key(key_prefix, hash_), value=value, time=expiry_seconds)
return hash_
def read(self, formdata_key, key_prefix="formcache"):
def read(self, formdata_key, key_prefix=DEFAULT_CACHE_NAME):
data = self.redis.get(self._key(key_prefix, formdata_key))
return pickle.loads(data) if data is not None else {}

18
tests/utils/test_form_cache.py
View File

@ -1,15 +1,21 @@
import pytest
from werkzeug.datastructures import ImmutableDict
from atst.utils.form_cache import DEFAULT_CACHE_NAME, cache_form_data, retrieve_form_data
from atst.utils.form_cache import DEFAULT_CACHE_NAME, FormCache
def test_cache_form_data(app):
@pytest.fixture
def form_cache(app):
return FormCache(app.redis)
def test_cache_form_data(app, form_cache):
data = ImmutableDict({"kessel_run": "12 parsecs"})
key = cache_form_data(app.redis, data)
key = form_cache.write(data)
assert app.redis.get("{}:{}".format(DEFAULT_CACHE_NAME, key))
def test_retrieve_form_data(app):
def test_retrieve_form_data(form_cache):
data = ImmutableDict({"class": "corellian"})
key = cache_form_data(app.redis, data)
retrieved = retrieve_form_data(app.redis, key)
key = form_cache.write(data)
retrieved = form_cache.read(key)
assert retrieved == data