fix(cron): avoid Windows Python launcher popups

This commit is contained in:
helix4u 2026-07-08 17:53:20 -06:00 committed by Teknium
parent d9ee342414
commit 1b63737f55
2 changed files with 145 additions and 3 deletions

View file

@ -2052,6 +2052,64 @@ def _get_script_timeout() -> int:
return _DEFAULT_SCRIPT_TIMEOUT
def _read_windows_pyvenv_cfg(venv_dir: Path) -> dict[str, str]:
cfg_path = venv_dir / "pyvenv.cfg"
try:
lines = cfg_path.read_text(encoding="utf-8").splitlines()
except OSError:
return {}
parsed: dict[str, str] = {}
for raw in lines:
if "=" not in raw:
continue
key, value = raw.split("=", 1)
parsed[key.strip().lower()] = value.strip()
return parsed
def _windows_cron_python_invocation(python_exe: str) -> tuple[str, dict[str, str]]:
"""Return an output-capable hidden Python invocation for Windows scripts.
Cron scripts capture stdout/stderr, so using ``pythonw.exe`` directly can
lose script output. uv-created venv ``python.exe`` launchers are also a
problem: even with CREATE_NO_WINDOW, the launcher can re-exec the base
console interpreter and flash a visible window. For uv venvs, bypass the
launcher and run the base ``python.exe`` directly with the venv paths
overlaid in the environment.
"""
if sys.platform != "win32":
return python_exe, {}
interpreter = Path(python_exe)
venv_dir = interpreter.parent.parent
env_overlay: dict[str, str] = {}
if interpreter.name.lower() == "pythonw.exe":
sibling = interpreter.with_name("python.exe")
if sibling.exists():
interpreter = sibling
cfg = _read_windows_pyvenv_cfg(venv_dir)
home = cfg.get("home", "")
site_packages = venv_dir / "Lib" / "site-packages"
if "uv" in cfg and home:
base_python = Path(home) / "python.exe"
if base_python.exists() and site_packages.exists():
interpreter = base_python
env_overlay["VIRTUAL_ENV"] = str(venv_dir)
pythonpath_entries = [
str(Path(__file__).resolve().parents[1]),
str(site_packages),
]
existing_pythonpath = os.environ.get("PYTHONPATH", "")
if existing_pythonpath:
pythonpath_entries.append(existing_pythonpath)
env_overlay["PYTHONPATH"] = os.pathsep.join(pythonpath_entries)
return str(interpreter), env_overlay
def _run_job_script(script_path: str) -> tuple[bool, str]:
"""Execute a cron job's data-collection script and capture its output.
@ -2129,22 +2187,28 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
f"Cannot run .sh/.bash script {path.name!r}: bash not found on PATH. "
"On Windows, install Git for Windows (which ships Git Bash) "
"or rewrite the script as Python (.py)."
)
)
argv = [_bash, str(path)]
env_overlay: dict[str, str] = {}
else:
argv = [sys.executable, str(path)]
python_exe, env_overlay = _windows_cron_python_invocation(sys.executable)
argv = [python_exe, str(path)]
try:
from tools.environments.local import _sanitize_subprocess_env
popen_kwargs = {"creationflags": windows_hide_flags()} if sys.platform == "win32" else {}
env = _sanitize_subprocess_env(os.environ.copy())
env.update(env_overlay)
result = subprocess.run(
argv,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=script_timeout,
cwd=str(path.parent),
env=_sanitize_subprocess_env(os.environ.copy()),
env=env,
**popen_kwargs,
)
stdout = (result.stdout or "").strip()

View file

@ -13,6 +13,7 @@ import sys
import textwrap
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
import pytest
@ -171,6 +172,83 @@ class TestRunJobScript:
assert success is True
assert output == "ABSENT"
def test_windows_uv_venv_python_script_bypasses_launcher(self, cron_env, tmp_path, monkeypatch):
from cron import scheduler as sched_mod
from cron.scheduler import _run_job_script
script = cron_env / "scripts" / "probe.py"
script.write_text('print("ok")\n')
venv = tmp_path / "venv"
venv_scripts = venv / "Scripts"
site_packages = venv / "Lib" / "site-packages"
base = tmp_path / "base"
venv_scripts.mkdir(parents=True)
site_packages.mkdir(parents=True)
base.mkdir()
venv_python = venv_scripts / "python.exe"
base_python = base / "python.exe"
venv_python.write_text("", encoding="utf-8")
base_python.write_text("", encoding="utf-8")
(venv / "pyvenv.cfg").write_text(f"home = {base}\nuv = true\n", encoding="utf-8")
captured = {}
def fake_run(argv, **kwargs):
captured["argv"] = argv
captured["kwargs"] = kwargs
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")
monkeypatch.setattr(sched_mod.sys, "platform", "win32")
monkeypatch.setattr(sched_mod.sys, "executable", str(venv_python))
monkeypatch.setattr(sched_mod, "windows_hide_flags", lambda: 0x08000000)
monkeypatch.setattr(sched_mod.subprocess, "run", fake_run)
success, output = _run_job_script("probe.py")
assert success is True
assert output == "ok"
assert captured["argv"] == [str(base_python), str(script.resolve())]
assert captured["kwargs"]["creationflags"] == 0x08000000
env = captured["kwargs"]["env"]
assert env["VIRTUAL_ENV"] == str(venv)
assert str(site_packages) in env["PYTHONPATH"]
def test_windows_pythonw_script_uses_sibling_python_for_captured_output(self, cron_env, tmp_path, monkeypatch):
from cron import scheduler as sched_mod
from cron.scheduler import _run_job_script
script = cron_env / "scripts" / "probe.py"
script.write_text('print("ok")\n')
venv = tmp_path / "venv"
venv_scripts = venv / "Scripts"
venv_scripts.mkdir(parents=True)
pythonw = venv_scripts / "pythonw.exe"
python = venv_scripts / "python.exe"
pythonw.write_text("", encoding="utf-8")
python.write_text("", encoding="utf-8")
captured = {}
def fake_run(argv, **kwargs):
captured["argv"] = argv
captured["kwargs"] = kwargs
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")
monkeypatch.setattr(sched_mod.sys, "platform", "win32")
monkeypatch.setattr(sched_mod.sys, "executable", str(pythonw))
monkeypatch.setattr(sched_mod, "windows_hide_flags", lambda: 0x08000000)
monkeypatch.setattr(sched_mod.subprocess, "run", fake_run)
success, output = _run_job_script("probe.py")
assert success is True
assert output == "ok"
assert captured["argv"] == [str(python), str(script.resolve())]
assert captured["kwargs"]["encoding"] == "utf-8"
assert captured["kwargs"]["errors"] == "replace"
def test_script_empty_output(self, cron_env):
from cron.scheduler import _run_job_script