Protect conda plugins from being deleted during reset (#61)

* Protect conda plugins from being deleted

* let subprocess print to stdout/err by default

* make PackageInfo.from_record() compatible with both PrefixRecord and PackageCacheRecord

* merge the two class methods

* debug

* do not return early

* save iterator as list to avoid exhausting it

* verbose pytest

* dedent

Author: jaimergp

.github/workflows/tests.yml

@@ -53,7 +53,12 @@ jobs:
           echo "channels: [${{ matrix.channel }}]" > .pixi/envs/test-py${{ matrix.python-version }}/.condarc
           pixi run --environment test-py${{ matrix.python-version }} python -m conda info
       - name: Run tests
-        run: pixi run --environment test-py${{ matrix.python-version }} test --basetemp=${{ runner.os == 'Windows' && 'D:\temp' || runner.temp }}
+        run: >
+          pixi run
+          --environment test-py${{ matrix.python-version }}
+          test
+          -vv
+          --basetemp=${{ runner.os == 'Windows' && 'D:\temp' || runner.temp }}
 
   build-conda:
     name: Build conda package (${{ matrix.os }})

conda_self/exceptions.py

@@ -19,5 +19,7 @@ def __init__(self, specs: list[str]):
 
 
 class NoDistInfoDirFound(CondaError):
-    def __init__(self, path: str | Path):
-        super().__init__(f"No *.dist-info directories found in {path}.")
+    def __init__(self, package_name: str, path: str | Path):
+        super().__init__(
+            f"No *.dist-info directories found for '{package_name}' in '{path}'."
+        )

conda_self/package_info.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
 import configparser
+import os
+import sys
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from conda.models.records import PrefixRecord
+
 from .exceptions import NoDistInfoDirFound
 
 if TYPE_CHECKING:
@@ -18,23 +22,34 @@ class CaseSensitiveConfigParser(configparser.ConfigParser):
 
 class PackageInfo:
     def __init__(self, dist_info_path: Path):
-        """Describe the dist-info for a conda package"""
+        """Describe the dist-info for a Python package installed as a conda package"""
         self.dist_info_path = dist_info_path
 
     @classmethod
-    def from_record(cls, record: PackageCacheRecord) -> list[PackageInfo]:
-        return cls.from_conda_extracted_package_path(record.extracted_package_dir)
-
-    @classmethod
-    def from_conda_extracted_package_path(cls, path: str | Path) -> list[PackageInfo]:
-        """Create a PackageInfo object given the path to an extracted conda package"""
-        path = Path(path)
-        matching_paths = [
-            p for p in path.rglob("**/*site-packages/*dist-info*") if p.is_dir()
-        ]
-        if len(matching_paths) == 0:
-            raise NoDistInfoDirFound(path)
-        return [cls(matching_path) for matching_path in matching_paths]
+    def from_record(
+        cls, record: PrefixRecord | PackageCacheRecord
+    ) -> list[PackageInfo]:
+        if not (paths := getattr(record, "files", None)):
+            try:
+                paths = (
+                    Path(record.extracted_package_dir, "info/files")
+                    .read_text()
+                    .splitlines()
+                )
+            except FileNotFoundError:
+                # missing info/files -> empty package
+                paths = []
+        dist_infos = set()
+        for path in paths:
+            if (maybe_dist_info := os.path.dirname(path)).endswith(".dist-info"):
+                dist_infos.add(maybe_dist_info)
+        if isinstance(record, PrefixRecord):
+            basedir = sys.prefix
+        else:
+            basedir = record.extracted_package_dir
+        if not dist_infos:
+            raise NoDistInfoDirFound(record.name, basedir)
+        return [cls(Path(basedir, p)) for p in dist_infos]
 
     def entry_points(self) -> dict[str, dict[str, str]]:
         """Get the entry points for a package.

conda_self/query.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import sys
+from contextlib import suppress
 from typing import TYPE_CHECKING
 
 from conda.base.context import context
@@ -17,6 +18,8 @@
 from conda.models.version import VersionOrder
 
 from .constants import PERMANENT_PACKAGES
+from .exceptions import NoDistInfoDirFound
+from .package_info import PackageInfo
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
@@ -66,11 +69,19 @@ def permanent_dependencies() -> set[str]:
     # In some dev environments, conda-self is installed as a PyPI package
     # and does not have its conda-meta/conda-self-*.json entry, which makes it
     # invisible to PrefixData()... unless we enable interoperability.
-    installed = PrefixData(sys.prefix, interoperability=True)
-    prefix_graph = PrefixGraph(installed.iter_records())
+    installed = list(PrefixData(sys.prefix, interoperability=True).iter_records())
+    prefix_graph = PrefixGraph(installed)
+
+    protect = [*PERMANENT_PACKAGES]
+    for record in installed:
+        with suppress(NoDistInfoDirFound):
+            for pkg_info in PackageInfo.from_record(record):
+                if "conda" in pkg_info.entry_points():
+                    protect.append(record.name)
+                    break
 
     packages = []
-    for pkg in PERMANENT_PACKAGES:
+    for pkg in dict.fromkeys(protect):
         node = next((rec for rec in prefix_graph.records if rec.name == pkg), None)
         if node:
             packages.append(node.name)

conda_self/testing.py

@@ -14,7 +14,7 @@
     from conda.models.records import PackageRecord
 
 
-def conda_cli_subprocess(prefix: str | Path, *args) -> CompletedProcess:
+def conda_cli_subprocess(prefix: str | Path, *args, **kwargs) -> CompletedProcess:
     prefix = Path(prefix)
     if os.name == "nt":
         python = prefix / "python.exe"
@@ -23,9 +23,8 @@ def conda_cli_subprocess(prefix: str | Path, *args) -> CompletedProcess:
 
     return run(
         [python, "-m", "conda", *args],
-        check=True,
-        capture_output=True,
-        text=True,
+        check=kwargs.pop("check", True),
+        **kwargs,
     )
 
 

tests/test_cli_reset.py

@@ -9,7 +9,8 @@ def test_help(conda_cli):
 
 
 def test_reset(conda_cli, tmp_env: TmpEnvFixture):
-    with tmp_env("conda", "conda-self") as prefix:
+    # Adding conda-index too to test that non-default plugins are kept
+    with tmp_env("conda", "conda-self", "conda-index") as prefix:
         assert not is_installed(prefix, "numpy")
 
         conda_cli("install", "numpy", "--yes", "--prefix", prefix)
@@ -20,5 +21,7 @@ def test_reset(conda_cli, tmp_env: TmpEnvFixture):
         assert is_installed(prefix, "conda")
         # make sure conda-self didn't remove itself
         assert is_installed(prefix, "conda-self")
+        # make sure conda-self didn't remove a non-default conda plugin
+        assert is_installed(prefix, "conda-index")
         # but numpy should be gone
         assert not is_installed(prefix, "numpy")