Method test_acl_log

tests/test_commands.py:440–483

(self, r, request)

    @skip_if_redis_enterprise()
    @pytest.mark.onlynoncluster
    def test_acl_log(self, r, request):
        username = "redis-py-user"

        def teardown():
            r.acl_deluser(username)

        request.addfinalizer(teardown)
        r.acl_setuser(
            username,
            enabled=True,
            reset=True,
            commands=["+get", "+set", "+select"],
            keys=["cache:*"],
            nopass=True,
        )
        r.acl_log_reset()

        user_client = _get_client(
            redis.Redis, 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(exceptions.NoPermissionError):
            user_client.get("violated_cache:0")

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

        assert isinstance(r.acl_log(), list)
        assert len(r.acl_log()) == 3
        assert len(r.acl_log(count=1)) == 1
        assert isinstance(r.acl_log()[0], dict)
        expected = r.acl_log(count=1)[0]
        assert_resp_response_in(
            r,
            "client-info",
            expected,
            expected.keys(),
        )

    @skip_if_server_version_lt("6.0.0")
    @skip_if_redis_enterprise()