Add unittest

Author: ptran1203

.gitignore

@@ -1,5 +1,4 @@
 __pycache__
 *.pyc
-test*.py
 dist
 *.egg-info

README.md

@@ -1,6 +1,6 @@
 # Redis cache for Python
 
-- Simple python redis cache library, mostly used as distributed caching, where application servers is in separated processes such as Gunicorn workers, K8s replicas.
+- Simple python redis cache library, mostly used as [distributed caching](https://redis.com/glossary/distributed-caching), where application servers is in separated processes such as Gunicorn workers, K8s replicas, .etc..
 - **Asyncio Support for FastAPI, Starlette**
 
 ## Requirements
@@ -45,4 +45,53 @@ print(result)
 # [4, 10, 18]
 ```
 
+**Asynchronous with asyncio**
+
+```python
+import asyncio
+from py_redis_cache.asyncio import AsyncRedisCache
+
+# init redis_cache instance and connection
+redis_cache = AsyncRedisCache(
+    host="127.0.0.1",
+    port=6379,
+    verbose=1  # Turn on logging for demonstration, set to 0 for silent caching
+)
+
+@redis_cache.aio_cache(ttl=10)  # Expire after 10 seconds
+async def heavy_compute(a: list, b: list):
+    length = max(len(a), len(b))
+    c = [[]] * length
+    for i in range(length):
+        c[i] = a[i] * b[i]
+    return c
+
+async def test_async_cache():
+    result = await heavy_compute([1, 2, 3], [4, 5, 6])
+    print(result)
+
+    # Now the result is cached
+    result2 = await heavy_compute([1, 2, 3], [4, 5, 6])
+
+    print(result2)
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(test_async_cache())
+
+# Cache added, key=redis_cache::11=__main__.heavy_compute(a=[1, 2, 3].b=[4, 5, 6]), size=0.000Kb
+# [4, 10, 18]
+# Cache hit, key=redis_cache::11=__main__.heavy_compute(a=[1, 2, 3].b=[4, 5, 6])
+# [4, 10, 18]
+```
+
+## Advanced usage
+
 for further examples and use cases please visit [examples](examples)
+
+## Testing
+
+**NOTE**: Please make sure you have redis running on `127.0.0.1:6379` to run test.
+
+```bash
+$ python3 -m unittest discover tests
+```

examples/simple/async.py

@@ -0,0 +1,33 @@
+# Example of caching async functions
+
+import sys
+sys.path.append(".")
+import asyncio
+from py_redis_cache.asyncio import AsyncRedisCache
+
+# init redis_cache instance and connection
+redis_cache = AsyncRedisCache(
+    host="127.0.0.1",
+    port=6379,
+    verbose=1  # Turn on logging for demonstration, set to 0 for silent caching
+)
+
+@redis_cache.aio_cache(ttl=10)  # Expire after 10 seconds
+async def heavy_compute(a: list, b: list):
+    length = max(len(a), len(b))
+    c = [[]] * length
+    for i in range(length):
+        c[i] = a[i] * b[i]
+    return c
+
+async def test_async_cache():
+    result = await heavy_compute([1, 2, 3], [4, 5, 6])
+    print(result)
+
+    # Now the result is cached
+    result2 = await heavy_compute([1, 2, 3], [4, 5, 6])
+
+    print(result2)
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(test_async_cache())

examples/basic.py renamed to examples/simple/basic.py

@@ -17,6 +17,7 @@
 current_aio_redis_cache = None
 
 class Singleton(type):
+    disable = False
     _instances = {}
     def __call__(cls, *args, **kwargs):
         from .asyncio import AsyncRedisCache
@@ -27,11 +28,16 @@ def __call__(cls, *args, **kwargs):
             cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
             if not is_aio_redis:
                 current_redis_cache = cls._instances[cls]
-                print("SET", current_redis_cache)
             else:
                 current_aio_redis_cache = cls._instances[cls]
-                print("SET AIO", current_aio_redis_cache)
+            
+        if cls.disable:
+            return cls._instances.pop(cls)
         return cls._instances[cls]
+    
+    @classmethod
+    def clear_instance(cls):
+        cls._instances = {}
 
 
 class RedisCache(metaclass=Singleton):

py_redis_cache/__init__.py

@@ -0,0 +1,46 @@
+import os
+import time
+import unittest
+from py_redis_cache import RedisCache, Singleton
+
+
+class TestRedisCache(unittest.TestCase):
+
+    def test_simple_cache(self):
+        # Enable Singleton for testing this case
+        redis_cache = RedisCache()
+        redis_cache2 = RedisCache()
+
+        assert redis_cache is redis_cache2  # Sample object
+
+    def test_ttl(self):
+        redis_cache = RedisCache()
+
+        @redis_cache.cache(ttl=1)
+        def add_two(a, b):
+            return a + b + time.time()
+
+        result = add_two(1, 2)
+        result_2 = add_two(1, 2)
+
+        assert result == result_2  # Should cached
+
+        time.sleep(1.1) # Cache will expire after 1 seconds
+
+        assert result != add_two(1, 2)
+
+    def test_kwargs(self):
+        redis_cache = RedisCache()
+
+        @redis_cache.cache(ttl=1)
+        def add_two(a, b):
+            return a + b + time.time()
+        
+        result = add_two(1, 2)
+        result_2 = add_two(1, b=2)
+        result_3 = add_two(a=1, b=2)  
+        assert result == result_2 == result_3
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_cache.py

@@ -0,0 +1,50 @@
+# Test redis cache connection from environment variable or argument passing to
+# RedisCache
+import os
+import unittest
+from py_redis_cache import RedisCache, Singleton
+from redis.exceptions import ConnectionError
+
+
+class TestRedisCacheConnection(unittest.TestCase):
+
+    def test_connect_via_host_port(self):
+        # Disable Singleton for testing
+        Singleton.disable = True
+        redis_cache = RedisCache(host="127.0.0.1", port=6379)
+        assert redis_cache.client.ping() is True
+
+    def test_connect_via_host_port_wrong(self):
+        # Redis does not listening on port 6380
+        Singleton.disable = True
+        redis_cache = RedisCache(
+            host="127.0.0.1",
+            port=99999,
+        )
+        self.assertRaises(
+            ConnectionError,
+            redis_cache.client.ping
+        )
+
+    def test_connect_via_url(self):
+        Singleton.disable = True
+        redis_cache = RedisCache(redis_url="redis://127.0.0.1:6379")
+        assert redis_cache.client.ping() is True
+
+    def test_connect_via_env(self):
+        Singleton.disable = True
+        os.environ["REDIS_URL"] = "redis://127.0.0.1:6379"
+        # Provide incorrect host, port to demostrate passing redis_url via Env possible
+        redis_cache = RedisCache(host="fakehost", port=1)
+        os.environ["REDIS_URL"] = "" # Reset
+        assert redis_cache.client.ping() is True
+
+        os.environ["REDIS_CACHE_URL"] = "redis://127.0.0.1:6379"
+        # Provide incorrect host, port to demostrate passing redis_url via Env possible
+        redis_cache = RedisCache(host="fakehost", port=1)
+        os.environ["REDIS_CACHE_URL"] = "" # Reset
+        assert redis_cache.client.ping() is True
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_connection.py

@@ -0,0 +1,29 @@
+import os
+import unittest
+from py_redis_cache import RedisCache, Singleton
+
+
+class TestRedisCacheConnection(unittest.TestCase):
+
+    def test_two_calls(self):
+        # Enable Singleton for testing this case
+        Singleton.disable = False
+
+        redis_cache = RedisCache()
+        redis_cache2 = RedisCache()
+
+        assert redis_cache is redis_cache2  # Sample object
+
+    def test_current_redis_cache(self):
+        # Enable Singleton for testing this case
+        Singleton.disable = False
+
+        redis_cache = RedisCache()
+        
+        from py_redis_cache import current_redis_cache
+
+        assert redis_cache is current_redis_cache  # Sample object
+
+
+if __name__ == '__main__':
+    unittest.main()