Function test_hgetex_expiration_configs

tests/test_asyncio/test_hash.py:337–380

(r)

@skip_if_server_version_lt("7.9.0")
async def test_hgetex_expiration_configs(r):
    await r.delete("test:hash")
    await r.hset(
        "test:hash", "foo", "bar", mapping={"1": 1, "3": "three", "4": b"four"}
    )

    test_keys = ["foo", "1", "4"]
    # test get with multiple fields with expiration set through 'ex'
    assert await r.hgetex("test:hash", *test_keys, ex=10) == [
        b"bar",
        b"1",
        b"four",
    ]
    ttls = await r.httl("test:hash", *test_keys)
    for ttl in ttls:
        assert pytest.approx(ttl) == 10

    # test get with multiple fields removing expiration settings with 'persist'
    assert await r.hgetex("test:hash", *test_keys, persist=True) == [
        b"bar",
        b"1",
        b"four",
    ]
    assert await r.httl("test:hash", *test_keys) == [-1, -1, -1]

    # test get with multiple fields with expiration set through 'px'
    assert await r.hgetex("test:hash", *test_keys, px=6000) == [
        b"bar",
        b"1",
        b"four",
    ]
    ttls = await r.httl("test:hash", *test_keys)
    for ttl in ttls:
        assert pytest.approx(ttl) == 6

    # test get single field with expiration set through 'pxat'
    expire_at = await redis_server_time(r) + timedelta(minutes=1)
    assert await r.hgetex("test:hash", "foo", pxat=expire_at) == [b"bar"]
    assert (await r.httl("test:hash", "foo"))[0] <= 61

    # test get single field with expiration set through 'exat'
    expire_at = await redis_server_time(r) + timedelta(seconds=10)
    assert await r.hgetex("test:hash", "foo", exat=expire_at) == [b"bar"]
    assert (await r.httl("test:hash", "foo"))[0] <= 10


@skip_if_server_version_lt("7.9.0")