feat: Add allowlist_file option to yaml migrators

conda_forge_tick/migrators/arch.py

@@ -1,5 +1,4 @@
 import copy
-import os
 from textwrap import dedent
 from typing import Any, Collection, Literal, Optional, Sequence
 
@@ -9,7 +8,13 @@
 from conda_forge_tick.make_graph import (
     get_deps_from_outputs_lut,
 )
-from conda_forge_tick.migrators.core import GraphMigrator, MiniMigrator, get_outputs_lut
+from conda_forge_tick.migrators.core import (
+    GraphMigrator,
+    MiniMigrator,
+    cut_graph_to_target_packages,
+    get_outputs_lut,
+    load_target_packages,
+)
 from conda_forge_tick.migrators_types import AttrsTypedDict, MigrationUidTypedDict
 from conda_forge_tick.os_utils import pushd
 from conda_forge_tick.utils import (
@@ -21,22 +26,6 @@
 
 from .migration_yaml import all_noarch
 
-MIGRATION_SUPPORT_DIRS = [
-    os.path.join(
-        os.environ["CONDA_PREFIX"],
-        "share",
-        "conda-forge",
-        "migration_support",
-    ),
-    # Deprecated
-    os.path.join(
-        os.environ["CONDA_PREFIX"],
-        "share",
-        "conda-forge",
-        "migrations",
-    ),
-]
-
 
 def _filter_excluded_deps(graph, excluded_dependencies):
     """Filter out excluded dependencies from the graph.
@@ -55,22 +44,6 @@ def _filter_excluded_deps(graph, excluded_dependencies):
     graph.remove_edges_from(nx.selfloop_edges(graph))
 
 
-def _cut_to_target_packages(graph, target_packages):
-    """Cut the graph to only the target packages.
-
-    **operates in place**
-    """
-    packages = target_packages.copy()
-    for target in target_packages:
-        if target in graph.nodes:
-            packages.update(nx.ancestors(graph, target))
-    for node in list(graph.nodes.keys()):
-        if node not in packages:
-            pluck(graph, node)
-    # post-plucking cleanup
-    graph.remove_edges_from(nx.selfloop_edges(graph))
-
-
 def _filter_stubby_and_ignored_nodes(graph, ignored_packages):
     """Remove any stub packages and ignored packages from the graph.
 
@@ -124,16 +97,7 @@ def __init__(
         if total_graph is not None:
             if target_packages is None:
                 # We are constraining the scope of this migrator
-                fname = None
-                for d in MIGRATION_SUPPORT_DIRS:
-                    fname = os.path.join(d, "arch_rebuild.txt")
-                    if os.path.exists(fname):
-                        break
-                assert fname is not None, (
-                    "Could not find arch_rebuild.txt in migration support dirs"
-                )
-                with open(fname) as f:
-                    target_packages = set(f.read().split())
+                target_packages = load_target_packages("arch_rebuild.txt")
 
             outputs_lut = get_outputs_lut(total_graph, graph, effective_graph)
 
@@ -155,7 +119,7 @@ def __init__(
             target_packages = set(target_packages)
             if target_packages:
                 target_packages.add("python")  # hack that is ~harmless?
-                _cut_to_target_packages(total_graph, target_packages)
+                cut_graph_to_target_packages(total_graph, target_packages)
 
             # filter out stub packages and ignored packages
             _filter_stubby_and_ignored_nodes(total_graph, self.ignored_packages)
@@ -264,15 +228,7 @@ def __init__(
         if total_graph is not None:
             if target_packages is None:
                 # We are constraining the scope of this migrator
-                fname = None
-                for d in MIGRATION_SUPPORT_DIRS:
-                    fname = os.path.join(d, self.pkg_list_filename)
-                    if os.path.exists(fname):
-                        break
-
-                with open(fname) as f:
-                    target_packages = set(f.read().split())
-
+                target_packages = load_target_packages(self.pkg_list_filename)
             outputs_lut = get_outputs_lut(total_graph, graph, effective_graph)
 
             # rebuild the graph to only use edges from the arch requirements
@@ -315,7 +271,7 @@ def __init__(
             # filter the graph down to the target packages
             if target_packages:
                 target_packages.add("python")  # hack that is ~harmless?
-                _cut_to_target_packages(total_graph, target_packages)
+                cut_graph_to_target_packages(total_graph, target_packages)
 
             # filter out stub packages and ignored packages
             _filter_stubby_and_ignored_nodes(total_graph, self.ignored_packages)

conda_forge_tick/migrators/core.py

@@ -4,6 +4,7 @@
 import copy
 import logging
 import math
+import os
 import re
 import secrets
 import time
@@ -29,12 +30,52 @@
 if typing.TYPE_CHECKING:
     from conda_forge_tick.utils import JsonFriendly
 
+MIGRATION_SUPPORT_DIRS = [
+    Path(os.environ["CONDA_PREFIX"]) / "share" / "conda-forge" / "migration_support",
+    # Deprecated
+    Path(os.environ["CONDA_PREFIX"]) / "share" / "conda-forge" / "migrations",
+]
+
 
 logger = logging.getLogger(__name__)
 
 RNG = secrets.SystemRandom()
 
 
+def load_target_packages(package_list_file: str) -> Set[str]:
+    # We are constraining the scope of this migrator
+    fname = None
+    for d in MIGRATION_SUPPORT_DIRS:
+        fname = d / package_list_file
+        if fname.exists():
+            break
+
+    assert fname is not None, (
+        f"Could not find {package_list_file} in migration support dirs"
+    )
+
+    with fname.open() as f:
+        target_packages = set(f.read().split())
+
+    return target_packages
+
+
+def cut_graph_to_target_packages(graph, target_packages):
+    """Cut the graph to only the target packages.
+
+    **operates in place**
+    """
+    packages = target_packages.copy()
+    for target in target_packages:
+        if target in graph.nodes:
+            packages.update(nx.ancestors(graph, target))
+    for node in list(graph.nodes.keys()):
+        if node not in packages:
+            pluck(graph, node)
+    # post-plucking cleanup
+    graph.remove_edges_from(nx.selfloop_edges(graph))
+
+
 def skip_migrator_due_to_schema(
     attrs: "AttrsTypedDict", allowed_schema_versions: List[int]
 ) -> bool:

conda_forge_tick/migrators/migration_yaml.py

@@ -16,7 +16,9 @@
     GraphMigrator,
     Migrator,
     MiniMigrator,
+    cut_graph_to_target_packages,
     get_outputs_lut,
+    load_target_packages,
 )
 from conda_forge_tick.os_utils import pushd
 from conda_forge_tick.utils import (
@@ -188,8 +190,13 @@ def __init__(
         force_pr_after_solver_attempts=10,
         longterm=False,
         paused=False,
+        allowlist_file: Optional[str] = None,
         **kwargs: Any,
     ):
+        if whitelist_file is not None:
+            target_packages = load_target_packages(whitelist_file)
+            cut_graph_to_target_packages(total_graph, target_packages)
+
         if not hasattr(self, "_init_args"):
             self._init_args = [yaml_contents, name]