Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 33 additions & 16 deletions sdk/python/tests/utils/cli_repo_creator.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,26 +63,43 @@ def _subprocess_env(self) -> dict[str, str]:
env["PYTHONPATH"] = os.pathsep.join(parent_paths)
return env

def run(self, args: List[str], cwd: Path) -> subprocess.CompletedProcess:
def run(
self, args: List[str], cwd: Path, attempts: int = 2
) -> subprocess.CompletedProcess:
# Apply a conservative timeout to prevent CI hangs from Dask atexit-handler
# stalls or other subprocess blockages.
timeout = 120

try:
return subprocess.run(
[sys.executable, cli.__file__] + args,
cwd=cwd,
capture_output=True,
timeout=timeout,
env=self._subprocess_env(),
)
except subprocess.TimeoutExpired:
return subprocess.CompletedProcess(
args=[sys.executable, cli.__file__] + args,
returncode=-1,
stdout=b"",
stderr=f"Command timed out after {timeout}s: {args}".encode(),
)
# Commands routed through here (e.g. `apply`) do not materialize, so they
# cannot trigger the Dask atexit hang described in the module docstring. A
# timeout is therefore almost always transient slowness — a cold child
# interpreter importing Feast (pandas/pyarrow/...) on a loaded CI runner,
# which has been observed to exceed the timeout on the slower macOS
# runners. Retry once so a single slow cold start does not flake the test:
# the retry runs with warm OS/page caches and typically finishes quickly,
# while a genuine repeated stall still fails once attempts are exhausted.
full_args = [sys.executable, cli.__file__] + args
result = None
for attempt in range(1, attempts + 1):
try:
return subprocess.run(
full_args,
cwd=cwd,
capture_output=True,
timeout=timeout,
env=self._subprocess_env(),
)
except subprocess.TimeoutExpired:
result = subprocess.CompletedProcess(
args=full_args,
returncode=-1,
stdout=b"",
stderr=(
f"Command timed out after {timeout}s "
f"(attempt {attempt}/{attempts}): {args}"
).encode(),
)
return result

def run_with_output(self, args: List[str], cwd: Path) -> Tuple[int, bytes]:
is_teardown = "teardown" in args
Expand Down
Loading