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>
This commit is contained in:
devin-ai-integration[bot] 2026-11-04 21:07:21 +00:00 • committed by GitHub
parent de85f59539
commit 81e8d652b0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -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)
)