Split --snapshot installer into installer-exact and installer-updated

conda_self/health_checks/base_protection.py

@@ -75,7 +75,7 @@ def fix(prefix: str, args: Namespace, confirm: ConfirmCallback) -> int:
     from conda.misc import clone_env
     from conda.models.environment import Environment
 
-    from ..constants import RESET_FILE_BASE_PROTECTION
+    from ..constants import RESET_FILE_BASE_PROTECTION, RESET_FILE_INSTALLER
     from ..query import permanent_dependencies
     from ..reset import reset
 
@@ -98,8 +98,10 @@ def fix(prefix: str, args: Namespace, confirm: ConfirmCallback) -> int:
         )
     confirm("Proceed?")
 
-    # Get packages to keep in base
-    uninstallable_packages = permanent_dependencies()
+    # Prefer the installer snapshot for resetting base so that
+    # installer-provided packages (e.g. mamba in Miniforge) are preserved.
+    installer_snapshot = base_prefix / "conda-meta" / RESET_FILE_INSTALLER
+    use_snapshot = installer_snapshot.exists()
 
     # Check destination environment
     dest_prefix_data = PrefixData.from_name(default_env)
@@ -136,7 +138,10 @@ def fix(prefix: str, args: Namespace, confirm: ConfirmCallback) -> int:
             verbose=False,
             quiet=True,
         )
-        reset(uninstallable_packages=uninstallable_packages)
+        if use_snapshot:
+            reset(snapshot=installer_snapshot)
+        else:
+            reset(uninstallable_packages=permanent_dependencies())
 
     # Freeze base
     try:

tests/test_health_check_base_protection.py

@@ -9,19 +9,47 @@
 import pytest
 from conda.base.constants import PREFIX_FROZEN_FILE
 from conda.core.prefix_data import PrefixData
+from conda.exceptions import CondaValueError
 
+from conda_self.constants import RESET_FILE_INSTALLER
 from conda_self.health_checks import base_protection
+from conda_self.plugin import conda_health_checks
 
 if TYPE_CHECKING:
     from pathlib import Path
 
     from pytest import CaptureFixture, MonkeyPatch
 
 
+class _FakePrefixData:
+    def __init__(self, tmp_path: Path):
+        self.prefix_path = tmp_path / "envs" / "default"
+
+    def is_environment(self) -> bool:
+        return False
+
+    def exists(self) -> bool:
+        return False
+
+
+class _FakeEnvironment:
+    external_packages: list = []
+
+
+class _FakeConfigFile:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        pass
+
+    def set_key(self, key, value):
+        pass
+
+
 @pytest.fixture
 def fake_base_env(tmp_path: Path, monkeypatch: MonkeyPatch) -> Path:
     """Create a fake base environment by patching sys.prefix."""
-    # Create conda-meta directory to make it look like a conda env
     conda_meta = tmp_path / "conda-meta"
     conda_meta.mkdir()
     monkeypatch.setattr(sys, "prefix", str(tmp_path))
@@ -36,138 +64,190 @@ def protected_base_env(fake_base_env: Path) -> Path:
     return fake_base_env
 
 
-# -- is_base_environment tests --
-
-
-def test_is_base_environment_returns_true_for_base():
-    """Returns True when prefix matches sys.prefix."""
-    assert base_protection.is_base_environment(sys.prefix) is True
-
-
-def test_is_base_environment_returns_false_for_other_prefix(tmp_path: Path):
-    """Returns False when prefix doesn't match sys.prefix."""
-    assert base_protection.is_base_environment(str(tmp_path)) is False
-
-
-# -- is_base_protected tests --
+@pytest.fixture
+def reset_calls():
+    return []
 
 
-def test_is_base_protected_returns_true_when_frozen(protected_base_env: Path):
-    """Returns True when PREFIX_FROZEN_FILE exists."""
-    # Clear the PrefixData cache to pick up our patched sys.prefix
-    PrefixData._cache_.clear()
-    assert base_protection.is_base_protected() is True
+@pytest.fixture
+def perm_deps_calls():
+    return []
 
 
-def test_is_base_protected_returns_false_when_not_frozen(fake_base_env: Path):
-    """Returns False when PREFIX_FROZEN_FILE doesn't exist."""
+@pytest.fixture
+def fixable_base_env(
+    fake_base_env: Path,
+    monkeypatch: MonkeyPatch,
+    reset_calls: list,
+    perm_deps_calls: list,
+):
+    """A fake base env wired with lightweight stubs for fix() to run end-to-end."""
     PrefixData._cache_.clear()
-    assert base_protection.is_base_protected() is False
-
-
-# -- check action tests --
 
+    def fake_reset(**kwargs):
+        reset_calls.append(kwargs)
+
+    def fake_perm_deps(**kwargs):
+        perm_deps_calls.append(kwargs)
+        return {"conda", "conda-self"}
+
+    def fake_get_exporter(fmt):
+        raise CondaValueError("no exporter")
+
+    monkeypatch.setattr("conda.base.context.context.quiet", True, raising=False)
+    monkeypatch.setattr(
+        "conda.base.context.context.plugin_manager.get_environment_exporter_by_format",
+        fake_get_exporter,
+    )
+    monkeypatch.setattr("conda_self.reset.reset", fake_reset)
+    monkeypatch.setattr("conda.misc.clone_env", lambda *a, **kw: None)
+    monkeypatch.setattr(
+        "conda.models.environment.Environment.from_prefix",
+        lambda *a, **kw: _FakeEnvironment(),
+    )
+    monkeypatch.setattr("conda_self.query.permanent_dependencies", fake_perm_deps)
+    monkeypatch.setattr(
+        PrefixData, "from_name", lambda *a, **kw: _FakePrefixData(fake_base_env)
+    )
+    monkeypatch.setattr(
+        "conda.cli.condarc.ConfigurationFile.from_user_condarc",
+        _FakeConfigFile,
+    )
 
-def test_check_skips_non_base_environment(tmp_path: Path, capsys: CaptureFixture):
-    """Prints skip message when not on base environment."""
-    base_protection.check(str(tmp_path), False)
-
-    captured = capsys.readouterr()
-    assert "Skipping" in captured.out
-    assert "not running on base environment" in captured.out
+    return fake_base_env
 
 
-def test_check_reports_protected_base(protected_base_env: Path, capsys: CaptureFixture):
-    """Reports OK when base is protected."""
+@pytest.mark.parametrize(
+    "use_base, expected",
+    [
+        (True, True),
+        (False, False),
+    ],
+    ids=["base", "other"],
+)
+def test_is_base_environment(tmp_path: Path, use_base: bool, expected: bool):
+    prefix = sys.prefix if use_base else str(tmp_path)
+    assert base_protection.is_base_environment(prefix) is expected
+
+
+@pytest.mark.parametrize(
+    "frozen, expected",
+    [
+        (True, True),
+        (False, False),
+    ],
+    ids=["frozen", "not-frozen"],
+)
+def test_is_base_protected(fake_base_env: Path, frozen: bool, expected: bool):
+    if frozen:
+        (fake_base_env / PREFIX_FROZEN_FILE).write_text("{}")
     PrefixData._cache_.clear()
-    base_protection.check(str(protected_base_env), False)
-
-    captured = capsys.readouterr()
-    assert "protected" in captured.out.lower()
-    assert "not protected" not in captured.out.lower()
-
-
-def test_check_reports_unprotected_base(fake_base_env: Path, capsys: CaptureFixture):
-    """Reports X when base is not protected."""
+    assert base_protection.is_base_protected() is expected
+
+
+@pytest.mark.parametrize(
+    "env_fixture, expected_output, unexpected_output",
+    [
+        ("tmp_path", "skipping", None),
+        ("protected_base_env", "protected", "not protected"),
+        ("fake_base_env", "not protected", None),
+    ],
+    ids=["non-base", "protected", "unprotected"],
+)
+def test_check(
+    env_fixture: str,
+    expected_output: str,
+    unexpected_output: str | None,
+    request: pytest.FixtureRequest,
+    capsys: CaptureFixture,
+):
+    prefix = str(request.getfixturevalue(env_fixture))
     PrefixData._cache_.clear()
-    base_protection.check(str(fake_base_env), False)
-
-    captured = capsys.readouterr()
-    assert "not protected" in captured.out.lower()
-    assert "conda doctor --fix" in captured.out
-
-
-# -- fix fixer tests --
-
-
-def test_fix_skips_non_base_environment(tmp_path: Path, capsys: CaptureFixture):
-    """Skips when not on base environment."""
-    args = Namespace()
-    confirm_called = []
-
-    def confirm(msg: str) -> None:
-        confirm_called.append(msg)
-
-    result = base_protection.fix(str(tmp_path), args, confirm)
+    base_protection.check(prefix, False)
 
-    assert result == 0
     captured = capsys.readouterr()
-    assert "Skipping" in captured.out
-    assert len(confirm_called) == 0
-
-
-def test_fix_skips_already_protected(protected_base_env: Path, capsys: CaptureFixture):
-    """Returns early when base is already protected."""
-    args = Namespace()
-    confirm_called = []
-
-    def confirm(msg: str) -> None:
-        confirm_called.append(msg)
+    assert expected_output in captured.out.lower()
+    if unexpected_output:
+        assert unexpected_output not in captured.out.lower()
+
+
+@pytest.mark.parametrize(
+    "env_fixture, expected_output",
+    [
+        ("tmp_path", "Skipping"),
+        ("protected_base_env", "already protected"),
+    ],
+    ids=["non-base", "already-protected"],
+)
+def test_fix_skips(
+    env_fixture: str,
+    expected_output: str,
+    request: pytest.FixtureRequest,
+    capsys: CaptureFixture,
+):
+    prefix = str(request.getfixturevalue(env_fixture))
+    confirm_called: list[str] = []
 
     PrefixData._cache_.clear()
-    result = base_protection.fix(str(protected_base_env), args, confirm)
+    result = base_protection.fix(prefix, Namespace(), confirm_called.append)
 
     assert result == 0
-    captured = capsys.readouterr()
-    assert "already protected" in captured.out
-    assert len(confirm_called) == 0
+    assert expected_output in capsys.readouterr().out
+    assert confirm_called == []
 
 
 def test_fix_calls_confirm_callback(fake_base_env: Path):
-    """Calls confirm callback when proceeding with fix."""
-    args = Namespace()
-    confirm_called = []
+    confirm_called: list[str] = []
 
     class UserCancelled(Exception):
-        """Raised when user cancels."""
+        pass
 
     def confirm(msg: str) -> None:
         confirm_called.append(msg)
-        # Simulate user cancellation to stop execution after first confirm
         raise UserCancelled()
 
     PrefixData._cache_.clear()
     with pytest.raises(UserCancelled):
-        base_protection.fix(str(fake_base_env), args, confirm)
+        base_protection.fix(str(fake_base_env), Namespace(), confirm)
 
     assert confirm_called == ["Proceed?"]
 
 
-# -- plugin registration tests --
+@pytest.mark.parametrize(
+    "create_snapshot",
+    [True, False],
+    ids=["with-installer-snapshot", "without-installer-snapshot"],
+)
+def test_fix_reset_strategy(
+    fixable_base_env: Path,
+    reset_calls: list,
+    perm_deps_calls: list,
+    create_snapshot: bool,
+):
+    if create_snapshot:
+        snapshot = fixable_base_env / "conda-meta" / RESET_FILE_INSTALLER
+        snapshot.write_text("@EXPLICIT\n")
+
+    base_protection.fix(str(fixable_base_env), Namespace(), lambda msg: None)
+
+    assert len(reset_calls) == 1
+    if create_snapshot:
+        assert reset_calls[0]["snapshot"] == snapshot
+        assert "uninstallable_packages" not in reset_calls[0]
+        assert perm_deps_calls == []
+    else:
+        assert reset_calls[0]["uninstallable_packages"] == {"conda", "conda-self"}
+        assert "snapshot" not in reset_calls[0]
+        assert len(perm_deps_calls) == 1
 
 
 def test_health_check_registered():
-    """Verify the health check is registered correctly."""
-
-    from conda_self.plugin import conda_health_checks
-
     health_checks = list(conda_health_checks())
 
     assert len(health_checks) == 1
     hc = health_checks[0]
     assert hc.name == "base-protection"
     assert hc.action == base_protection.check
-
     assert hc.fixer == base_protection.fix
     assert hc.summary is not None
     assert hc.fix is not None