source/projects/async_cache/test/test_ttl.py

57 lines
1.4 KiB
Python

import asyncio
import time
from async_cache import ATTL
@ATTL(ttl=60)
async def long_expiration_fn(wait: int):
await asyncio.sleep(wait)
return wait
@ATTL(ttl=5)
async def short_expiration_fn(wait: int):
await asyncio.sleep(wait)
return wait
@ATTL(ttl=3)
async def short_cleanup_fn(wait: int):
await asyncio.sleep(wait)
return wait
def test_cache_hit():
t1 = time.time()
asyncio.get_event_loop().run_until_complete(long_expiration_fn(4))
t2 = time.time()
asyncio.get_event_loop().run_until_complete(long_expiration_fn(4))
t3 = time.time()
t_first_exec = (t2 - t1) * 1000
t_second_exec = (t3 - t2) * 1000
print(t_first_exec)
print(t_second_exec)
assert t_first_exec > 4000
assert t_second_exec < 4000
def test_cache_expiration():
t1 = time.time()
asyncio.get_event_loop().run_until_complete(short_expiration_fn(1))
t2 = time.time()
asyncio.get_event_loop().run_until_complete(short_expiration_fn(1))
t3 = time.time()
time.sleep(5)
t4 = time.time()
asyncio.get_event_loop().run_until_complete(short_expiration_fn(1))
t5 = time.time()
t_first_exec = (t2 - t1) * 1000
t_second_exec = (t3 - t2) * 1000
t_third_exec = (t5 - t4) * 1000
print(t_first_exec)
print(t_second_exec)
print(t_third_exec)
assert t_first_exec > 1000
assert t_second_exec < 1000
assert t_third_exec > 1000