Forks Plugin¶

@dataclass(kw_only=True)
class ForkCovariantParameter:
    """Value list for a fork covariant parameter in a given fork."""

    names: List[str]
    values: List[ParameterSet]

class ForkParametrizer:
    """A parametrizer for a test case that is parametrized by the fork."""

    fork: Fork | TransitionFork
    fork_covariant_parameters: List[ForkCovariantParameter] = field(
        default_factory=list
    )

    def __init__(
        self,
        fork: Fork | TransitionFork,
        marks: List[pytest.MarkDecorator | pytest.Mark] | None = None,
        fork_covariant_parameters: List[ForkCovariantParameter] | None = None,
    ):
        """
        Initialize a new fork parametrizer object for a given fork.

        Args:
          fork: The fork for which the test cases will be parametrized.
          marks: A list of pytest marks to apply to all the test cases
                 parametrized by the fork.
          fork_covariant_parameters: A list of fork covariant parameters
                                     for the test case, for unit testing
                                     purposes only.

        """
        if marks is None:
            marks = []
        self.fork_covariant_parameters = [
            ForkCovariantParameter(
                names=["fork"],
                values=[
                    pytest.param(
                        fork,
                        marks=marks,
                    )
                ],
            )
        ]
        if fork_covariant_parameters is not None:
            self.fork_covariant_parameters.extend(fork_covariant_parameters)
        self.fork = fork

    @property
    def argnames(self) -> List[str]:
        """Return the parameter names for the test case."""
        argnames = []
        for p in self.fork_covariant_parameters:
            argnames.extend(p.names)
        return argnames

    @property
    def argvalues(self) -> List[ParameterSet]:
        """Return the parameter values for the test case."""
        parameter_set_combinations = itertools.product(
            # Add the values for each parameter, all of them are lists of at
            # least one element.
            *[p.values for p in self.fork_covariant_parameters],
        )

        parameter_set_list: List[ParameterSet] = []
        for parameter_set_combination in parameter_set_combinations:
            params: List[Any] = []
            marks: List[pytest.Mark | pytest.MarkDecorator] = []
            test_id: str | None = None
            for p in parameter_set_combination:
                assert isinstance(p, ParameterSet)
                params.extend(p.values)
                if p.marks:
                    marks.extend(p.marks)
                if p.id:
                    if test_id is None:
                        test_id = f"fork_{self.fork.name()}-{p.id}"
                    else:
                        test_id = f"{test_id}-{p.id}"
            parameter_set_list.append(
                pytest.param(*params, marks=marks, id=test_id)
            )

        return parameter_set_list

class CovariantDescriptor:
    """
    A descriptor for a parameter that is covariant with the fork: the
    parametrized values change depending on the fork.
    """

    argnames: List[str] = []
    fn: Callable[[Fork | TransitionFork], List[Any] | Iterable[Any]] | None = (
        None
    )

    selector: Callable[..., bool] | None = None
    marks: (
        None
        | pytest.Mark
        | pytest.MarkDecorator
        | List[pytest.Mark | pytest.MarkDecorator]
    ) = None

    def __init__(
        self,
        argnames: List[str] | str,
        fn: Callable[[Fork | TransitionFork], List[Any] | Iterable[Any]]
        | None = None,
        *,
        selector: Callable[..., bool] | None = None,
        marks: None
        | pytest.Mark
        | pytest.MarkDecorator
        | List[pytest.Mark | pytest.MarkDecorator] = None,
    ):
        """
        Initialize a new covariant descriptor.

        Args:
          argnames: The names of the parameters that are covariant with the
                    fork.
          fn: A function that takes the fork as the single parameter and
              returns the values for the parameter for each fork.
          selector: A function that filters the values for the parameter.
          marks: A list of pytest marks to apply to the test cases
                 parametrized by the parameter.

        """
        self.argnames = (
            [argname.strip() for argname in argnames.split(",")]
            if isinstance(argnames, str)
            else argnames
        )
        self.fn = fn
        self.selector = selector
        self.marks = marks

    def process_value(
        self,
        parameters_values: Any | List[Any] | Tuple[Any] | ParameterSet,
    ) -> ParameterSet | None:
        """
        Process a value for a covariant parameter.

        The `selector` is applied to parameters_values in order to filter them.
        """
        if isinstance(parameters_values, ParameterSet):
            return parameters_values

        if len(self.argnames) == 1:
            # Wrap values that are meant for a single parameter in a list
            parameters_values = [parameters_values]
        marks = self.marks
        if self.selector is None or self.selector(
            *parameters_values[: self.selector.__code__.co_argcount]
        ):
            if isinstance(marks, FunctionType):
                marks = marks(*parameters_values[: marks.__code__.co_argcount])
            assert not isinstance(marks, FunctionType), (
                "marks must be a list or None"
            )
            if marks is None:
                marks = []
            elif not isinstance(marks, list):
                marks = [marks]

            return pytest.param(*parameters_values, marks=marks)

        return None

    def process_values(self, values: Iterable[Any]) -> List[ParameterSet]:
        """
        Filter the values for the covariant parameter.

        I.e. if the marker has an argument, the argument is interpreted as a
        lambda function that filters the values.
        """
        processed_values: List[ParameterSet] = []
        for value in values:
            processed_value = self.process_value(value)
            if processed_value is not None:
                processed_values.append(processed_value)
        return processed_values

    def add_values(self, fork_parametrizer: ForkParametrizer) -> None:
        """Add the values for the covariant parameter to the parametrizer."""
        if self.fn is None:
            return
        fork = fork_parametrizer.fork
        values = self.fn(fork)
        values = self.process_values(values)
        assert len(values) > 0
        fork_parametrizer.fork_covariant_parameters.append(
            ForkCovariantParameter(names=self.argnames, values=values)
        )

class CovariantDecorator(CovariantDescriptor):
    """
    A marker used to parametrize a function by a covariant parameter with the
    values returned by a fork method.

    The decorator must be subclassed with the appropriate class variables
    before initialization.

    Attributes:
      marker_name: Name of the marker.
      description: Description of the marker.
      fork_attribute_name: Name of the method to call on the fork to
                           get the values.
      marker_parameter_names: Names of the parameters to be parametrized
                              in the test function.
      indirect: Whether the parameters should be passed through fixtures
                (indirect parametrization).

    """

    marker_name: ClassVar[str]
    description: ClassVar[str]
    fork_attribute_name: ClassVar[str]
    marker_parameter_names: ClassVar[List[str]]
    indirect: ClassVar[bool]

    def __init__(self, metafunc: Metafunc):
        """
        Initialize the covariant decorator.

        The decorator must already be subclassed with the appropriate class
        variables before initialization.

        Args:
            metafunc: The metafunc object that pytest uses when generating
                tests.

        """
        self.metafunc = metafunc

        m = metafunc.definition.iter_markers(self.marker_name)
        if m is None:
            return
        marker_list = list(m)
        assert len(marker_list) <= 1, (
            f"Multiple markers {self.marker_name} found"
        )
        if len(marker_list) == 0:
            return
        marker = marker_list[0]

        assert marker is not None
        assert len(marker.args) == 0, "Only keyword arguments are supported"

        kwargs = dict(marker.kwargs)

        selector = kwargs.pop("selector", lambda _: True)
        assert isinstance(selector, FunctionType), (
            "selector must be a function"
        )

        marks = kwargs.pop("marks", None)

        if len(kwargs) > 0:
            raise ValueError(
                f"Unknown arguments to {self.marker_name}: {kwargs}"
            )

        def fn(fork: Fork | TransitionFork) -> List[Any]:
            if fork.is_transition_fork:
                raise ValueError(f"Transition forks are not supported: {fork}")
            return getattr(fork, self.fork_attribute_name)()

        super().__init__(
            argnames=self.marker_parameter_names,
            fn=fn,
            selector=selector,
            marks=marks,
        )

@dataclass(kw_only=True)
class ValidityMarker(ABC):
    """
    Abstract class to represent any fork validity marker.

    Subclassing this class allows for the creation of new validity markers.

    Instantiation must be done per test function, and the `process` method must
    be called to process the fork arguments.

    When subclassing, the following optional parameters can be set:
    - marker_name: Name of the marker, if not set, the class name is
                   converted to underscore.
    - mutually_exclusive: List of other marker types incompatible
                          with this one.
    - flag: Whether the marker is a flag and should always be included.
    """

    marker_name: ClassVar[str]
    mutually_exclusive: ClassVar[List[Type["ValidityMarker"]]]
    flag: ClassVar[bool]

    mark: Mark | None

    class ValidityMarkerCombinationError(Exception):
        """
        Combination of two validity markers generates an empty fork range.
        """

        pass

    def __init_subclass__(
        cls,
        marker_name: str | None = None,
        mutually_exclusive: List[Type["ValidityMarker"]] | None = None,
        flag: bool = False,
        **kwargs: Any,
    ) -> None:
        """Register the validity marker subclass."""
        super().__init_subclass__(**kwargs)
        if marker_name is None:
            # Use the class name converted to underscore:
            # https://stackoverflow.com/a/1176023
            marker_name = MARKER_NAME_REGEX.sub("_", cls.__name__).lower()
        cls.marker_name = marker_name
        cls.mutually_exclusive = (
            mutually_exclusive if mutually_exclusive else []
        )
        cls.flag = flag
        if marker_name in ALL_VALIDITY_MARKERS:
            raise ValueError(f"Duplicate validity marker class: {cls}")
        ALL_VALIDITY_MARKERS[marker_name] = cls

    def __post_init__(self) -> None:
        """Post-initialize the validity marker."""
        if self.flag:
            return
        if self.mark is None:
            raise Exception(f"Marker error '{self.marker_name}'")
        if len(self.mark.args) == 0:
            raise Exception(
                f"Missing fork argument with '{self.marker_name}' marker"
            )

    def process_fork_arguments(
        self, *fork_args: str
    ) -> Set[Fork | TransitionFork]:
        """Process the fork arguments."""
        fork_eips_set = ForkEIPSetAdapter.validate_python(fork_args)
        if len(fork_eips_set) != len(fork_args):
            raise Exception(
                f"Duplicate argument specified in '{self.marker_name}'"
            )
        forks_set: Set[Fork | TransitionFork] = set()
        for fork_eip in fork_eips_set:
            if fork_eip.is_transition_fork:
                forks_set.add(fork_eip)
            else:
                if not fork_eip.is_eip():
                    forks_set.add(fork_eip)
                else:
                    forks_set |= fork_eip.enabling_forks()
        return forks_set

    @staticmethod
    def get_all_validity_markers(
        markers: Iterator[pytest.Mark],
    ) -> List["ValidityMarker"]:
        """Get all the validity markers applied to the test function."""
        markers_dict: Dict[str, ValidityMarker] = {}
        for marker in markers:
            for marker_name in ALL_VALIDITY_MARKERS:
                if marker.name == marker_name:
                    cls = ALL_VALIDITY_MARKERS[marker.name]
                    new_marker = cls(mark=marker)
                    try:
                        existing_marker = markers_dict[marker_name]
                    except KeyError:
                        markers_dict[marker_name] = new_marker
                    else:
                        existing_marker.update(new_marker)

        for cls in ALL_VALIDITY_MARKERS.values():
            if cls.flag and cls.marker_name not in markers_dict:
                markers_dict[cls.marker_name] = cls(mark=None)

        for marker_name, validity_marker in markers_dict.items():
            for incompatible_marker in validity_marker.mutually_exclusive:
                if incompatible_marker.marker_name in markers_dict:
                    raise Exception(
                        f"The markers '{incompatible_marker.marker_name}' and "
                        f"'{marker_name}' can't be combined. "
                    )
        return list(markers_dict.values())

    @staticmethod
    def get_test_fork_set(
        validity_markers: List["ValidityMarker"],
    ) -> Set[Fork | TransitionFork]:
        """
        Get the set of forks where a test is valid from the validity markers
        and filters.
        """
        test_fork_set: Set[Fork | TransitionFork] = set()
        if not len(
            [
                validity_marker
                for validity_marker in validity_markers
                if not validity_marker.flag
            ]
        ):
            # Limit to non-transition forks if no validity markers were applied
            test_fork_set |= set(ALL_FORKS)
        else:
            # Start with all forks and transitions if any validity markers were
            # applied
            test_fork_set |= set(ALL_FORKS_WITH_TRANSITIONS)

        for v in validity_markers:
            # Apply the validity markers to the test function if applicable
            test_fork_set = v.process(test_fork_set)

        return test_fork_set

    @staticmethod
    def get_test_fork_set_from_markers(
        markers: Iterator[pytest.Mark],
    ) -> Set[Fork | TransitionFork]:
        """
        Get the set of forks where a test is valid using the markers applied to
        the test.
        """
        return ValidityMarker.get_test_fork_set(
            ValidityMarker.get_all_validity_markers(markers)
        )

    @staticmethod
    def get_test_fork_set_from_metafunc(
        metafunc: Metafunc,
    ) -> Set[Fork | TransitionFork]:
        """
        Get the set of forks where a test is valid using its pytest
        meta-function.
        """
        return ValidityMarker.get_test_fork_set_from_markers(
            metafunc.definition.iter_markers()
        )

    @staticmethod
    def is_validity_or_filter_marker(name: str) -> bool:
        """Check if a marker is a validity or filter marker."""
        return name in ALL_VALIDITY_MARKERS

    def process(
        self, forks: Set[Fork | TransitionFork]
    ) -> Set[Fork | TransitionFork]:
        """Process the fork arguments."""
        if self.mark is None:
            fork_set = self._process_with_marker_args()
        else:
            fork_set = self._process_with_marker_args(
                *self.mark.args, **self.mark.kwargs
            )
        if self.flag:
            return forks - fork_set
        if not fork_set:
            # Test is marked for an EIP that is not yet enabled in any
            # fork.
            return fork_set
        resulting_set = forks & fork_set
        if not resulting_set:
            raise ValidityMarker.ValidityMarkerCombinationError()
        return resulting_set

    @abstractmethod
    def _process_with_marker_args(
        self, *args: Any, **kwargs: Any
    ) -> Set[Fork | TransitionFork]:
        """
        Process the fork arguments as specified for the marker.

        Method must be implemented by the subclass.

        If the validity marker is of flag type, the returned forks will be
        subtracted from the fork set, otherwise the returned forks will be
        intersected with the current set.
        """
        pass

    def update(self, other: Self) -> None:
        """
        Update `self` to be the more strict of `self` or `other`.

        For example:

        >>> first = ValidFrom("Frontier")
        >>> second = ValidFrom("Osaka")
        >>> first.update(second)
        >>> print(first)
        ValidFrom("Osaka")

        If `self` cannot be updated (no merging is possible/implemented),
        raises an exception.
        """
        del other
        raise Exception(
            f"Too many '{self.marker_name}' markers applied to test"
        )