diff --git a/cli_audit/reconcile.py b/cli_audit/reconcile.py index 82e1826..dd14a49 100644 --- a/cli_audit/reconcile.py +++ b/cli_audit/reconcile.py @@ -153,6 +153,30 @@ def summary(self) -> str: """ +# Environment-name patterns for env managers without a pyvenv.cfg (conda etc.). +# Mirrors the venv skip list in scripts/lib/capability.sh:detect_all_installations. +_ENV_DIR_PATTERNS = ( + "/venv/bin", "/.venv/bin", "/env/bin", + "/venvs/", "/.venvs/", "/virtualenvs/", "/.virtualenvs/", + "/envs/", "/conda/", "/miniconda", "/anaconda", +) + + +def _is_virtualenv_bin(bin_dir: str) -> bool: + """True if bin_dir is a virtualenv/conda environment's bin directory. + + Environments are not installations: their binaries vanish with the env, + and classifying them by method (e.g. `uv` because the tool also appears + in `uv tool list`) makes removal delete a DIFFERENT installation. + """ + # Definitive signal: PEP 405 venvs carry pyvenv.cfg next to bin/ + if os.path.isfile(os.path.join(os.path.dirname(bin_dir), "pyvenv.cfg")): + return True + # Name-based fallback for conda/virtualenvwrapper layouts + normalized = bin_dir.rstrip("/") + "/" + return any(pat in normalized for pat in _ENV_DIR_PATTERNS) + + def detect_installations( tool_name: str, candidates: Sequence[str] | None = None, @@ -193,6 +217,10 @@ def detect_installations( # Search each PATH directory for path_dir in path_dirs: + # Virtualenv/conda bins are environments, not installations + if _is_virtualenv_bin(path_dir): + vlog(f" Skipping environment dir: {path_dir}", verbose) + continue for candidate in candidates: full_path = os.path.join(path_dir, candidate) @@ -212,6 +240,11 @@ def detect_installations( if real_path in seen_paths: continue + # A symlink can point into an environment as well + if _is_virtualenv_bin(os.path.dirname(real_path)): + vlog(f" Skipping environment binary: {real_path}", verbose) + continue + seen_paths.add(real_path) # Get THIS install's own version by running its actual binary path. @@ -854,19 +887,25 @@ def _check_path_ordering( return tuple(issues) +# bulk_reconcile prompts from ThreadPool workers: without serialization all +# prompts print at once and the answers race for stdin. +_prompt_lock = threading.Lock() + + def _confirm_removal(tool_name: str, to_remove: list[Installation]) -> bool: - """Prompt user to confirm removal.""" + """Prompt user to confirm removal. Serialized across worker threads.""" if not sys.stdin.isatty(): # Non-interactive mode return False - print(f"\n⚠️ WARNING: About to remove {len(to_remove)} installation(s) of {tool_name}") - for inst in to_remove: - print(f" - {inst.version} ({inst.method}, {inst.path})") + with _prompt_lock: + print(f"\n⚠️ WARNING: About to remove {len(to_remove)} installation(s) of {tool_name}") + for inst in to_remove: + print(f" - {inst.version} ({inst.method}, {inst.path})") - print("\nProceed with removal? [y/N]: ", end="") - response = input().strip().lower() - return response in ('y', 'yes') + print("\nProceed with removal? [y/N]: ", end="") + response = input().strip().lower() + return response in ('y', 'yes') def _uninstall_installation(installation: Installation, verbose: bool) -> tuple[bool, str | None]: diff --git a/tests/test_reconcile_venv.py b/tests/test_reconcile_venv.py new file mode 100644 index 0000000..6b4b644 --- /dev/null +++ b/tests/test_reconcile_venv.py @@ -0,0 +1,156 @@ +"""Tests for reconcile venv exclusion and prompt serialization. + +`make reconcile-all` flagged binaries inside an activated virtualenv +(~/.venv/bin/black) as duplicate installations, classified them as `uv` +(black appears in `uv tool list` — a different install), and then removed +the real uv tool instead of the displayed venv path. Virtualenvs are +environments, not installations — the shell layer (capability.sh) already +skips them; detection here must too. + +Additionally, bulk_reconcile prompts from ThreadPool workers interleaved: +all confirmation prompts printed at once before any input was consumed. +""" + +from __future__ import annotations + +import os +import sys +import threading +import time +from pathlib import Path +from unittest.mock import patch + +import pytest + +skip_on_windows = pytest.mark.skipif( + sys.platform == "win32", + reason="Uses Unix-style paths and PATH separator (:)" +) + +from cli_audit.reconcile import ( # noqa: E402 + _confirm_removal, + clear_detection_cache, + detect_installations, +) + + +def _make_bin(bin_dir: Path, name: str) -> Path: + bin_dir.mkdir(parents=True, exist_ok=True) + binary = bin_dir / name + binary.write_text("#!/bin/sh\necho 'faketool 1.0.0'\n") + binary.chmod(0o755) + return binary + + +@skip_on_windows +class TestDetectSkipsVirtualenvs: + def _detect(self, monkeypatch, path_dirs: list[Path], tool: str): + clear_detection_cache() + monkeypatch.setenv("PATH", os.pathsep.join(str(d) for d in path_dirs)) + return detect_installations(tool) + + def test_dot_venv_bin_is_skipped(self, tmp_path, monkeypatch): + venv_bin = tmp_path / ".venv" / "bin" + _make_bin(venv_bin, "faketool_venv_a") + (tmp_path / ".venv" / "pyvenv.cfg").write_text("home = /usr\n") + plain_bin = tmp_path / "tools" / "bin" + plain = _make_bin(plain_bin, "faketool_venv_a") + + installs = self._detect(monkeypatch, [venv_bin, plain_bin], "faketool_venv_a") + paths = [i.path for i in installs] + assert str(plain) in paths + assert not any(".venv" in p for p in paths), ( + "virtualenv binaries are environments, not installations" + ) + + def test_pyvenv_cfg_detected_regardless_of_dir_name(self, tmp_path, monkeypatch): + # A venv named anything (not just venv/.venv) carries pyvenv.cfg + env_bin = tmp_path / "myproject-env" / "bin" + _make_bin(env_bin, "faketool_venv_b") + (tmp_path / "myproject-env" / "pyvenv.cfg").write_text("home = /usr\n") + + installs = self._detect(monkeypatch, [env_bin], "faketool_venv_b") + assert installs == [] + + def test_conda_env_bin_is_skipped_by_name(self, tmp_path, monkeypatch): + conda_bin = tmp_path / "miniconda3" / "envs" / "dev" / "bin" + _make_bin(conda_bin, "faketool_venv_c") + + installs = self._detect(monkeypatch, [conda_bin], "faketool_venv_c") + assert installs == [] + + def test_symlink_into_venv_is_skipped(self, tmp_path, monkeypatch): + venv_bin = tmp_path / ".venv" / "bin" + real = _make_bin(venv_bin, "faketool_venv_d") + (tmp_path / ".venv" / "pyvenv.cfg").write_text("home = /usr\n") + link_dir = tmp_path / "linkbin" + link_dir.mkdir() + (link_dir / "faketool_venv_d").symlink_to(real) + + installs = self._detect(monkeypatch, [link_dir], "faketool_venv_d") + assert installs == [] + + def test_non_venv_installs_still_detected(self, tmp_path, monkeypatch): + bin_a = tmp_path / "a" / "bin" + bin_b = tmp_path / "b" / "bin" + _make_bin(bin_a, "faketool_venv_e") + _make_bin(bin_b, "faketool_venv_e") + + installs = self._detect(monkeypatch, [bin_a, bin_b], "faketool_venv_e") + assert len(installs) == 2 + + +@skip_on_windows +class TestConfirmRemovalSerialized: + def test_concurrent_prompts_do_not_interleave(self): + """With bulk_reconcile's ThreadPool, every worker's prompt printed + before any input was read. Prompts must be serialized: while one + prompt awaits input, another thread's prompt must not appear.""" + from cli_audit.reconcile import Installation + + events: list[str] = [] + first_prompt_shown = threading.Event() + release_first_input = threading.Event() + + def fake_print(*args, **kwargs): + text = " ".join(str(a) for a in args) + if "About to remove" in text: + events.append(f"prompt:{threading.current_thread().name}") + first_prompt_shown.set() + + def fake_input(): + events.append(f"input:{threading.current_thread().name}") + release_first_input.wait(timeout=5) + return "n" + + inst = Installation( + tool="faketool", version="1.0", method="uv", + path="/opt/faketool", active=False, valid=True, + ) + + def worker(): + _confirm_removal("faketool", [inst]) + + with patch("cli_audit.reconcile.sys.stdin") as stdin_mock, \ + patch("builtins.print", side_effect=fake_print), \ + patch("builtins.input", side_effect=fake_input): + stdin_mock.isatty.return_value = True + t1 = threading.Thread(target=worker, name="w1") + t2 = threading.Thread(target=worker, name="w2") + t1.start() + first_prompt_shown.wait(timeout=5) + t2.start() + # give w2 a chance to (incorrectly) print its prompt while w1 + # still awaits input + time.sleep(0.3) + interleaved = [e for e in events if e.startswith("prompt")] + release_first_input.set() + t1.join(timeout=5) + t2.join(timeout=5) + + assert len(interleaved) == 1, ( + f"second prompt appeared while first awaited input: {events}" + ) + # both prompts eventually happened, strictly prompt->input->prompt->input + kinds = [e.split(":")[0] for e in events] + assert kinds == ["prompt", "input", "prompt", "input"], events