File size: 6,189 Bytes
674d56a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
"""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)


@pytest.mark.live_system_guard_bypass
@pytest.mark.skipif(
    os.name != "nt" or POWERSHELL is None,
    reason="needs Windows and PowerShell",
)
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)