Extend cache-key handling to xpending_range and sort_ro

Following review, cover the other allow-listed commands that reached
execute_command without `keys`:

- xpending_range() sent XPENDING with no keys, so it raised
  ValueError("Cannot create cache key.") under CSC. Pass keys=[name].
- sort_ro() delegated to sort(), which sends SORT, so SORT_RO (on the allow
  list) was never cached. Send SORT_RO with keys=[key] via a private
  _command argument to sort(); the write path (sort(store=...)) still sends
  SORT. The SORT response callback is a no-op without `groups` (which sort_ro
  never uses), so response parsing is unaffected.

Add a regression test per command in tests/test_cache.py.

(Deferred: georadius/georadiusbymember _RO caching. The GEO response
callbacks are keyed on the base command names, so sending the _RO variants
would change response parsing for withdist/withcoord/withhash calls, and the
_RO commands only exist since Redis 3.2.10. That needs _RO parser aliases and
version gating and is better handled separately.)

Author: uttam12331

redis/commands/core.py

@@ -5399,6 +5399,7 @@ def sort(
         alpha: bool = False,
         store: str | None = None,
         groups: bool | None = False,
+        _command: str = "SORT",
     ) -> SortResponse | Awaitable[SortResponse]:
         """
         Sort and return the list, set or sorted set at ``name``.
@@ -5459,7 +5460,7 @@ def sort(
 
         options = {"groups": len(get) if groups else None}
         options["keys"] = [name]
-        return self.execute_command("SORT", *pieces, **options)
+        return self.execute_command(_command, *pieces, **options)
 
     @overload
     def sort_ro(
@@ -5515,7 +5516,14 @@ def sort_ro(
         For more information, see https://redis.io/commands/sort_ro
         """
         return self.sort(
-            key, start=start, num=num, by=by, get=get, desc=desc, alpha=alpha
+            key,
+            start=start,
+            num=num,
+            by=by,
+            get=get,
+            desc=desc,
+            alpha=alpha,
+            _command="SORT_RO",
         )
 
 
@@ -7796,7 +7804,7 @@ def xpending_range(
         if consumername:
             pieces.append(consumername)
 
-        return self.execute_command("XPENDING", *pieces, parse_detail=True)
+        return self.execute_command("XPENDING", *pieces, parse_detail=True, keys=[name])
 
     @overload
     def xrange(

tests/test_cache.py

@@ -194,6 +194,81 @@ def test_zrank_zrevrank_are_cacheable(self, r, r2):
         # rank of "b" shifts from 1 to 2 after inserting a lower-scored member
         assert r.zrank("myzset", "b") == 2
 
+    @pytest.mark.parametrize(
+        "r",
+        [
+            {
+                "cache": DefaultCache(CacheConfig(max_size=5)),
+                "single_connection_client": True,
+            },
+            {
+                "cache": DefaultCache(CacheConfig(max_size=5)),
+                "single_connection_client": False,
+            },
+        ],
+        ids=["single", "pool"],
+        indirect=True,
+    )
+    @pytest.mark.onlynoncluster
+    def test_xpending_range_is_cacheable(self, r):
+        # XPENDING is on the allow list, but xpending_range passed no `keys`,
+        # so client-side caching raised ValueError("Cannot create cache key.").
+        cache = r.get_cache()
+        stream, group = "stream", "group"
+        r.delete(stream)
+        r.xadd(stream, {"foo": "bar"})
+        r.xgroup_create(stream, group, 0)
+        # must not raise, and the result must be cached under the stream key
+        assert r.xpending_range(stream, group, min="-", max="+", count=5) == []
+        assert (
+            cache.get(
+                CacheKey(
+                    command="XPENDING",
+                    redis_keys=(stream,),
+                    redis_args=("XPENDING", stream, group, "-", "+", 5),
+                )
+            )
+            is not None
+        )
+
+    @pytest.mark.parametrize(
+        "r",
+        [
+            {
+                "cache": DefaultCache(CacheConfig(max_size=5)),
+                "single_connection_client": True,
+            },
+            {
+                "cache": DefaultCache(CacheConfig(max_size=5)),
+                "single_connection_client": False,
+            },
+        ],
+        ids=["single", "pool"],
+        indirect=True,
+    )
+    @pytest.mark.onlynoncluster
+    def test_sort_ro_is_cacheable(self, r, r2):
+        # sort_ro delegated to sort() and sent SORT, so SORT_RO (on the allow
+        # list) was never actually cached. It now sends SORT_RO with the key.
+        cache = r.get_cache()
+        r.delete("mylist")
+        r.rpush("mylist", "3", "1", "2")
+        assert r.sort_ro("mylist") == [b"1", b"2", b"3"]
+        assert (
+            cache.get(
+                CacheKey(
+                    command="SORT_RO",
+                    redis_keys=("mylist",),
+                    redis_args=("SORT_RO", "mylist"),
+                )
+            )
+            is not None
+        )
+        # mutate from a second client -> invalidation
+        r2.rpush("mylist", "0")
+        time.sleep(0.1)
+        assert r.sort_ro("mylist") == [b"0", b"1", b"2", b"3"]
+
     @pytest.mark.parametrize(
         "r",
         [