From 81e8d652b08dc275ef50d8a8647a7cb615336d8e Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 4 Nov 2026 21:07:21 +0000 Subject: [PATCH] test(observability): reserve sink port throughout outage (#50100) Co-authored-by: mateo <277851410+mateo-berri@users.noreply.github.com> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../test_arize_otel_v2_openinference_chaos.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py b/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py index 79b0d9084aa..0095080d944 100644 --- a/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py +++ b/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py @@ -1,12 +1,13 @@ from __future__ import annotations +import socket import threading import uuid from concurrent.futures import ThreadPoolExecutor from contextlib import ExitStack from pathlib import Path +from queue import SimpleQueue from typing import Final -from urllib.parse import urlsplit import httpx import pytest @@ -44,7 +45,7 @@ from _openinference_support import ( ) from integration._support.client import Gateway, eventually from integration._support.process import graceful_stop_seconds -from integration._support.wire import Reply, Request, wire_server +from integration._support.wire import Reply, Request, Wire, wire_server def _call( @@ -249,11 +250,12 @@ def test_arize_otel_v2_f1_sink_outage_and_recovery(gateway: Gateway, tmp_path: P _assert_response(response, marker, surface, stream, model_name) with ExitStack() as servers: - initial_stack: Final = servers.enter_context(ExitStack()) - stopped_destination: Final = initial_stack.enter_context(wire_server(_owned_sink_handler(sink))) - sink_port: Final = urlsplit(stopped_destination.url).port - assert sink_port is not None, stopped_destination.url - initial_stack.close() + outage_socket: Final = servers.enter_context(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) + outage_socket.bind(("127.0.0.1", 0)) # test-quality-ok: socket stays bound throughout the outage + sink_port: Final = outage_socket.getsockname()[1] + stopped_destination: Final = Wire( + f"http://127.0.0.1:{sink_port}", SimpleQueue(), SimpleQueue(), SimpleQueue(), SimpleQueue() + ) with _rig( gateway, tmp_path, @@ -272,6 +274,7 @@ def test_arize_otel_v2_f1_sink_outage_and_recovery(gateway: Gateway, tmp_path: P with httpx.Client(trust_env=False) as client, pytest.raises(httpx.ConnectError): client.get(stopped_destination.url + "/health", timeout=2) recovered_stack: Final = servers.enter_context(ExitStack()) + outage_socket.close() recovered_destination: Final = recovered_stack.enter_context( wire_server(_owned_sink_handler(sink), port=sink_port) )