from __future__ import annotations from collections import namedtuple from typing import TYPE_CHECKING, Any, Iterator, Sequence, Tuple import pytest from packaging.requirements import Requirement from packaging.version import Version from resolvelib import ( AbstractProvider, BaseReporter, InconsistentCandidate, ResolutionImpossible, ) from resolvelib.resolvers import ( RequirementsConflicted, Resolution, Resolver, ) if TYPE_CHECKING: from typing import Iterable, Mapping from resolvelib.resolvers import ( Criterion, RequirementInformation, ) def test_candidate_inconsistent_error(): requirement = "foo" candidate = "bar" class Provider(AbstractProvider): def __init__(self, requirement, candidate): self.requirement = requirement self.candidate = candidate def identify(self, requirement_or_candidate): assert requirement_or_candidate is self.requirement return requirement_or_candidate def get_preference(self, **_): return 0 def get_dependencies(self, **_): return [] def find_matches(self, identifier, requirements, incompatibilities): assert list(requirements[identifier]) == [self.requirement] assert next(incompatibilities[identifier], None) is None return [self.candidate] def is_satisfied_by(self, requirement, candidate): assert requirement is self.requirement assert candidate is self.candidate return False resolver = Resolver(Provider(requirement, candidate), BaseReporter()) with pytest.raises(InconsistentCandidate) as ctx: resolver.resolve([requirement]) assert str(ctx.value) == "Provided candidate 'bar' does not satisfy 'foo'" assert ctx.value.candidate is candidate assert list(ctx.value.criterion.iter_requirement()) == [requirement] @pytest.mark.parametrize("specifiers", [["1", "12"], ["12", "1"]]) def test_candidate_depends_on_requirements_of_same_identifier(specifiers): # This test ensures if a candidate has multiple dependencies under the same # identifier, all dependencies of that identifier are correctly pulled in. # The parametrization ensures both requirement ordering work. # Parent depends on child twice, one allows v2, the other does not. # Each candidate is a 3-tuple (name, version, dependencies). # Each requirement is a 2-tuple (name, allowed_versions). # Candidate v2 is in from so it is preferred when both are allowed. all_candidates = { "parent": [("parent", "1", [("child", s) for s in specifiers])], "child": [("child", "2", []), ("child", "1", [])], } class Provider(AbstractProvider): def identify(self, requirement_or_candidate): return requirement_or_candidate[0] def get_preference(self, **_): return 0 def get_dependencies(self, candidate): return candidate[2] def find_matches(self, identifier, requirements, incompatibilities): assert not list(incompatibilities[identifier]) return ( candidate for candidate in all_candidates[identifier] if all(candidate[1] in r[1] for r in requirements[identifier]) ) def is_satisfied_by(self, requirement, candidate): return candidate[1] in requirement[1] # Now when resolved, both requirements to child specified by parent should # be pulled, and the resolver should choose v1, not v2 (happens if the # v1-only requirement is dropped). resolver = Resolver(Provider(), BaseReporter()) result = resolver.resolve([("parent", {"1"})]) assert set(result.mapping) == {"parent", "child"} assert result.mapping["child"] == ("child", "1", []) def test_resolving_conflicts(): Candidate = namedtuple( "Candidate", ["name", "version", "requirements"] ) # name, version, requirements _Requirement = namedtuple("Requirement", ["name", "versions"]) # name, versions a1 = Candidate("a", 1, [_Requirement("q", {1})]) a2 = Candidate("a", 2, [_Requirement("q", {2})]) b = Candidate("b", 1, [_Requirement("q", {1})]) q1 = Candidate("q", 1, []) q2 = Candidate("q", 2, []) all_candidates = { "a": [a1, a2], "b": [b], "q": [q1, q2], } class Reporter(BaseReporter): def __init__(self): self.backtracking_causes = None def resolving_conflicts(self, causes): self.backtracking_causes = causes class Provider(AbstractProvider): def identify(self, requirement_or_candidate): return requirement_or_candidate[0] def get_preference(self, **_): return 0 def get_dependencies(self, candidate): return candidate.requirements def find_matches(self, identifier, requirements, incompatibilities): bad_versions = {c.version for c in incompatibilities[identifier]} candidates = [ c for c in all_candidates[identifier] if all(c.version in r.versions for r in requirements[identifier]) and c.version not in bad_versions ] return sorted(candidates, key=lambda c: c.version, reverse=True) def is_satisfied_by(self, requirement, candidate): return candidate.version in requirement.versions def run_resolver(*args): reporter = Reporter() resolver = Resolver(Provider(), reporter) try: resolver.resolve(*args) return reporter.backtracking_causes except ResolutionImpossible as e: return e.causes backtracking_causes = run_resolver( [_Requirement("a", {1, 2}), _Requirement("b", {1})] ) exception_causes = run_resolver([_Requirement("a", {2}), _Requirement("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]] all_candidates: Mapping[str, Sequence[Candidate]] = { "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"), []), ], } class Provider(AbstractProvider[str, Candidate, str]): def identify(self, requirement_or_candidate: 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: str, *args: Any, **kwargs: Any) -> str: # prefer child over parent (alphabetically) return identifier def get_dependencies(self, candidate: Candidate) -> Sequence[str]: return candidate[2] def find_matches( self, identifier: str, requirements: Mapping[str, Iterator[str]], incompatibilities: Mapping[str, Iterator[Candidate]], ) -> 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: str, candidate: Candidate) -> bool: return candidate[1] in Requirement(requirement).specifier # patch Resolution._get_updated_criteria to collect rejected states rejected_criteria: list[Criterion] = [] get_updated_criteria_orig = Resolution._get_updated_criteria 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[str, Candidate, str] = Resolver(Provider(), reporter) result = resolver.resolve(["child", "parent"]) def get_child_versions( information: 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")