diff --git a/tests/integration/observability/test_cache_hit_guardrail_metrics_chaos.py b/tests/integration/observability/test_cache_hit_guardrail_metrics_chaos.py index faaa1fc7325..c77b8eebe33 100644 --- a/tests/integration/observability/test_cache_hit_guardrail_metrics_chaos.py +++ b/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))