import collections
import json
import operator
import os
from collections import defaultdict

import packaging.markers
import packaging.requirements
import packaging.utils
import packaging.version
import pytest

import resolvelib.resolvers.resolution
from resolvelib import AbstractProvider, ResolutionImpossible, Resolver

Candidate = collections.namedtuple("Candidate", "name version extras")


def _eval_marker(marker, extras=(None,)):
    if not marker:
        return True
    if not isinstance(marker, packaging.markers.Marker):
        marker = packaging.markers.Marker(marker)
    return any(marker.evaluate({"extra": extra}) for extra in extras)


def _iter_resolved(data):
    for k, v in data.items():
        if not isinstance(v, dict):
            v = {"version": v}
        yield k, v


class PythonInputProvider(AbstractProvider):
    def __init__(self, filename):
        with open(filename) as f:
            case_data = json.load(f)

        index_name = os.path.normpath(
            os.path.join(filename, "..", "..", "index", case_data["index"] + ".json"),
        )
        with open(index_name) as f:
            self.index = json.load(f)

        self.root_requirements = [
            packaging.requirements.Requirement(r) for r in case_data["requested"]
        ]

        if "resolved" in case_data:
            self.expected_resolution = {
                k: packaging.version.parse(v["version"])
                for k, v in _iter_resolved(case_data["resolved"])
                if _eval_marker(v.get("marker"))
            }
        else:
            self.expected_resolution = None

        if "conflicted" in case_data:
            self.expected_confliction = set(case_data["conflicted"])
        else:
            self.expected_confliction = None

        if "unvisited" in case_data:
            self.expected_unvisited = case_data["unvisited"]
        else:
            self.expected_unvisited = None

        if "needs_optimistic" in case_data:
            self.needs_optimistic = case_data["needs_optimistic"]
        else:
            self.needs_optimistic = False

    def identify(self, requirement_or_candidate):
        name = packaging.utils.canonicalize_name(requirement_or_candidate.name)
        if requirement_or_candidate.extras:
            extras_str = ",".join(sorted(requirement_or_candidate.extras))
            return f"{name}[{extras_str}]"
        return name

    def get_preference(
        self,
        identifier,
        resolutions,
        candidates,
        information,
        backtrack_causes,
    ):
        transitive = all(p is not None for _, p in information[identifier])
        return (transitive, identifier)

    def _iter_matches(self, identifier, requirements, incompatibilities):
        name, _, _ = identifier.partition("[")
        bad_versions = {c.version for c in incompatibilities[identifier]}
        extras = {e for r in requirements[identifier] for e in r.extras}
        for key in self.index[name]:
            v = packaging.version.parse(key)
            if any(v not in r.specifier for r in requirements[identifier]):
                continue
            if v in bad_versions:
                continue
            yield Candidate(name=name, version=v, extras=extras)

    def find_matches(self, identifier, requirements, incompatibilities):
        candidates = sorted(
            self._iter_matches(identifier, requirements, incompatibilities),
            key=operator.attrgetter("version"),
            reverse=True,
        )
        return candidates

    def is_satisfied_by(self, requirement, candidate):
        return candidate.version in requirement.specifier

    def _iter_dependencies(self, candidate):
        name = packaging.utils.canonicalize_name(candidate.name)
        if candidate.extras:
            r = f"{name}=={candidate.version}"
            yield packaging.requirements.Requirement(r)
        for r in self.index[name][str(candidate.version)]["dependencies"]:
            requirement = packaging.requirements.Requirement(r)
            if not _eval_marker(requirement.marker, candidate.extras):
                continue
            yield requirement

    def get_dependencies(self, candidate):
        return list(self._iter_dependencies(candidate))


class PythonInputProviderNarrowRequirements(PythonInputProvider):
    def narrow_requirement_selection(
        self, identifiers, resolutions, candidates, information, backtrack_causes
    ):
        # Consider requirements that have 0 candidates (a resolution end point
        # that can be backtracked from) or 1 candidate (speeds up situations where
        # ever requirement is pinned to 1 specific version)
        number_of_candidates = defaultdict(list)
        for identifier in identifiers:
            number_of_candidates[len(list(candidates[identifier]))].append(identifier)

        min_candidates = min(number_of_candidates.keys())
        if min_candidates in (0, 1):
            return number_of_candidates[min_candidates]

        return identifiers


INPUTS_DIR = os.path.abspath(os.path.join(__file__, "..", "inputs"))

CASE_DIR = os.path.join(INPUTS_DIR, "case")

CASE_NAMES = [name for name in os.listdir(CASE_DIR) if name.endswith(".json")]


XFAIL_CASES = {
    "pyrex-1.9.8.json": "Too many rounds (>500)",
}


def create_params(provider_class):
    return [
        pytest.param(
            (os.path.join(CASE_DIR, n), provider_class),
            marks=pytest.mark.xfail(strict=True, reason=XFAIL_CASES[n]),
        )
        if n in XFAIL_CASES
        else (os.path.join(CASE_DIR, n), provider_class)
        for n in CASE_NAMES
    ]


@pytest.fixture(
    params=[
        *create_params(PythonInputProvider),
        *create_params(PythonInputProviderNarrowRequirements),
    ],
    ids=[
        f"{n[:-5]}-{cls.__name__}"
        for cls in [PythonInputProvider, PythonInputProviderNarrowRequirements]
        for n in CASE_NAMES
    ],
)
def provider(request):
    path, provider_class = request.param
    return provider_class(path)


def _format_confliction(exception):
    return {
        packaging.utils.canonicalize_name(cause.requirement.name)
        for cause in exception.causes
    }


def _format_resolution(result):
    return {
        identifier: candidate.version
        for identifier, candidate in result.mapping.items()
        if not candidate.extras
    }


def test_resolver(provider, reporter):
    resolver = Resolver(provider, reporter)

    if provider.expected_confliction:
        with pytest.raises(ResolutionImpossible) as ctx:
            result = resolver.resolve(provider.root_requirements)
            print(_format_resolution(result))  # Provide some debugging hints.
        assert _format_confliction(ctx.value) == provider.expected_confliction
    else:
        resolution = resolver.resolve(provider.root_requirements)
        assert _format_resolution(resolution) == provider.expected_resolution

    if provider.expected_unvisited:
        visited_versions = defaultdict(set)
        for visited_candidate in reporter.visited:
            visited_versions[visited_candidate.name].add(str(visited_candidate.version))

        for name, versions in provider.expected_unvisited.items():
            if name not in visited_versions:
                continue

            unexpected_versions = set(versions).intersection(visited_versions[name])
            assert not unexpected_versions, (
                f"Unexpcted versions visited {name}: {', '.join(unexpected_versions)}"
            )


def test_no_optimistic_backtracking_resolver(provider, reporter, monkeypatch):
    """
    Tests the resolver works with optimistic backtracking disabled for all
    cases, except for skipping candidates that is known to require optimistic
    backtracking.
    """
    monkeypatch.setattr(
        resolvelib.resolvers.resolution, "_OPTIMISTIC_BACKJUMPING_RATIO", 0.0
    )
    resolver = Resolver(provider, reporter)

    if provider.expected_confliction:
        with pytest.raises(ResolutionImpossible) as ctx:
            result = resolver.resolve(provider.root_requirements)
            print(_format_resolution(result))  # Provide some debugging hints.
        assert _format_confliction(ctx.value) == provider.expected_confliction
    else:
        resolution = resolver.resolve(provider.root_requirements)
        assert _format_resolution(resolution) == provider.expected_resolution

    if provider.expected_unvisited and not provider.needs_optimistic:
        visited_versions = defaultdict(set)
        for visited_candidate in reporter.visited:
            visited_versions[visited_candidate.name].add(str(visited_candidate.version))

        for name, versions in provider.expected_unvisited.items():
            if name not in visited_versions:
                continue

            unexpected_versions = set(versions).intersection(visited_versions[name])
            assert not unexpected_versions, (
                f"Unexpcted versions visited {name}: {', '.join(unexpected_versions)}"
            )