From 52851e10f64dae1f23d5e82132a0114812ecadd0 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 7 Nov 2026 07:48:38 -0800
Subject: [PATCH] test(routing): keep the Redis MONITOR capture alive across
 idle gaps

The capture read monitor.listen() with a 1 s socket timeout. listen() is a
generator, so the first timeout closed it and the thread stopped recording
without an error. Read with next_command() and no timeout, stop on the
capture's own stop marker, and fail clearly when that marker is never seen.
---
 .../routing/test_usage_based_routing_redis_reads.py | 13 +++++--------
 1 file changed, 5 insertions(+), 8 deletions(-)

diff --git a/tests/integration/routing/test_usage_based_routing_redis_reads.py b/tests/integration/routing/test_usage_based_routing_redis_reads.py
index a9dd10003a4..a580364ffaa 100644
--- a/tests/integration/routing/test_usage_based_routing_redis_reads.py
+++ b/tests/integration/routing/test_usage_based_routing_redis_reads.py
@@ -21,7 +21,6 @@ from integration._support.redis_process import owned_redis
 from integration._support.wire import Reply, Request, wire_server
 from pydantic import JsonValue, TypeAdapter
 from redis import Redis
-from redis.exceptions import TimeoutError as RedisTimeoutError
 
 JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 MODEL_INFO_ENTRIES: Final = TypeAdapter(tuple[dict[str, JsonValue], ...])
@@ -130,21 +129,19 @@ def _capture_redis_commands(host: str, port: int) -> Iterator[SimpleQueue[str]]:
     stop_marker: Final = f"monitor-stop-{uuid.uuid4().hex}"
 
     def capture() -> None:
-        with Redis(host=host, port=port, socket_timeout=1, decode_responses=True) as client:
+        with Redis(host=host, port=port, decode_responses=True) as client:
             with client.monitor() as monitor:
                 started.set()
-                stream: Final = iter(monitor.listen())
                 while not stopped.is_set():
-                    try:
-                        record: Final = MONITOR_COMMAND.validate_python(next(stream))
-                    except RedisTimeoutError:
-                        continue
+                    record: Final = MONITOR_COMMAND.validate_python(monitor.next_command())
                     command: Final = record.get("command")
                     if not isinstance(command, str):
                         continue
                     commands.put(command)
                     if ready_marker in command:
                         armed.set()
+                    if stop_marker in command:
+                        stopped.set()
 
     thread: Final = threading.Thread(target=capture, daemon=True)
     thread.start()
@@ -155,9 +152,9 @@ def _capture_redis_commands(host: str, port: int) -> Iterator[SimpleQueue[str]]:
         assert armed.wait(timeout=5), "Redis MONITOR did not capture its readiness command"
         yield commands
     finally:
-        stopped.set()
         with Redis(host=host, port=port, socket_timeout=1, decode_responses=True) as client:
             client.set(stop_marker, "stop", ex=1)
+        assert stopped.wait(timeout=5), "Redis MONITOR stopped before it captured the stop command"
         thread.join(timeout=5)
         assert not thread.is_alive(), "Redis MONITOR thread survived cleanup"