Fix reject of valid state in criteria compatibility check

news/91.bugfix.rst

@@ -0,0 +1,6 @@
+Some valid states that were previously rejected are now accepted. This affects
+states where multiple candidates for the same dependency conflict with each
+other. The ``information`` argument passed to
+``AbstractProvider.get_preference`` may now contain empty iterators. This has
+always been allowed by the method definition but it was previously not possible
+in practice.

src/resolvelib/resolvers.py

@@ -173,6 +173,31 @@ def _add_to_criteria(self, criteria, requirement, parent):
             raise RequirementsConflicted(criterion)
         criteria[identifier] = criterion
 
+    def _remove_information_from_criteria(self, criteria, parents):
+        """Remove information from parents of criteria.
+
+        Concretely, removes all values from each criterion's ``information``
+        field that have one of ``parents`` as provider of the requirement.
+
+        :param criteria: The criteria to update.
+        :param parents: Identifiers for which to remove information from all criteria.
+        """
+        if not parents:
+            return
+        for key, criterion in criteria.items():
+            criteria[key] = Criterion(
+                criterion.candidates,
+                [
+                    information
+                    for information in criterion.information
+                    if (
+                        information[1] is None
+                        or self._p.identify(information[1]) not in parents
+                    )
+                ],
+                criterion.incompatibilities,
+            )
+
     def _get_preference(self, name):
         return self._p.get_preference(
             identifier=name,
@@ -367,6 +392,11 @@ def resolve(self, requirements, max_rounds):
                 self._r.ending(state=self.state)
                 return self.state
 
+            # keep track of satisfied names to calculate diff after pinning
+            satisfied_names = set(self.state.criteria.keys()) - set(
+                unsatisfied_names
+            )
+
             # Choose the most preferred unpinned criterion to try.
             name = min(unsatisfied_names, key=self._get_preference)
             failure_causes = self._attempt_to_pin_criterion(name)
@@ -383,6 +413,17 @@ def resolve(self, requirements, max_rounds):
                 if not success:
                     raise ResolutionImpossible(self.state.backtrack_causes)
             else:
+                # discard as information sources any invalidated names
+                # (unsatisfied names that were previously satisfied)
+                newly_unsatisfied_names = {
+                    key
+                    for key, criterion in self.state.criteria.items()
+                    if key in satisfied_names
+                    and not self._is_current_pin_satisfying(key, criterion)
+                }
+                self._remove_information_from_criteria(
+                    self.state.criteria, newly_unsatisfied_names
+                )
                 # Pinning was successful. Push a new state to do another pin.
                 self._push_new_state()
 

src/resolvelib/resolvers.pyi

@@ -55,6 +55,18 @@ class ResolutionImpossible(ResolutionError, Generic[RT, CT]):
 class ResolutionTooDeep(ResolutionError):
     round_count: int
 
+# This should be a NamedTuple, but Python 3.6 has a bug that prevents it.
+# https://stackoverflow.com/a/50531189/1376863
+class State(tuple, Generic[RT, CT, KT]):
+    mapping: Mapping[KT, CT]
+    criteria: Mapping[KT, Criterion[RT, CT, KT]]
+    backtrack_causes: Collection[RequirementInformation[RT, CT]]
+
+class Resolution(Generic[RT, CT, KT]):
+    def resolve(
+        self, requirements: Iterable[RT], max_rounds: int
+    ) -> State[RT, CT, KT]: ...
+
 class Result(Generic[RT, CT, KT]):
     mapping: Mapping[KT, CT]
     graph: DirectedGraph[Optional[KT]]

tests/test_resolvers.py

@@ -1,4 +1,18 @@
+from typing import (
+    Any,
+    Iterable,
+    Iterator,
+    List,
+    Mapping,
+    Sequence,
+    Set,
+    Tuple,
+    Union,
+)
+
 import pytest
+from packaging.requirements import Requirement
+from packaging.version import Version
 
 from resolvelib import (
     AbstractProvider,
@@ -7,6 +21,12 @@
     ResolutionImpossible,
     Resolver,
 )
+from resolvelib.resolvers import (
+    Criterion,
+    RequirementInformation,
+    RequirementsConflicted,
+    Resolution,
+)
 
 
 def test_candidate_inconsistent_error():
@@ -143,3 +163,109 @@ def run_resolver(*args):
     backtracking_causes = run_resolver([("a", {1, 2}), ("b", {1})])
     exception_causes = run_resolver([("a", {2}), ("b", {1})])
     assert exception_causes == backtracking_causes
+
+
+def test_pin_conflict_with_self(monkeypatch, reporter):
+    # type: (Any, BaseReporter) -> None
+    """
+    Verify correct behavior of attempting to pin a candidate version that conflicts
+    with a previously pinned (now invalidated) version for that same candidate (#91).
+    """
+    Candidate = Tuple[
+        str, Version, Sequence[str]
+    ]  # name, version, requirements
+    all_candidates = {
+        "parent": [("parent", Version("1"), ["child<2"])],
+        "child": [
+            ("child", Version("2"), ["grandchild>=2"]),
+            ("child", Version("1"), ["grandchild<2"]),
+            ("child", Version("0.1"), ["grandchild"]),
+        ],
+        "grandchild": [
+            ("grandchild", Version("2"), []),
+            ("grandchild", Version("1"), []),
+        ],
+    }  # type: Mapping[str, Sequence[Candidate]]
+
+    class Provider(AbstractProvider):  # AbstractProvider[str, Candidate, str]
+        def identify(self, requirement_or_candidate):
+            # type: (Union[str, Candidate]) -> str
+            result = (
+                Requirement(requirement_or_candidate).name
+                if isinstance(requirement_or_candidate, str)
+                else requirement_or_candidate[0]
+            )
+            assert result in all_candidates, "unknown requirement_or_candidate"
+            return result
+
+        def get_preference(self, identifier, *args, **kwargs):
+            # type: (str, *object, **object) -> str
+            # prefer child over parent (alphabetically)
+            return identifier
+
+        def get_dependencies(self, candidate):
+            # type: (Candidate) -> Sequence[str]
+            return candidate[2]
+
+        def find_matches(
+            self,
+            identifier,  # type: str
+            requirements,  # type: Mapping[str, Iterator[str]]
+            incompatibilities,  # type: Mapping[str, Iterator[Candidate]]
+        ):
+            # type: (...) -> Iterator[Candidate]
+            return (
+                candidate
+                for candidate in all_candidates[identifier]
+                if all(
+                    self.is_satisfied_by(req, candidate)
+                    for req in requirements[identifier]
+                )
+                if candidate not in incompatibilities[identifier]
+            )
+
+        def is_satisfied_by(self, requirement, candidate):
+            # type: (str, Candidate) -> bool
+            return candidate[1] in Requirement(requirement).specifier
+
+    # patch Resolution._get_updated_criteria to collect rejected states
+    rejected_criteria = []  # type: List[Criterion]
+    get_updated_criteria_orig = (
+        Resolution._get_updated_criteria  # type: ignore[attr-defined]
+    )
+
+    def get_updated_criteria_patch(self, candidate):
+        try:
+            return get_updated_criteria_orig(self, candidate)
+        except RequirementsConflicted as e:
+            rejected_criteria.append(e.criterion)
+            raise
+
+    monkeypatch.setattr(
+        Resolution, "_get_updated_criteria", get_updated_criteria_patch
+    )
+
+    resolver = Resolver(
+        Provider(), reporter
+    )  # type: Resolver[str, Candidate, str]
+    result = resolver.resolve(["child", "parent"])
+
+    def get_child_versions(information):
+        # type: (Iterable[RequirementInformation[str, Candidate]]) -> Set[str]
+        return {
+            str(inf.parent[1])
+            for inf in information
+            if inf.parent is not None and inf.parent[0] == "child"
+        }
+
+    # verify that none of the rejected criteria are based on more than one candidate for
+    # child
+    assert not any(
+        len(get_child_versions(criterion.information)) > 1
+        for criterion in rejected_criteria
+    )
+
+    assert set(result.mapping) == {"parent", "child", "grandchild"}
+    assert result.mapping["parent"][1] == Version("1")
+    assert result.mapping["child"][1] == Version("1")
+    assert result.mapping["grandchild"][1] == Version("1")