wally/suits/job.py

import abc
from typing import Dict, Any, Tuple, cast, Union
from collections import OrderedDict

from ..common_types import Storable


class JobParams(metaclass=abc.ABCMeta):
    """Class contains all job parameters, which significantly affects job results.
    Like block size or operation type, but not file name or file size.
    Can be used as key in dictionary
    """

    def __init__(self, **params: Dict[str, Any]) -> None:
        self.params = params

    @abc.abstractproperty
    def summary(self) -> str:
        """Test short summary, used mostly for file names and short image description"""
        pass

    @abc.abstractproperty
    def long_summary(self) -> str:
        """Readable long summary for management and deployment engineers"""
        pass

    @abc.abstractmethod
    def copy(self, **updated) -> 'JobParams':
        pass

    def __getitem__(self, name: str) -> Any:
        return self.params[name]

    def __setitem__(self, name: str, val: Any) -> None:
        self.params[name] = val

    def __hash__(self) -> int:
        return hash(self.char_tpl)

    def __eq__(self, o: object) -> bool:
        if not isinstance(o, self.__class__):
            raise TypeError("Can't compare {!r} to {!r}".format(self.__class__.__qualname__, type(o).__qualname__))
        return sorted(self.params.items()) == sorted(cast(JobParams, o).params.items())

    def __lt__(self, o: object) -> bool:
        if not isinstance(o, self.__class__):
            raise TypeError("Can't compare {!r} to {!r}".format(self.__class__.__qualname__, type(o).__qualname__))
        return self.char_tpl < cast(JobParams, o).char_tpl

    @abc.abstractproperty
    def char_tpl(self) -> Tuple[Union[str, int, float, bool], ...]:
        pass


class JobConfig(Storable, metaclass=abc.ABCMeta):
    """Job config class"""

    def __init__(self, idx: int) -> None:
        # job id, used in storage to distinct jobs with same summary
        self.idx = idx

        # time interval, in seconds, when test was running on all nodes
        self.reliable_info_range = None  # type: Tuple[int, int]

        # all job parameters, both from suite file and config file
        self.vals = OrderedDict()  # type: Dict[str, Any]

    @property
    def storage_id(self) -> str:
        """unique string, used as key in storage"""
        return "{}_{}".format(self.summary, self.idx)

    @abc.abstractproperty
    def params(self) -> JobParams:
        """Should return a copy"""
        pass

    @property
    def summary(self) -> str:
        return self.params.summary