on-restart-benchmarks/tests/minizinc_testing/spec.py

from enum import Enum
from . import yaml
import minizinc as mzn
from minizinc.helpers import check_result
import pathlib
import re


@yaml.mapping(u"!Test")
class Test:
    """
    Represents a test case

    Encoded using `!Test` in YAML.
    """

    def __init__(self, **kwargs):
        self.name = yaml.Undefined
        self.solvers = ["gecode", "cbc", "chuffed"]
        self.check_against = []
        self.expected = []
        self.options = {}
        self.extra_files = []
        self.markers = []
        self.type = "solve"

        for key, value in kwargs.items():
            setattr(self, key, value)

    def run(self, mzn_file, solver, default_options={}):
        """
        Runs this test case given an mzn file path, a solver name and some default options.

        Any options specified in this test case directly will override those provided in
        default options.

        Returns a tuple containing:
        - the model produced
        - the result of running the solver
        - a list of expected results
        - the actual obtained result

        Pass the actual obtained result to passed() to determine if the test case passed.
        """
        options = {k: v for k, v in default_options.items()}
        options.update(self.options)
        file = pathlib.Path(mzn_file)
        extra_files = [file.parent.joinpath(other) for other in self.extra_files]

        try:
            model = mzn.Model([file] + extra_files)
            solver = mzn.Solver.lookup(solver)
            instance = mzn.Instance(solver, model)
            if self.type == "solve":
                instance.output_type = Solution
                result = instance.solve(**options)
                obtained = Result.from_mzn(result)
            elif self.type == "compile":
                with instance.flat(**options) as (fzn, ozn, stats):
                    obtained = FlatZinc.from_mzn(fzn, file.parent)
                    result = obtained
            elif self.type == "output-model":
                with instance.flat(**options) as (fzn, ozn, stats):
                    obtained = OutputModel.from_mzn(ozn, file.parent)
                    result = obtained
            else:
                raise NotImplementedError("Unknown test case type")
        except mzn.MiniZincError as error:
            result = error
            obtained = Error.from_mzn(result)

        required = self.expected if isinstance(self.expected, list) else [self.expected]

        return model, result, required, obtained

    def passed(self, actual):
        """
        Returns whether the given result satisfies this test case
        """
        if isinstance(self.expected, list):
            return any(exp.check(actual) for exp in self.expected)
        return self.expected.check(actual)


@yaml.mapping(u"!Result")
class Result:
    """
    Represents a result generated by calling `solve()` on an instance.

    Encoded using `!Result` in YAML.
    """

    def __init__(self, **kwargs):
        self.status = yaml.Undefined
        self.solution = yaml.Undefined

        for k, v in kwargs.items():
            setattr(self, k, v)

    def check(self, actual):
        """
        Returns whether this result and the given result match.
        """
        if not isinstance(actual, mzn.Result):
            return False

        if self.status is not yaml.Undefined and str(actual.status) != self.status:
            return False

        if self.solution is not yaml.Undefined and self.solution is not None:
            required = self.solution
            obtained = actual.solution

            if isinstance(required, list):
                return len(required) == len(obtained) and all(
                    x.is_satisfied(y) for x, y in zip(required, obtained)
                )
            else:
                return required.is_satisfied(obtained)

        return True

    @staticmethod
    def from_mzn(result):
        """
        Constructs a Result from a minizinc Result type produced by running a solver
        as opposed to deserializing from YAML.
        """
        instance = Result()
        instance.status = str(result.status)
        instance.solution = result.solution
        return instance


@yaml.sequence(u"!SolutionSet")
class SolutionSet:
    """
    Represents a set of expected solutions that passes when
    every obtained solution matches an entry in this set.

    Represented with `!SolutionSet` in YAML
    """

    def __init__(self, *args):
        self.items = args

    def __iter__(self):
        return iter(self.items)

    def is_satisfied(self, other):
        """
        Returns whether or not every solution in `other` matches one in this solution set
        """
        return all(any(item.is_satisfied(o) for item in self.items) for o in other)


@yaml.mapping(u"!Solution")
class Solution:
    """
    A solution which is satisfied when every member of the this solution matches a member in the obtained solution.

    That is, the obtained solution can contain extra items that are not compared.

    Represented by `!Solution` in YAML.
    """

    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)

    def items(self):
        return vars(self).items()

    def is_satisfied(self, other):
        """
        Returns whether or not this solution is satisfied by an actual solution
        """

        def convertEnums(data):
            # Convert enums to strings so that normal equality can be used
            if isinstance(data, Enum):
                return data.name
            if isinstance(data, list):
                return [convertEnums(d) for d in data]
            if isinstance(data, set):
                return set(convertEnums(d) for d in data)
            return data

        def in_other(k, v):
            try:
                return v == convertEnums(getattr(other, k))
            except AttributeError:
                return False

        return all(in_other(k, v) for k, v in self.__dict__.items())


@yaml.mapping(u"!Error")
class Error:
    """
    A MiniZinc error with a type name and message.

    Represented by `!Error` in YAML.
    """

    def __init__(self, **kwargs):
        self.type = yaml.Undefined
        self.message = yaml.Undefined
        self.regex = yaml.Undefined

        for key, value in kwargs.items():
            setattr(self, key, value)

    def check(self, actual):
        """
        Checks if this error matches an actual obtained result/error
        """
        if not isinstance(actual, mzn.MiniZincError):
            return False

        if self.type is not yaml.Undefined:
            kind = type(actual)
            classes = (kind,) + kind.__bases__
            if not any(self.type == c.__name__ for c in classes):
                return False

        if self.message is not yaml.Undefined:
            return self.message == str(actual)

        if self.regex is not yaml.Undefined:
            return re.match(self.regex, str(actual), flags=re.M | re.S) is not None

        return True

    @staticmethod
    def from_mzn(error):
        """
        Creates an Error from a MiniZinc error (as opposed to from deserializing YAML)
        """
        instance = Error()
        instance.type = type(error).__name__
        instance.message = str(error)
        return instance


class CachedResult:
    """
    An object used to store a model and result from solving an instance

    This is so that `CheckItem`s do not have to re-solve an instance when
    verifying against another solver.
    """

    def __init__(self):
        self.model = None
        self.result = None
        self.obtained = None

    def test(self, solver):
        """
        Checks that the result stored in this object, when given as data for the
        stored model with the given solver, is satisfied.
        """
        solver_instance = mzn.Solver.lookup(solver)
        passed = check_result(
            model=self.model, result=self.result, solver=solver_instance
        )
        return passed


@yaml.scalar(u"!FlatZinc")
class FlatZinc:
    """
    A FlatZinc result, encoded by !FlatZinc in YAML.
    """

    def __init__(self, path):
        self.path = path
        self.base = None
        self.fzn = None

    def check(self, actual):
        if not isinstance(actual, FlatZinc):
            return False

        if self.fzn is None:
            with open(actual.base.joinpath(self.path)) as f:
                self.fzn = f.read()

        return self.normalized() == actual.normalized()

    def normalized(self):
        """
        Quick and dirty normalization of FlatZinc

        Used to compare two `.fzn` files a bit more robustly than naive string comparison
        """
        fzn = self.fzn
        var_nums = re.findall(r"X_INTRODUCED_(\d+)_", fzn)
        for new, old in enumerate(sorted(set(int(x) for x in var_nums))):
            # Compress the introduced variable numbers, removing any holes and starting from 0
            needle = "X_INTRODUCED_{}_".format(old)
            replacement = "X_INTRODUCED_{}_".format(new)
            fzn = fzn.replace(needle, replacement)
        return "\n".join(sorted(fzn.split("\n")))  # Sort the FlatZinc lines

    def get_value(self):
        if self.fzn is None:
            return self.path
        return self.normalized()

    @staticmethod
    def from_mzn(fzn, base):
        """
        Creates a `FlatZinc` object from a `File` returned by `flat()` in the minizinc interface.

        Also takes the base path the mzn file was from, so that when loading the expected fzn file
        it can be done relative to the mzn path.
        """
        with open(fzn.name) as file:
            instance = FlatZinc(None)
            instance.fzn = file.read()
            instance.base = base
            return instance


@yaml.scalar(u"!OutputModel")
class OutputModel:
    """
    An OZN result, encoded by !OutputModel in YAML.
    """

    def __init__(self, path):
        self.path = path
        self.base = None
        self.ozn = None

    def check(self, actual):
        if not isinstance(actual, OutputModel):
            return False

        if self.ozn is None:
            with open(actual.base.joinpath(self.path)) as f:
                self.ozn = f.read()

        return self.ozn == actual.ozn

    def get_value(self):
        if self.ozn is None:
            return self.path
        return self.ozn

    @staticmethod
    def from_mzn(ozn, base):
        """
        Creates a `OutputModel` object from a `File` returned by `flat()` in the minizinc interface.

        Also takes the base path the mzn file was from, so that when loading the expected ozn file
        it can be done relative to the mzn path.
        """
        with open(ozn.name) as file:
            instance = OutputModel(None)
            instance.ozn = file.read()
            instance.base = base
            return instance


@yaml.mapping(u"!Suite")
class Suite:
    """
    Represents a test suite configuration

    Encoded using `!Suite` in YAML.
    """

    def __init__(self, **kwargs):
        self.solvers = None
        self.options = {}
        self.strict = True
        self.regex = r"."

        for key, value in kwargs.items():
            setattr(self, key, value)