Download tests/scripts/install/test_install_ps1_venv_process_tree.py from SaylorTwift/hermes-agent: direct link, hf CLI and curl.
- Browser
- Download file 6.19 kB
-
https://huggingface.co/SaylorTwift/hermes-agent/resolve/main/tests/scripts/install/test_install_ps1_venv_process_tree.py
- Command line
-
hf download hf://SaylorTwift/hermes-agent/tests/scripts/install/test_install_ps1_venv_process_tree.py
-
curl -L -o test_install_ps1_venv_process_tree.py https://huggingface.co/SaylorTwift/hermes-agent/resolve/main/tests/scripts/install/test_install_ps1_venv_process_tree.py
6.19 kB
| """Windows installer regression for Hermes children outside the venv. | |
| The venv sweep deliberately selects process roots by executable path so it | |
| does not kill unrelated Python processes. A selected Hermes process can spawn | |
| a managed-runtime child whose executable lives outside the venv, though. The | |
| installer must stop that whole tree before replacing the venv. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import shutil | |
| import subprocess | |
| import time | |
| from pathlib import Path | |
| import psutil | |
| import pytest | |
| from tests.install_ps1_fake_uv import compile_fake_uv | |
| REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent | |
| INSTALL_PS1 = REPO_ROOT / "scripts" / "install.ps1" | |
| POWERSHELL = next( | |
| (candidate for candidate in ("powershell", "pwsh") if shutil.which(candidate)), | |
| None, | |
| ) | |
| def _pid_is_running(pid: int) -> bool: | |
| return psutil.pid_exists(pid) | |
| def _wait_until_stopped(pid: int, timeout: float = 10) -> bool: | |
| deadline = time.monotonic() + timeout | |
| while time.monotonic() < deadline: | |
| if not _pid_is_running(pid): | |
| return True | |
| time.sleep(0.1) | |
| return not _pid_is_running(pid) | |
| def _find_child_pid(parent_pid: int, executable: Path, timeout: float = 10) -> int: | |
| expected = str(executable).replace("'", "''") | |
| query = ( | |
| f"$expected = '{expected}'; " | |
| f"Get-CimInstance Win32_Process -Filter 'ParentProcessId = {parent_pid}' | " | |
| "Where-Object { $_.ExecutablePath -and " | |
| "[string]::Equals($_.ExecutablePath, $expected, " | |
| "[System.StringComparison]::OrdinalIgnoreCase) } | " | |
| "Select-Object -First 1 -ExpandProperty ProcessId" | |
| ) | |
| deadline = time.monotonic() + timeout | |
| while time.monotonic() < deadline: | |
| result = subprocess.run( | |
| [POWERSHELL, "-NoProfile", "-Command", query], | |
| capture_output=True, | |
| text=True, | |
| ) | |
| if result.returncode == 0 and result.stdout.strip(): | |
| return int(result.stdout.strip()) | |
| time.sleep(0.1) | |
| raise AssertionError(f"child process did not start under PID {parent_pid}") | |
| def _stop_tree(pid: int) -> None: | |
| subprocess.run( | |
| ["taskkill", "/PID", str(pid), "/T", "/F"], | |
| capture_output=True, | |
| text=True, | |
| ) | |
| def _write_cmd(path: Path, text: str) -> None: | |
| with path.open("w", encoding="ascii", newline="\r\n") as handle: | |
| handle.write(text) | |
| def test_venv_sweep_stops_managed_runtime_children_but_not_unrelated_processes( | |
| tmp_path: Path, | |
| ) -> None: | |
| hermes_home = tmp_path / "hermes-home" | |
| install_dir = hermes_home / "hermes-agent" | |
| venv_scripts = install_dir / "venv" / "Scripts" | |
| runtime_dir = install_dir / ".hermes-runtime" / "python" / "generation-test" | |
| unrelated_dir = tmp_path / "unrelated" | |
| fake_bin = tmp_path / "fake-bin" | |
| for directory in (venv_scripts, runtime_dir, unrelated_dir, fake_bin): | |
| directory.mkdir(parents=True) | |
| system_cmd = Path(os.environ["SystemRoot"]) / "System32" / "cmd.exe" | |
| venv_parent_exe = venv_scripts / "python.exe" | |
| runtime_child_exe = runtime_dir / "python.exe" | |
| unrelated_exe = unrelated_dir / "python.exe" | |
| for target in (venv_parent_exe, runtime_child_exe, unrelated_exe): | |
| shutil.copy2(system_cmd, target) | |
| parent_script = tmp_path / "parent.cmd" | |
| _write_cmd( | |
| parent_script, | |
| f'@"{runtime_child_exe}" /d /c ping -t 127.0.0.1 ^>nul\n', | |
| ) | |
| unrelated_script = tmp_path / "unrelated.cmd" | |
| _write_cmd(unrelated_script, "@ping -t 127.0.0.1 >nul\n") | |
| # Keep the test away from real gateway tasks and real Hermes launchers | |
| # while still exercising the installer's actual process enumeration and | |
| # per-PID taskkill behavior. | |
| _write_cmd(fake_bin / "schtasks.cmd", "@exit /b 0\n") | |
| _write_cmd( | |
| fake_bin / "taskkill.cmd", | |
| "@echo off\n" | |
| 'echo %* | "%SystemRoot%\\System32\\findstr.exe" /I ' | |
| '/C:"/IM hermes.exe" >nul\n' | |
| "if not errorlevel 1 exit /b 0\n" | |
| '"%SystemRoot%\\System32\\taskkill.exe" %*\n', | |
| ) | |
| uv = hermes_home / "bin" / "uv.exe" | |
| uv.parent.mkdir(parents=True) | |
| compile_fake_uv(POWERSHELL, uv) | |
| wrapper = tmp_path / "run-venv-stage.ps1" | |
| wrapper.write_text( | |
| f"function global:schtasks {{ & '{fake_bin / 'schtasks.cmd'}' @args }}\n" | |
| f"function global:taskkill {{ & '{fake_bin / 'taskkill.cmd'}' @args }}\n" | |
| f"& '{INSTALL_PS1}' -Stage venv -NonInteractive " | |
| f"-InstallDir '{install_dir}' -HermesHome '{hermes_home}'\n" | |
| "exit $LASTEXITCODE\n", | |
| encoding="utf-8", | |
| ) | |
| creation_flags = subprocess.CREATE_NO_WINDOW | |
| parent = subprocess.Popen( | |
| [str(venv_parent_exe), "/d", "/c", str(parent_script)], | |
| creationflags=creation_flags, | |
| ) | |
| unrelated = subprocess.Popen( | |
| [str(unrelated_exe), "/d", "/c", str(unrelated_script)], | |
| creationflags=creation_flags, | |
| ) | |
| child_pid = 0 | |
| try: | |
| child_pid = _find_child_pid(parent.pid, runtime_child_exe) | |
| assert _pid_is_running(child_pid) | |
| assert _pid_is_running(unrelated.pid) | |
| env = os.environ | { | |
| "OS": "Windows_NT", | |
| "PATH": str(fake_bin) + os.pathsep + os.environ["PATH"], | |
| "FAKE_UV_LOG": str(tmp_path / "uv.log"), | |
| "FAKE_MANAGED_PYTHON": str(runtime_child_exe), | |
| "FAKE_THIRD_PARTY_PYTHON": str(unrelated_exe), | |
| } | |
| result = subprocess.run( | |
| [ | |
| POWERSHELL, | |
| "-NoProfile", | |
| "-File", | |
| str(wrapper), | |
| ], | |
| cwd=tmp_path, | |
| env=env, | |
| capture_output=True, | |
| text=True, | |
| ) | |
| assert result.returncode == 0, result.stdout + result.stderr | |
| assert _wait_until_stopped(parent.pid) | |
| assert _wait_until_stopped(child_pid) | |
| assert _pid_is_running(unrelated.pid) | |
| finally: | |
| for pid in (child_pid, parent.pid, unrelated.pid): | |
| if pid: | |
| _stop_tree(pid) | |