Test filter function

lib/spack/spack/binary_distribution.py

@@ -1346,7 +1346,7 @@ def fail(self) -> None:
         tty.info(f"{self.pre}Failed to push {self.pretty_spec}")
 
 
-def _filter_specs(specs: List[spack.spec.Spec], exclude: List[str], include: List[str]):
+def filter_specs(specs: List[spack.spec.Spec], exclude: List[str], include: List[str]):
     """
     Determine the intersection of include/exclude filters
     Tie goes to keeping
@@ -1414,7 +1414,7 @@ def _url_push(
     if not specs_to_upload:
         return skipped, errors
 
-    filter, filtrate = _filter_specs(specs_to_upload, exclusions, inclusions)
+    filter, filtrate = filter_specs(specs_to_upload, exclusions, inclusions)
 
     skipped.extend(filtrate)
     specs_to_upload = filter
@@ -1724,7 +1724,7 @@ def _oci_push(
     if not blobs_to_upload:
         return skipped, base_images, checksums, []
 
-    filter, filtrate = _filter_specs(blobs_to_upload, exclusions, inclusions)
+    filter, filtrate = filter_specs(blobs_to_upload, exclusions, inclusions)
 
     skipped.extend(filtrate)
     blobs_to_upload = filter

lib/spack/spack/test/bindist.py

@@ -59,6 +59,32 @@
 legacy_mirror_dir = os.path.join(test_path, "data", "mirrors", "legacy_yaml")
 
 
+INPUT_SPEC_STRS = ["foo@main", "foo@main dev_path=/tmp", "foo@2.1.3"]
+
+
+@pytest.mark.parametrize(
+    "include,exclude,gold",
+    [
+        ([], [], [0, 1, 2]),
+        (["dev_path=*", "@main"], [], [0, 1, 2]),
+        ([], ["dev_path=*", "@main"], [2]),
+        (["dev_path=*"], ["@main"], [1, 2]),
+    ],
+)
+def test_filter_specs(include, exclude, gold):
+    input_specs = [spack.spec.Spec(s) for s in INPUT_SPEC_STRS]
+    filter, filtrate = bindist.filter_specs(input_specs, exclude, include)
+
+    assert filter is not None
+    assert filtrate is not None
+
+    # lossless
+    assert (set(filter) | set(filtrate)) == set(input_specs)
+
+    for i in gold:
+        assert input_specs[i] in filter
+
+
 @pytest.fixture(scope="function")
 def cache_directory(tmpdir):
     fetch_cache_dir = tmpdir.ensure("fetch_cache", dir=True)