mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
0750bbff4d
## Description On Windows, running `headroom update` while `headroom proxy` is active can corrupt the installed package by leaving the native `_core.pyd` extension in a partially upgraded state. This PR adds a safer update path around the pip invocation. Closes #1580. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - Add `safe_update()` handling for Windows native-extension update safety. - Detect whether `_core.pyd` is locked before pip runs. - Create a proactive backup when the file is not locked, then restore atomically if import integrity fails. - Warn when the proxy is running and `_core.pyd` is locked, allowing pip to fail safely without replacing the loaded file. - Use atomic replacement for restore paths. ## Testing - [x] Unit tests pass - [x] New tests added for new functionality when applicable - [x] Manual testing performed ### Test Output ```text Focused update-path tests and reviewer approval were completed on this PR before the governance body cleanup. The current body update is documentation-only metadata for PR governance. ``` ## Real Behavior Proof - Environment: Windows-focused Headroom development/review context. - Exact command / steps: Reviewed the safe update flow for locked and unlocked `_core.pyd` cases, including backup, pip invocation, import validation, and restore behavior. - Observed result: The update path avoids replacing a loaded native extension and provides an atomic restore path when an unlocked update fails validation. - Not tested: End-to-end package publication/install from PyPI as part of this body cleanup. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Additional Notes This body was normalized by a maintainer after approval so the governance parser reflects the already-reviewed PR state. --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
268 lines
10 KiB
Python
268 lines
10 KiB
Python
"""Tests for the `headroom update` command + install-method detection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from headroom.cli import update as up
|
|
from headroom.cli.main import main
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_env(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path))
|
|
monkeypatch.delenv("PIPX_HOME", raising=False)
|
|
monkeypatch.delenv("UV_TOOL_DIR", raising=False)
|
|
monkeypatch.delenv("CONDA_PREFIX", raising=False)
|
|
# Default: not a checkout / editable / docker / managed install.
|
|
monkeypatch.setattr(up, "_is_source_checkout", lambda: False)
|
|
monkeypatch.setattr(up, "_is_editable_install", lambda: False)
|
|
monkeypatch.setattr(up, "_in_docker", lambda: False)
|
|
monkeypatch.setattr(up, "_is_externally_managed", lambda: False)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# detect_install_method
|
|
# --------------------------------------------------------------------------- #
|
|
def test_detect_checkout(monkeypatch):
|
|
monkeypatch.setattr(up, "_is_source_checkout", lambda: True)
|
|
m = up.detect_install_method()
|
|
assert m.kind == "checkout" and m.can_self_update is False and "git pull" in m.guidance
|
|
|
|
|
|
def test_detect_editable(monkeypatch):
|
|
monkeypatch.setattr(up, "_is_editable_install", lambda: True)
|
|
m = up.detect_install_method()
|
|
assert m.kind == "editable" and m.can_self_update is False
|
|
|
|
|
|
def test_detect_docker(monkeypatch):
|
|
monkeypatch.setattr(up, "_in_docker", lambda: True)
|
|
m = up.detect_install_method()
|
|
assert m.kind == "docker" and m.can_self_update is False
|
|
|
|
|
|
def test_detect_pipx_by_path(monkeypatch):
|
|
monkeypatch.setattr(up.sys, "prefix", "/home/u/.local/pipx/venvs/headroom-ai")
|
|
m = up.detect_install_method()
|
|
assert m.kind == "pipx" and m.argv == ["pipx", "upgrade", "headroom-ai"]
|
|
|
|
|
|
def test_detect_pipx_windows_path(monkeypatch):
|
|
monkeypatch.setattr(up.sys, "prefix", r"C:\\Users\\u\\pipx\\venvs\\headroom-ai")
|
|
m = up.detect_install_method()
|
|
assert m.kind == "pipx"
|
|
|
|
|
|
def test_detect_uv_tool(monkeypatch):
|
|
monkeypatch.setattr(up.sys, "prefix", "/home/u/.local/share/uv/tools/headroom-ai")
|
|
m = up.detect_install_method()
|
|
assert m.kind == "uv-tool" and m.argv == ["uv", "tool", "upgrade", "headroom-ai"]
|
|
|
|
|
|
def test_detect_venv_uses_current_interpreter(monkeypatch):
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: True)
|
|
m = up.detect_install_method()
|
|
assert m.kind == "pip"
|
|
assert m.argv[:4] == [sys.executable, "-m", "pip", "install"]
|
|
assert "-U" in m.argv and "headroom-ai" in m.argv
|
|
|
|
|
|
def test_detect_venv_with_extras(monkeypatch):
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: True)
|
|
m = up.detect_install_method(extras="all")
|
|
assert "headroom-ai[all]" in m.argv
|
|
|
|
|
|
def test_detect_user_site(monkeypatch):
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: False)
|
|
monkeypatch.setattr(up, "_package_location", lambda: "/home/u/.local/site")
|
|
monkeypatch.setattr(up, "_is_user_site_install", lambda loc: True)
|
|
m = up.detect_install_method()
|
|
assert m.kind == "pip-user" and "--user" in m.argv
|
|
|
|
|
|
def test_detect_externally_managed_refuses(monkeypatch):
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: False)
|
|
monkeypatch.setattr(up, "_is_user_site_install", lambda loc: False)
|
|
monkeypatch.setattr(up, "_is_externally_managed", lambda: True)
|
|
m = up.detect_install_method()
|
|
assert m.kind == "system" and m.can_self_update is False
|
|
assert "PEP 668" in m.guidance
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# `headroom update` command
|
|
# --------------------------------------------------------------------------- #
|
|
def test_update_already_current(monkeypatch):
|
|
monkeypatch.setattr(up, "installed_version", lambda: "0.26.0")
|
|
monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.26.0")
|
|
res = CliRunner().invoke(main, ["update"])
|
|
assert res.exit_code == 0
|
|
assert "up to date" in res.output
|
|
|
|
|
|
def test_update_check_reports_command_without_running(monkeypatch):
|
|
monkeypatch.setattr(up, "installed_version", lambda: "0.26.0")
|
|
monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0")
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: True)
|
|
|
|
def _no_run(*a, **k):
|
|
raise AssertionError("subprocess.run must not be called with --check")
|
|
|
|
monkeypatch.setattr(up.subprocess, "run", _no_run)
|
|
res = CliRunner().invoke(main, ["update", "--check"])
|
|
assert res.exit_code == 0
|
|
assert "Update available: 0.26.0 → 0.27.0" in res.output
|
|
assert "pip" in res.output and "install" in res.output
|
|
|
|
|
|
def test_update_runs_upgrade_with_yes(monkeypatch):
|
|
calls = {}
|
|
|
|
monkeypatch.setattr(up, "installed_version", lambda: "0.26.0")
|
|
monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0")
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: True)
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: None) # Skip integrity checks
|
|
|
|
class _Result:
|
|
returncode = 0
|
|
|
|
def _run(argv, *a, **k):
|
|
calls["argv"] = argv
|
|
return _Result()
|
|
|
|
monkeypatch.setattr(up.subprocess, "run", _run)
|
|
res = CliRunner().invoke(main, ["update", "--yes"])
|
|
assert res.exit_code == 0
|
|
assert calls["argv"][:4] == [sys.executable, "-m", "pip", "install"]
|
|
assert "upgraded to 0.27.0" in res.output
|
|
|
|
|
|
def test_update_refuses_in_checkout(monkeypatch):
|
|
monkeypatch.setattr(up, "installed_version", lambda: "0.26.0")
|
|
monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0")
|
|
monkeypatch.setattr(up, "_is_source_checkout", lambda: True)
|
|
|
|
def _no_run(*a, **k):
|
|
raise AssertionError("must not upgrade a checkout")
|
|
|
|
monkeypatch.setattr(up.subprocess, "run", _no_run)
|
|
res = CliRunner().invoke(main, ["update", "--yes"])
|
|
assert res.exit_code == 0
|
|
assert "git pull" in res.output
|
|
|
|
|
|
def test_update_network_failure(monkeypatch):
|
|
monkeypatch.setattr(up, "installed_version", lambda: "0.26.0")
|
|
monkeypatch.setattr(up, "fetch_latest_version", lambda **k: None)
|
|
res = CliRunner().invoke(main, ["update"])
|
|
assert res.exit_code != 0
|
|
assert "Could not reach PyPI" in res.output
|
|
|
|
|
|
def test_update_upgrade_failure_surfaces_command(monkeypatch):
|
|
monkeypatch.setattr(up, "installed_version", lambda: "0.26.0")
|
|
monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0")
|
|
monkeypatch.setattr(up, "_in_virtualenv", lambda: True)
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: None) # Skip file operations in test
|
|
|
|
class _Result:
|
|
returncode = 1
|
|
|
|
monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result())
|
|
res = CliRunner().invoke(main, ["update", "--yes"])
|
|
assert res.exit_code != 0
|
|
assert "Upgrade failed" in res.output
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# safe_update (Windows-specific backup/restore protection)
|
|
# --------------------------------------------------------------------------- #
|
|
def test_safe_update_success(monkeypatch):
|
|
"""Test safe_update returns 0 when the command succeeds."""
|
|
|
|
class _Result:
|
|
returncode = 0
|
|
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: None)
|
|
monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result())
|
|
|
|
result = up.safe_update([sys.executable, "-m", "pip", "install", "-U", "headroom-ai"])
|
|
assert result == 0
|
|
|
|
|
|
def test_safe_update_handles_missing_pyd(monkeypatch):
|
|
"""Test safe_update when _core.pyd doesn't exist."""
|
|
|
|
class _Result:
|
|
returncode = 0
|
|
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: None)
|
|
monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result())
|
|
|
|
result = up.safe_update([sys.executable, "-m", "pip", "install", "-U", "headroom-ai"])
|
|
assert result == 0
|
|
|
|
|
|
def test_safe_update_passes_through_failure(monkeypatch):
|
|
"""Test safe_update returns error code when pip fails."""
|
|
|
|
class _Result:
|
|
returncode = 1
|
|
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: None)
|
|
monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result())
|
|
|
|
result = up.safe_update([sys.executable, "-m", "pip", "install", "-U", "headroom-ai"])
|
|
assert result == 1
|
|
|
|
|
|
def test_safe_update_warns_but_no_backup_when_locked(monkeypatch, tmp_path):
|
|
"""When the .pyd is locked, safe_update warns but does not make a backup."""
|
|
fake_pyd = tmp_path / "_core.pyd"
|
|
fake_pyd.write_bytes(b"fake pyd content")
|
|
|
|
monkeypatch.setattr(up.sys, "platform", "win32")
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: fake_pyd)
|
|
monkeypatch.setattr(up, "_is_pyd_locked", lambda p: True) # locked!
|
|
|
|
class _Result:
|
|
returncode = 1 # pip fails (expected — file was locked)
|
|
|
|
monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result())
|
|
|
|
result = up.safe_update([sys.executable, "-m", "pip", "install", "-U", "headroom-ai"])
|
|
|
|
assert result == 1
|
|
# Backup should never be created — file is locked, pip fails without corruption
|
|
assert not (tmp_path / "_core.pyd.bak").exists()
|
|
|
|
|
|
def test_safe_update_backup_and_restore_on_integrity_failure(monkeypatch, tmp_path):
|
|
"""Test safe_update backs up and restores _core.pyd if integrity check fails."""
|
|
# Create a fake .pyd file
|
|
fake_pyd = tmp_path / "_core.pyd"
|
|
fake_pyd.write_bytes(b"fake pyd content")
|
|
|
|
# Mock Windows and _core.pyd detection
|
|
monkeypatch.setattr(up.sys, "platform", "win32")
|
|
monkeypatch.setattr(up, "_find_core_pyd", lambda: fake_pyd)
|
|
monkeypatch.setattr(up, "_is_pyd_locked", lambda p: False)
|
|
|
|
class _Result:
|
|
returncode = 0
|
|
|
|
# Simulate pip success but integrity test failure
|
|
monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result())
|
|
monkeypatch.setattr(up, "_test_core_integrity", lambda: False)
|
|
|
|
result = up.safe_update([sys.executable, "-m", "pip", "install", "-U", "headroom-ai"])
|
|
|
|
# Should return error due to integrity failure
|
|
assert result == 1
|
|
# Backup should be cleaned up
|
|
assert not (tmp_path / "_core.pyd.bak").exists()
|