fix(tests): match the OS bind error in the owned-proxy port-race retry (#44462)

* fix(tests): match the OS bind error in the owned-proxy port-race retry

* test(integration): keep the port-race predicate pure so its unit tests stay in-process

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-03 17:07:10 -07:00 • committed by GitHub
parent f0eda6d2a6
commit dfdd496db8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 45 additions and 4 deletions

View file

@ -1,3 +1,4 @@
import errno
import os
import signal
import socket
@ -112,6 +113,7 @@ def _stop(process: subprocess.Popen[bytes]) -> None:
_PORT_ATTEMPTS: Final = 3
_BIND_COLLISION: Final = os.strerror(errno.EADDRINUSE)
def _free_port() -> int:
@ -142,8 +144,8 @@ def _launch(command: tuple[str, ...], root: Path, environment: Mapping[str, str]
return _Launch(process, port, log_path)
def _lost_port_race(launch: _Launch) -> bool:
return launch.process.poll() is not None and "address already in use" in launch.log.read_text()
def _lost_port_race(exit_code: int | None, log: Path) -> bool:
return exit_code is not None and _BIND_COLLISION in log.read_text()
def _wait_until_ready(launch: _Launch) -> None:
@ -165,13 +167,14 @@ def _launch_until_bound(
launch: Final = _launch(command, root, environment, output)
try:
_wait_until_ready(launch)
assert launch.process.poll() is None or (attempts > 1 and _lost_port_race(launch)), (
exit_code: Final = launch.process.poll()
assert exit_code is None or (attempts > 1 and _lost_port_race(exit_code, launch.log)), (
"Owned proxy exited before readiness"
)
except BaseException:
_stop(launch.process)
raise
if launch.process.poll() is None:
if exit_code is None:
return launch
_stop(launch.process)
return _launch_until_bound(command, root, environment, output, attempts - 1)

View file

@ -0,0 +1,38 @@
from __future__ import annotations
import errno
import importlib
import os
from pathlib import Path
from types import ModuleType
from typing import Final
import pytest
TESTS_DIR: Final = Path(__file__).resolve().parents[2]
BIND_ERROR_LINE: Final = f"ERROR: {OSError(errno.EADDRINUSE, os.strerror(errno.EADDRINUSE))}\n"
UNRELATED_CRASH: Final = "Traceback (most recent call last):\nModuleNotFoundError: No module named 'litellm'\n"
@pytest.fixture
def process_module(monkeypatch: pytest.MonkeyPatch) -> ModuleType:
monkeypatch.syspath_prepend(str(TESTS_DIR))
return importlib.import_module("integration._support.process")
def _written_log(directory: Path, text: str) -> Path:
log: Final = directory / "owned-proxy.log"
log.write_text(text)
return log
def test_lost_port_race_matches_the_bind_error_the_server_logs(process_module: ModuleType, tmp_path: Path) -> None:
assert process_module._lost_port_race(1, _written_log(tmp_path, BIND_ERROR_LINE))
def test_lost_port_race_ignores_an_exit_for_another_reason(process_module: ModuleType, tmp_path: Path) -> None:
assert not process_module._lost_port_race(1, _written_log(tmp_path, UNRELATED_CRASH))
def test_lost_port_race_needs_the_process_to_have_exited(process_module: ModuleType, tmp_path: Path) -> None:
assert not process_module._lost_port_race(None, _written_log(tmp_path, BIND_ERROR_LINE))