Test redis expire argument

Procfile

worker: python test_redis.py

requirements.txt

https://github.com/gabbhack/aiogram/archive/redis_storage.zip
aioredis

runtime.txt

python-3.7.3

test_redis_expire.py

import os
import asyncio

from aioredis.util import parse_url
from aiogram.contrib.fsm_storage.redis import RedisStorage2, STATE_KEY, STATE_DATA_KEY, STATE_BUCKET_KEY


REDIS_URI = os.getenv("REDIS_URL")
(HOST, PORT), OPTIONS = parse_url(REDIS_URI)
REDIS_SETTINGS = dict(host=HOST, port=PORT, **OPTIONS)
TEST_DATA = {"foo": "bar"}

storage = RedisStorage2(**REDIS_SETTINGS, state_ttl=10, data_ttl=20, bucket_ttl=30)


async def main():
    print("Start")

    redis = await storage.redis()

    # test set_state
    key = storage.generate_key(1, 1, STATE_KEY)
    await storage.set_state(chat=1, user=1, state="test")
    ttl = await redis.ttl(key)
    assert ttl == 10

    # test set_data
    key = storage.generate_key(1, 1, STATE_DATA_KEY)
    await storage.set_data(chat=1, user=1, data=TEST_DATA)
    ttl = await redis.ttl(key)
    assert ttl == 20

    # test update_data
    await storage.set_data(chat=1, user=1, data=TEST_DATA)
    await storage.update_data(chat=1, user=1)
    ttl = await redis.ttl(key)
    assert ttl == 20

    # test set_bucket
    key = storage.generate_key(1, 1, STATE_BUCKET_KEY)
    await storage.set_bucket(chat=1, user=1, bucket=TEST_DATA)
    ttl = await redis.ttl(key)
    assert ttl == 30

    # test update_bucket
    await storage.set_bucket(chat=1, user=1, bucket=TEST_DATA)
    await storage.update_bucket(chat=1, user=1)
    ttl = await redis.ttl(key)
    assert ttl == 30

    await storage.close()
    await storage.wait_closed()

    print("Finish")


loop = asyncio.get_event_loop().run_until_complete(main())

aiogram_ru/67491