test(guardrails): run the cache-hit redis outage test on the shared owned_redis helper (#42925)

* test(guardrails): run the cache-hit redis outage test on the shared owned_redis helper

test_redis_outage_keeps_serving_in_memory_hits (#42780) spawned `redis-server`
straight from PATH. The CircleCI integration machine has no redis-server
binary, so the test died with FileNotFoundError before reaching the proxy.

Use integration._support.redis_process.owned_redis, which runs the local
binary when there is one and otherwise the job's redis-cache container, and
take the outage with its stop()/start() pair the way test_redis_recovery
already does. The assertions are unchanged.

* test(guardrails): describe the owned_redis stop as an outage, not a kill

tests/integration/observability/test_cache_hit_guardrail_metrics_chaos.py

@@ -1,12 +1,9 @@
 import json
 import signal
-import socket
-import subprocess
 import threading
 import uuid
-from collections.abc import Callable, Generator
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
 from pathlib import Path
 from typing import Final
 
@@ -14,6 +11,7 @@ import httpx
 import psutil
 from integration._support.client import Gateway, eventually, object_value, string_value
 from integration._support.process import owned_proxy_process
+from integration._support.redis_process import owned_redis
 from integration._support.wire import Reply, Request, wire_server
 from prometheus_client.parser import text_string_to_metric_families
 from test_cache_hit_guardrail_metrics import (
@@ -31,22 +29,6 @@ from test_cache_hit_guardrail_metrics import (
 BURST: Final = 10
 
 
-@contextmanager
-def _redis(port: int) -> Generator[subprocess.Popen[bytes], None, None]:
-    process: Final = subprocess.Popen(["redis-server", "--port", str(port), "--save", ""], stdout=subprocess.DEVNULL)
-    try:
-        yield process
-    finally:
-        process.kill()
-        process.wait(timeout=10)
-
-
-def _free_port() -> int:
-    with socket.socket() as reserve:
-        reserve.bind(("127.0.0.1", 0))
-        return reserve.getsockname()[1]
-
-
 def _deployment_id(candidate: Gateway, model_name: str) -> str:
     entries: Final = candidate.get("/model/info")["data"]
     assert isinstance(entries, list)
@@ -224,18 +206,16 @@ def test_stalled_guardrail_sink_recovers_and_counts(gateway: Gateway, tmp_path:
 
 
 def test_redis_outage_keeps_serving_in_memory_hits(gateway: Gateway, tmp_path: Path) -> None:
-    """X2: the redis cache keeps an in-memory shadow, so a redis kill does not stop cache-hit rejects."""
+    """X2: the redis cache keeps an in-memory shadow, so a redis outage does not stop cache-hit rejects."""
     marker: Final = uuid.uuid4().hex
-    port: Final = _free_port()
-    with _redis(port) as redis_one:
-        with _rig(gateway, tmp_path, marker, env={"REDIS_HOST": "127.0.0.1", "REDIS_PORT": str(port)}) as rig:
+    with owned_redis(tmp_path) as cache:
+        with _rig(gateway, tmp_path, marker, env={"REDIS_HOST": cache.host, "REDIS_PORT": str(cache.port)}) as rig:
             bodies: Final = _burst_bodies(rig, marker, None)[:BURST]
             _warm(rig, bodies)
             reject: Final = rig.candidate.request("POST", *bodies[0])
             assert reject.status_code == 400, reject.text
             warmed_hits: Final = rig.provider.received.qsize()
-            redis_one.kill()
-            redis_one.wait(timeout=10)
+            cache.stop()
             outcomes: Final = _fire(rig, bodies[1:])
             assert all(status == 400 for status, _ in outcomes), outcomes
             assert rig.provider.received.qsize() == warmed_hits, (
@@ -243,13 +223,13 @@ def test_redis_outage_keeps_serving_in_memory_hits(gateway: Gateway, tmp_path: P
                 warmed_hits,
                 rig.provider.received.qsize(),
             )
-            with _redis(port):
-                recovered: Final = rig.candidate.request(
-                    "POST",
-                    "/v1/chat/completions",
-                    _chat_body(rig.model_name, "x2 rehit " + marker, rig.guardrail_name),
-                )
-                assert recovered.status_code == 400, recovered.text
+            cache.start()
+            recovered: Final = rig.candidate.request(
+                "POST",
+                "/v1/chat/completions",
+                _chat_body(rig.model_name, "x2 rehit " + marker, rig.guardrail_name),
+            )
+            assert recovered.status_code == 400, recovered.text
             _expect_exactly_once(rig, (rig.model_name,), (rig.deployment_id,), 1 + len(bodies))