Method test_acl_log

tests/test_cluster.py:2700–2741

(self, r, request)

    @skip_if_server_version_lt("6.0.0")
    @skip_if_redis_enterprise()
    def test_acl_log(self, r, request):
        key = "{cache}:"
        node = r.get_node_from_key(key)
        username = "redis-py-user"

        def teardown():
            r.acl_deluser(username, target_nodes="primaries")

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

        user_client = _get_client(
            RedisCluster, request, flushdb=False, username=username
        )

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

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

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

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

    def generate_lib_code(self, lib_name):
        return f"""#!js api_version=1.0 name={lib_name}\n redis.registerFunction('foo', ()=>{{return 'bar'}})"""  # noqa