Removes the usage of conda._vendor.toolz from constructor

constructor/conda_interface.py

@@ -28,9 +28,6 @@
     # Flatten VersionOrder.version, skip epoch, and keep only major and minor
     CONDA_MAJOR_MINOR = tuple(chain.from_iterable(_conda_version))[1:3]
 
-    from conda._vendor.toolz.itertoolz import (
-        concatv as _concatv, get as _get, groupby as _groupby,
-    )
     from conda.api import SubdirData # noqa
     from conda.base.context import (
         context as _conda_context, replace_context_default as _conda_replace_context_default,
@@ -56,7 +53,7 @@
     # used by fcp.py
     PackageCacheData = _PackageCacheData
     Solver, read_paths_json = _Solver, _read_paths_json
-    concatv, get, groupby, all_channel_urls = _concatv, _get, _groupby, _all_channel_urls
+    all_channel_urls = _all_channel_urls
     conda_context, env_vars = _conda_context, _env_vars
     conda_replace_context_default = _conda_replace_context_default
     download, PackageCacheRecord = _download, _PackageCacheRecord

constructor/fcp.py

@@ -12,13 +12,15 @@
 import json
 import os
 from os.path import isdir, isfile, join, splitext
+from itertools import groupby
+
 import sys
 import tempfile
 
 from constructor.utils import hash_files, filename_dist
 from .conda_interface import (PackageCacheData, PackageCacheRecord, Solver, SubdirData,
-                              VersionOrder, concatv, conda_context, conda_replace_context_default,
-                              download, env_vars, groupby, read_paths_json, all_channel_urls,
+                              VersionOrder, conda_context, conda_replace_context_default,
+                              download, env_vars, read_paths_json, all_channel_urls,
                               cc_platform)
 
 
@@ -44,11 +46,12 @@ def warn_menu_packages_missing(precs, menu_packages):
 
 
 def check_duplicates(precs):
-    groups = groupby(lambda x: x.name, precs)
-    for precs in groups.values():
-        if len(precs) > 1:
-            sys.exit("Error: '%s' listed multiple times: %s" %
-                     (precs[0].name, ', '.join(prec.fn for prec in precs)))
+    prec_groups = {key: tuple(value) for key, value in groupby(precs, lambda prec: prec.name)}
+
+    for name, precs in prec_groups.items():
+        filenames = sorted(prec.fn for prec in precs)
+        if len(filenames) > 1:
+            sys.exit(f"Error: {name} listed multiple times: {' , '.join(filenames)}")
 
 
 def exclude_packages(precs, exclude=()):
@@ -57,13 +60,11 @@ def exclude_packages(precs, exclude=()):
             if bad_char in name:
                 sys.exit("Error: did not expect '%s' in package name: %s" % (bad_char, name))
 
-    groups = groupby(lambda x: x.name in exclude, precs)
-    excluded_precs = groups.get(True, [])
-    accepted_precs = groups.get(False, [])
-    for name in exclude:
-        if not any(prec.name == name for prec in excluded_precs):
-            sys.exit("Error: no package named '%s' to remove" % name)
-    return accepted_precs
+    unknown_precs = set(exclude).difference(prec.name for prec in precs)
+    if unknown_precs:
+        sys.exit(f"Error: no package(s) named {', '.join(unknown_precs)} to remove")
+
+    return [prec for prec in precs if prec.name not in exclude]
 
 
 def _find_out_of_date_precs(precs, channel_urls, platform):
@@ -245,15 +246,12 @@ def _main(name, version, download_dir, platform, channel_urls=(), channels_remap
           transmute_file_type=''):
     # Add python to specs, since all installers need a python interpreter. In the future we'll
     # probably want to add conda too.
-    specs = list(concatv(specs, ("python",)))
+    specs = (*specs, "python")
     if verbose:
         print("specs: %r" % specs)
 
     # Append channels_remap srcs to channel_urls
-    channel_urls = tuple(concatv(
-        channel_urls,
-        (x['src'] for x in channels_remap),
-    ))
+    channel_urls = (*channel_urls, *(x['src'] for x in channels_remap))
 
     if environment_file or environment:
         # set conda to be the user's conda (what is in the environment)

tests/test_fcp.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from contextlib import nullcontext
+
+import pytest
+
+from constructor.fcp import check_duplicates, exclude_packages
+
+
+class GenericObject:
+    """We use this for testing the check_duplicates function"""
+    def __init__(self, name):
+        self.name = name
+        self.fn = 'filename.txt'
+
+    def __eq__(self, other):
+        return self.name == other.name
+
+
+@pytest.mark.parametrize('values,expected_fails', (
+    (
+        (GenericObject('NameOne'), GenericObject('NameTwo'), GenericObject('NameTwo')), 1
+    ),
+    (
+        (GenericObject('NameOne'), GenericObject('NameTwo')), 0
+    ),
+    (
+        (GenericObject('NameOne'), GenericObject('NameTwo'), GenericObject('NameThree')), 0
+    ),
+))
+def test_check_duplicates(values: tuple[..., GenericObject], expected_fails: int):
+    if expected_fails:
+        context = pytest.raises(SystemExit)
+    else:
+        context = nullcontext()
+
+    with context:
+        check_duplicates(values)
+
+
+@pytest.mark.parametrize('values,expected_value', (
+    (
+        (
+            (GenericObject('NameOne'), GenericObject('NameTwo'), GenericObject('NameThree')),
+            ('NameThree',)
+        ),
+        [GenericObject('NameOne'), GenericObject('NameTwo')]
+    ),
+    (
+        (
+            (GenericObject('NameOne'), GenericObject('NameTwo'), GenericObject('NameThree')),
+            ('Not in list',)
+        ),
+        False
+    ),
+))
+def test_exclude_packages(values: tuple[..., GenericObject], expected_value):
+    if expected_value is False:
+        context = pytest.raises(SystemExit)
+    else:
+        context = nullcontext()
+
+    with context:
+        packages = exclude_packages(*values)
+        assert packages == expected_value