diff --git a/gluon/tests/test_redis.py b/gluon/tests/test_redis.py new file mode 100644 index 00000000..41f76441 --- /dev/null +++ b/gluon/tests/test_redis.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +""" Unit tests for redis """ + +import unittest +import os +import time +from datetime import datetime + +from gluon._compat import to_bytes +from gluon.storage import Storage +from gluon.utils import web2py_uuid +from gluon.globals import Request, Response, Session, current +from gluon.contrib.redis_utils import RConn +from gluon.contrib.redis_session import RedisSession +from gluon.contrib.redis_cache import RedisCache + + +def setup_clean_session(): + request = Request(env={}) + request.application = 'a' + request.controller = 'c' + request.function = 'f' + request.folder = 'applications/admin' + response = Response() + session = Session() + session.connect(request, response) + from gluon.globals import current + current.request = request + current.response = response + current.session = session + return current + + +def setUpModule(): + pass + + +def tearDownModule(): + if os.path.isfile('appconfig.json'): + os.unlink('appconfig.json') + + +class TestRedis(unittest.TestCase): + """ Tests the Redis contrib packages """ + + def test_0_redis_session(self): + """ Basic redis read-write """ + current = setup_clean_session() + response = current.response + rconn = RConn(host='redis') + db = RedisSession(redis_conn=rconn, session_expiry=False) + tname = 'testtablename' + Field = db.Field + db.define_table( + tname, + Field('locked', 'boolean', default=False), + Field('client_ip', length=64), + Field('created_datetime', 'datetime', + default=datetime.now), + Field('modified_datetime', 'datetime'), + Field('unique_key', length=64), + Field('session_data', 'blob'), + ) + table = db[tname] + unique_key = web2py_uuid() + dd = dict( + locked=0, + client_ip=response.session_client, + modified_datetime=datetime.now, + unique_key=unique_key + ) + record_id = table.insert(**dd) + data_from_db = db(table.id == record_id).select() + #print('data_from_db=', data_from_db) + self.assertDictEqual(dd, data_from_db) +