Method test_acl_log

tests/test_asyncio/test_cluster.py:2587–2630

(
        self, r: RedisCluster, create_redis: Callable[..., RedisCluster]
    )

    @skip_if_server_version_lt("6.0.0")
    @skip_if_redis_enterprise()
    async def test_acl_log(
        self, r: RedisCluster, create_redis: Callable[..., RedisCluster]
    ) -> None:
        key = "{cache}:"
        node = r.get_node_from_key(key)
        username = "redis-py-user"

        await r.acl_setuser(
            username,
            enabled=True,
            reset=True,
            commands=["+get", "+set", "+select", "+cluster", "+command", "+info"],
            keys=["{cache}:*"],
            nopass=True,
            target_nodes="primaries",
        )
        await r.acl_log_reset(target_nodes=node)

        user_client = await create_redis(
            cls=RedisCluster, flushdb=False, username=username
        )

        # Valid operation and key
        assert await user_client.set("{cache}:0", 1)
        assert await user_client.get("{cache}:0") == b"1"

        # Invalid key
        with pytest.raises(NoPermissionError):
            await user_client.get("{cache}violated_cache:0")

        # Invalid operation
        with pytest.raises(NoPermissionError):
            await user_client.hset("{cache}:0", "hkey", "hval")

        assert isinstance(await r.acl_log(target_nodes=node), list)
        assert len(await r.acl_log(target_nodes=node)) == 3
        assert len(await r.acl_log(count=1, target_nodes=node)) == 1
        assert isinstance((await r.acl_log(target_nodes=node))[0], dict)
        assert "client-info" in (await r.acl_log(count=1, target_nodes=node))[0]
        assert await r.acl_log_reset(target_nodes=node)

        await r.acl_deluser(username, target_nodes="primaries")

        await user_client.aclose()

    @skip_if_server_version_lt("8.5.240")
    async def test_hotkeys_cluster(self, r: RedisCluster) -> None: