kalash.config

Module containing the entire configuration data model for Kalash

Type Aliases:

  • TestPath = str
  • AuxiliaryPath = str
  • UseCase = str
  • LastResult = str
  • TestId = str
  • Workbench = str
  • Device = str
  • Suite = str
  • FunctionalityItem = str
  • Toggle = bool
  • KalashYamlObj = Dict[str, Any]
  • ArbitraryYamlObj = Dict[str, Any]
  • ConstructorArgsTuple = Tuple[Any, ...]
  • TestModule = ModuleType
  • TemplateVersion = str
  • OneOrList = Union[List[T], T]

  • PathOrIdForWhatIf = List[str]

  • CollectorArtifact = Tuple[unittest.TestSuite, PathOrIdForWhatIf]
  • Collector = Callable[[TestPath, Trigger], CollectorArtifact]
View Source
from __future__ import annotations

__doc__ = """"""

from collections import defaultdict
from types import ModuleType
from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union
from dataclasses import dataclass, field
from toolz import pipe
from dataclasses_jsonschema import JsonSchemaMixin

import unittest
import logging
import os
import yaml
import inspect

from .smuggle import smuggle
from .spec import Spec

T = TypeVar('T')
TestPath = str
AuxiliaryPath = str
UseCase = str
LastResult = str
TestId = str
Workbench = str
Device = str
Suite = str
FunctionalityItem = str
Toggle = bool
KalashYamlObj = Dict[str, Any]
ArbitraryYamlObj = Dict[str, Any]
ConstructorArgsTuple = Tuple[Any, ...]
TestModule = ModuleType
TemplateVersion = str
OneOrList = Union[List[T], T]

# Please document type aliases below:

__doc__ += """
Module containing the entire configuration data model for Kalash

Type Aliases:

* `TestPath` = `str`
* `AuxiliaryPath` = `str`
* `UseCase` = `str`
* `LastResult` = `str`
* `TestId` = `str`
* `Workbench` = `str`
* `Device` = `str`
* `Suite` = `str`
* `FunctionalityItem` = `str`
* `Toggle` = `bool`
* `KalashYamlObj` = `Dict[str, Any]`
* `ArbitraryYamlObj` = `Dict[str, Any]`
* `ConstructorArgsTuple` = `Tuple[Any, ...]`
* `TestModule` = `ModuleType`
* `TemplateVersion` = `str`
* `OneOrList` = `Union[List[T], T]`
"""


@dataclass
class CliConfig:
    """A class collecting all CLI options fed into
    the application. The instance is created by the
    main function and used downstream in the call stack.

    Args:
        file (Optional[str]): config filename (YAML or Python file)
        log_dir (str): base directory for log files
        group_by (Optional[str]): group logs by a particular property
            from the metadata tag
        no_recurse (bool): don't recurse into subfolders when scanning
            for tests to run
        debug (bool): run in debug mode
        no_log (bool): suppress logging
        no_log_echo (bool): suppress log echoing to STDOUT
        spec_path (str): custom YAML/Meta specification path, the file
            should be in YAML format
        log_level (int): `logging` module log level
        log_format (str): formatter string for `logging` module logger
        what_if (Optional[str]): either 'ids' or 'paths', prints hypothetical
            list of IDs or paths of collected tests instead of running the
            actual tests, useful for debugging and designing test suites
        fail_fast (bool): if `True` the test suite won't be continued if
            at least one of the tests that have been collected and triggered
            has failed
    """
    file:        Optional[str] = None
    # if not running in CLI context we initialize reasonable defaults:
    log_dir:     str           = '.'
    group_by:    Optional[str] = None
    no_recurse:  bool          = False
    debug:       bool          = False
    no_log:      bool          = False
    no_log_echo: bool          = False
    spec_path:   str           = 'spec.yaml'
    log_level:   int           = logging.INFO
    log_format:  str           = '%(message)s'
    what_if:     Optional[str] = None
    fail_fast:   bool          = False

    def __post_init__(self):
        spec_abspath = os.path.join(os.path.dirname(__file__), self.spec_path)
        self.spec = Spec.load_spec(spec_abspath)
        self.log_format = self.spec.cli_config.log_formatter


class classproperty(object):
    """https://stackoverflow.com/a/13624858
    Only Python 3.9 allows stacking `@classmethod`
    and `@property` decorators to obtain static
    properties. We use this decorator as a workaround
    since we wish to support 3.7+ for quite a while.
    """
    def __init__(self, fget):
        self.fget = fget

    def __get__(self, owner_self, owner_cls):
        return self.fget(owner_cls)


@dataclass
class SharedMetaElements:
    """Collects Metadata-modifying methods with `CliConfig` instance
    providing a parameter closure. Most methods here are related
    to built-in interpolation of patterns like `$(WorkDir)`.
    """

    cli_config: CliConfig

    def _interpolate_workdir(self, ipt: str) -> str:
        """Interpolates CWD variable. This variable is used to
        resolve paths within Kalash YAML relative to the current
        working directory. Equivalent to using the dotted file path.

        Args:
            ipt (str): input string to interpolate
            yaml_abspath (str): path to the Kalash YAML file.

        Returns: interpolated string
        """
        return os.path.normpath(
            ipt.replace(
                self.cli_config.spec.test.interp_cwd, os.getcwd()
            )
        )

    def _interpolate_this_file(self, ipt: str, yaml_abspath: str) -> str:
        """Interpolates THIS_FILE variable. THIS_FILE is used to resolve
        paths within Kalash YAML relative to the YAML file itself.

        Args:
            ipt (str): input string to interpolate
            yaml_abspath (str): path to the Kalash YAML file
                or the `.py` configuration file

        Returns: interpolated string
        """
        return os.path.normpath(
            ipt.replace(
                self.cli_config.spec.test.interp_this_file,
                os.path.dirname(yaml_abspath)
            )
        )

    def _interpolate_all(self, ipt: Union[str, None], yaml_abspath: str) -> Union[str, None]:
        """Interpolates all variable values using a toolz.pipe

        Args:
            ipt (str): input string to interpolate
            yaml_abspath (str): path to the Kalash YAML file
                or the `.py` configuration file

        Returns: interpolated string
        """
        if ipt:
            return pipe(
                self._interpolate_this_file(ipt, yaml_abspath),
                self._interpolate_workdir
            )
        return ipt

    def resolve_interpolables(self, o: object, yaml_abspath: str):
        for k, v in o.__dict__.items():
            if type(v) is str:
                setattr(o, k, self._interpolate_all(v, yaml_abspath))


@dataclass
class Base:
    """Base config class. `Meta`, `Config` and `Test`
    inherit from this minimal pseudo-abstract base class.
    """
    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Base:
        raise NotImplementedError("Base class methods should be overridden")

    def get(self, argname: str):
        """`getattr` alias for those who wish to use this
        from within the `TestCase` class.
        """
        return getattr(self, argname, None)


@dataclass
class Meta(Base, JsonSchemaMixin):
    """Provides a specification outline for the Metadata tag
    in test templates.

    Args:
        id (Optional[TestId]): unique test ID
        version (Optional[TemplateVersion]): template version
        use_cases (Optional[OneOrList[UseCase]]): one or more
            use case IDs (preferably from a task tracking system
            like Jira) that a particular test refers to
        workbenches (Optional[OneOrList[Workbench]]): one or more
            physical workbenches where the test should be triggered
        devices (Optional[OneOrList[Device]]): one or more device
            categories for which this test has been implemented
        suites (Optional[OneOrList[Suite]]): one or more arbitrary
            suite tags (should be used only if remaining tags don't
            provide enough possibilities to describe the context of
            the test script)
        functionality (Optional[OneOrList[FunctionalityItem]]): one
            or more functionality descriptors for the test script
    """
    id:            Optional[TestId] = None
    version:       Optional[TemplateVersion] = None
    use_cases:     Optional[OneOrList[UseCase]] = None
    workbenches:   Optional[OneOrList[Workbench]] = None
    devices:       Optional[OneOrList[Device]] = None
    suites:        Optional[OneOrList[Suite]] = None
    functionality: Optional[OneOrList[FunctionalityItem]] = None
    cli_config:    CliConfig = CliConfig()

    def __post_init__(self):
        frame = inspect.stack()[1]
        module = inspect.getmodule(frame[0])
        if module:
            module_path = os.path.abspath(module.__file__)
            SharedMetaElements(self.cli_config).resolve_interpolables(self, module_path)

    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Meta:
        block_spec = cli_config.spec.test
        meta_spec = cli_config.spec.meta
        params = dict(
            id=yaml_obj.get(block_spec.id, None),
            version=yaml_obj.get(meta_spec.template_version, None),
            use_cases=yaml_obj.get(meta_spec.related_usecase, None),
            workbenches=yaml_obj.get(meta_spec.workbench, None),
            devices=yaml_obj.get(block_spec.devices, None),
            suites=yaml_obj.get(block_spec.suites, None),
            functionality=yaml_obj.get(block_spec.functionality, None)
        )
        return Meta(
            **params
        )


@dataclass
class Test(Meta, JsonSchemaMixin):
    """Provides a specification outline for a single category
    of tests that should be collected, e.g. by path, ID or any
    other parameter inherited from `Meta`.

    Args:
        path (Optional[OneOrList[TestPath]]): path to a test
            directory or a single test path
        id (Optional[OneOrList[TestId]]): one or more IDs to
            filter for
        no_recurse (Optional[Toggle]): if `True`, subfolders
            will not be searched for tests, intended for use with
            the `path` parameter
        last_result (Optional[LastResult]): if `OK` then filters
            out only the tests that have passed in the last run,
            if `NOK` then it only filters out those tests that
            have failed in the last run
        setup (Optional[AuxiliaryPath]): path to a setup script;
            runs once at the start of the test category run
        teardown (Optional[AuxiliaryPath]): path to a teardown
            script; runs once at the end of the test category
            run
    """
    path:          Optional[OneOrList[TestPath]] = None
    id:            Optional[OneOrList[TestId]] = None
    no_recurse:    Optional[Toggle] = None
    last_result:   Optional[LastResult] = None
    setup:         Optional[AuxiliaryPath] = None
    teardown:      Optional[AuxiliaryPath] = None
    cli_config:    CliConfig = CliConfig()

    def __post_init__(self):
        frame = inspect.stack()[1]
        module = inspect.getmodule(frame[0])
        if module:
            module_path = os.path.abspath(module.__file__)
            SharedMetaElements(self.cli_config).resolve_interpolables(self, module_path)

    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Test:
        """Loads `Test` blocks from a YAML object."""

        block_spec = cli_config.spec.test
        base_class_instance = super().from_yaml_obj(yaml_obj, cli_config)
        return Test(
            path=yaml_obj.get(block_spec.path, None),
            no_recurse=yaml_obj.get(block_spec.no_recurse, None),
            last_result=yaml_obj.get(block_spec.last_result, None),
            setup=yaml_obj.get(block_spec.setup_script, None),
            teardown=yaml_obj.get(block_spec.teardown_script, None),
            **base_class_instance.__dict__
        )

    @classproperty
    def _non_filters(cls):
        # ID is listed as non-filter beacuse it's handled
        # differently. A `Test` definition can filter for
        # multiple IDs. A `Meta` definition can only have
        # one ID (1 ID == 1 test case). Hence ID is handled
        # separately in the `apply_filters` function using
        # `match_id` helper
        return ['setup', 'teardown', 'path', 'id']


@dataclass
class Config(Base, JsonSchemaMixin):
    """Provides a specification outline for the runtime
    parameters. Where `Test` defines what tests to collect,
    this class defines global parameters determining how
    to run tests.

    Args:
        report (str): directory path where reports will
            be stored in XML format
        setup (Optional[AuxiliaryPath]): path to a setup script;
            runs once at the start of the complete run
        teardown (Optional[AuxiliaryPath]): path to a teardown
            script; runs once at the end of the complete run
    """
    report: str = './kalash_reports'
    setup: Optional[AuxiliaryPath] = None
    teardown: Optional[AuxiliaryPath] = None
    cli_config: CliConfig = CliConfig()

    def __post_init__(self):
        SharedMetaElements(self.cli_config).resolve_interpolables(self, __file__)

    @classmethod
    def from_yaml_obj(cls, yaml_obj: Optional[ArbitraryYamlObj], cli_config: CliConfig) -> Config:
        """Loads `Test` blocks from a YAML object."""
        config_spec = cli_config.spec.config
        if yaml_obj:
            return Config(
                yaml_obj.get(config_spec.report, None),
                yaml_obj.get(config_spec.one_time_setup_script, None),
                yaml_obj.get(config_spec.one_time_teardown_script, None)
            )
        else:
            return Config()


@dataclass
class Trigger(JsonSchemaMixin):
    """Main configuration class collecting all information for
    a test run, passed down throughout the whole call stack.

    Args:
        tests (List[Test]): list of `Test` categories, each
            describing a sliver of a test suite that shares certain
            test collection parameters
        config (Config): a `Config` object defining parameters
            telling Kalash *how* to run the tests
        cli_config (CliConfig): a `CliConfig` object representing
            command-line parameters used to trigger the test run
            modifying behavior of certain aspects of the application
            like logging or triggering speculative runs instead of
            real runs
    """
    tests:  List[Test] = field(default_factory=list)
    config: Config     = field(default_factory=lambda: Config())
    cli_config: CliConfig = field(default_factory=lambda: CliConfig())

    @classmethod
    def from_file(cls, file_path: str, cli_config: CliConfig):
        """Creates a `Trigger` instance from a YAML or JSON file."""
        with open(file_path, 'r') as f:
            yaml_obj: ArbitraryYamlObj = defaultdict(lambda: None, yaml.safe_load(f))
        list_blocks: List[ArbitraryYamlObj] = \
            yaml_obj[cli_config.spec.test.tests]
        cfg_section: ArbitraryYamlObj = yaml_obj[cli_config.spec.config.cfg]
        tests = [Test.from_yaml_obj(i, cli_config) for i in list_blocks]
        config = Config.from_yaml_obj(cfg_section, cli_config)
        return Trigger(tests, config, cli_config)

    def _resolve_interpolables(self, path: str):
        sm = SharedMetaElements(self.cli_config)
        for test in self.tests:
            sm.resolve_interpolables(test, path)
        sm.resolve_interpolables(self.config, path)

    @classmethod
    def infer_trigger(cls, cli_config: CliConfig, default_path: str = '.kalash.yaml'):
        """Creates the Trigger instance from a YAML file or
        a Python file.

        Args:
            path (str): path to the configuration file.

        Returns: `Tests` object
        """
        path = cli_config.file if cli_config.file else default_path
        if path.endswith('.yaml') or path.endswith('.json'):
            t = cls()
            t = Trigger.from_file(os.path.abspath(path), cli_config)
            t._resolve_interpolables(path)
            return t
        else:
            module = smuggle(os.path.abspath(path))
            for _, v in module.__dict__.items():
                if type(v) is cls:
                    v._resolve_interpolables(path)
                    return v
            else:
                raise ValueError(
                    f"No {cls.__name__} instance found in file {path}"
                )


PathOrIdForWhatIf = List[str]
CollectorArtifact = Tuple[unittest.TestSuite, PathOrIdForWhatIf]  # can be a list of IDs or paths
                                                                  # or a full test suite
Collector = Callable[[TestPath, Trigger], CollectorArtifact]

__doc__ += """
* `PathOrIdForWhatIf` = `List[str]`
* `CollectorArtifact` = `Tuple[unittest.TestSuite, PathOrIdForWhatIf]`
* `Collector` = `Callable[[TestPath, Trigger], CollectorArtifact]`
"""
#  
@dataclass
class CliConfig:
View Source
@dataclass
class CliConfig:
    """A class collecting all CLI options fed into
    the application. The instance is created by the
    main function and used downstream in the call stack.

    Args:
        file (Optional[str]): config filename (YAML or Python file)
        log_dir (str): base directory for log files
        group_by (Optional[str]): group logs by a particular property
            from the metadata tag
        no_recurse (bool): don't recurse into subfolders when scanning
            for tests to run
        debug (bool): run in debug mode
        no_log (bool): suppress logging
        no_log_echo (bool): suppress log echoing to STDOUT
        spec_path (str): custom YAML/Meta specification path, the file
            should be in YAML format
        log_level (int): `logging` module log level
        log_format (str): formatter string for `logging` module logger
        what_if (Optional[str]): either 'ids' or 'paths', prints hypothetical
            list of IDs or paths of collected tests instead of running the
            actual tests, useful for debugging and designing test suites
        fail_fast (bool): if `True` the test suite won't be continued if
            at least one of the tests that have been collected and triggered
            has failed
    """
    file:        Optional[str] = None
    # if not running in CLI context we initialize reasonable defaults:
    log_dir:     str           = '.'
    group_by:    Optional[str] = None
    no_recurse:  bool          = False
    debug:       bool          = False
    no_log:      bool          = False
    no_log_echo: bool          = False
    spec_path:   str           = 'spec.yaml'
    log_level:   int           = logging.INFO
    log_format:  str           = '%(message)s'
    what_if:     Optional[str] = None
    fail_fast:   bool          = False

    def __post_init__(self):
        spec_abspath = os.path.join(os.path.dirname(__file__), self.spec_path)
        self.spec = Spec.load_spec(spec_abspath)
        self.log_format = self.spec.cli_config.log_formatter

A class collecting all CLI options fed into the application. The instance is created by the main function and used downstream in the call stack.

Args: file (Optional[str]): config filename (YAML or Python file) log_dir (str): base directory for log files group_by (Optional[str]): group logs by a particular property from the metadata tag no_recurse (bool): don't recurse into subfolders when scanning for tests to run debug (bool): run in debug mode no_log (bool): suppress logging no_log_echo (bool): suppress log echoing to STDOUT spec_path (str): custom YAML/Meta specification path, the file should be in YAML format log_level (int): logging module log level log_format (str): formatter string for logging module logger what_if (Optional[str]): either 'ids' or 'paths', prints hypothetical list of IDs or paths of collected tests instead of running the actual tests, useful for debugging and designing test suites fail_fast (bool): if True the test suite won't be continued if at least one of the tests that have been collected and triggered has failed

#   CliConfig( file: Optional[str] = None, log_dir: str = '.', group_by: Optional[str] = None, no_recurse: bool = False, debug: bool = False, no_log: bool = False, no_log_echo: bool = False, spec_path: str = 'spec.yaml', log_level: int = 20, log_format: str = '%(message)s', what_if: Optional[str] = None, fail_fast: bool = False )
#   file: Optional[str] = None
#   log_dir: str = '.'
#   group_by: Optional[str] = None
#   no_recurse: bool = False
#   debug: bool = False
#   no_log: bool = False
#   no_log_echo: bool = False
#   spec_path: str = 'spec.yaml'
#   log_level: int = 20
#   log_format: str = '%(message)s'
#   what_if: Optional[str] = None
#   fail_fast: bool = False
#   class classproperty:
View Source
class classproperty(object):
    """https://stackoverflow.com/a/13624858
    Only Python 3.9 allows stacking `@classmethod`
    and `@property` decorators to obtain static
    properties. We use this decorator as a workaround
    since we wish to support 3.7+ for quite a while.
    """
    def __init__(self, fget):
        self.fget = fget

    def __get__(self, owner_self, owner_cls):
        return self.fget(owner_cls)

https://stackoverflow.com/a/13624858 Only Python 3.9 allows stacking @classmethod and @property decorators to obtain static properties. We use this decorator as a workaround since we wish to support 3.7+ for quite a while.

#   classproperty(fget)
View Source
    def __init__(self, fget):
        self.fget = fget
#  
@dataclass
class SharedMetaElements:
View Source
@dataclass
class SharedMetaElements:
    """Collects Metadata-modifying methods with `CliConfig` instance
    providing a parameter closure. Most methods here are related
    to built-in interpolation of patterns like `$(WorkDir)`.
    """

    cli_config: CliConfig

    def _interpolate_workdir(self, ipt: str) -> str:
        """Interpolates CWD variable. This variable is used to
        resolve paths within Kalash YAML relative to the current
        working directory. Equivalent to using the dotted file path.

        Args:
            ipt (str): input string to interpolate
            yaml_abspath (str): path to the Kalash YAML file.

        Returns: interpolated string
        """
        return os.path.normpath(
            ipt.replace(
                self.cli_config.spec.test.interp_cwd, os.getcwd()
            )
        )

    def _interpolate_this_file(self, ipt: str, yaml_abspath: str) -> str:
        """Interpolates THIS_FILE variable. THIS_FILE is used to resolve
        paths within Kalash YAML relative to the YAML file itself.

        Args:
            ipt (str): input string to interpolate
            yaml_abspath (str): path to the Kalash YAML file
                or the `.py` configuration file

        Returns: interpolated string
        """
        return os.path.normpath(
            ipt.replace(
                self.cli_config.spec.test.interp_this_file,
                os.path.dirname(yaml_abspath)
            )
        )

    def _interpolate_all(self, ipt: Union[str, None], yaml_abspath: str) -> Union[str, None]:
        """Interpolates all variable values using a toolz.pipe

        Args:
            ipt (str): input string to interpolate
            yaml_abspath (str): path to the Kalash YAML file
                or the `.py` configuration file

        Returns: interpolated string
        """
        if ipt:
            return pipe(
                self._interpolate_this_file(ipt, yaml_abspath),
                self._interpolate_workdir
            )
        return ipt

    def resolve_interpolables(self, o: object, yaml_abspath: str):
        for k, v in o.__dict__.items():
            if type(v) is str:
                setattr(o, k, self._interpolate_all(v, yaml_abspath))

Collects Metadata-modifying methods with CliConfig instance providing a parameter closure. Most methods here are related to built-in interpolation of patterns like $(WorkDir).

#   SharedMetaElements(cli_config: kalash.config.CliConfig)
#   def resolve_interpolables(self, o: object, yaml_abspath: str):
View Source
    def resolve_interpolables(self, o: object, yaml_abspath: str):
        for k, v in o.__dict__.items():
            if type(v) is str:
                setattr(o, k, self._interpolate_all(v, yaml_abspath))
#  
@dataclass
class Base:
View Source
@dataclass
class Base:
    """Base config class. `Meta`, `Config` and `Test`
    inherit from this minimal pseudo-abstract base class.
    """
    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Base:
        raise NotImplementedError("Base class methods should be overridden")

    def get(self, argname: str):
        """`getattr` alias for those who wish to use this
        from within the `TestCase` class.
        """
        return getattr(self, argname, None)

Base config class. Meta, Config and Test inherit from this minimal pseudo-abstract base class.

#   Base()
#  
@classmethod
def from_yaml_obj( cls, yaml_obj: Dict[str, Any], cli_config: kalash.config.CliConfig ) -> kalash.config.Base:
View Source
    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Base:
        raise NotImplementedError("Base class methods should be overridden")
#   def get(self, argname: str):
View Source
    def get(self, argname: str):
        """`getattr` alias for those who wish to use this
        from within the `TestCase` class.
        """
        return getattr(self, argname, None)

getattr alias for those who wish to use this from within the TestCase class.

#  
@dataclass
class Meta(Base, dataclasses_jsonschema.JsonSchemaMixin):
View Source
@dataclass
class Meta(Base, JsonSchemaMixin):
    """Provides a specification outline for the Metadata tag
    in test templates.

    Args:
        id (Optional[TestId]): unique test ID
        version (Optional[TemplateVersion]): template version
        use_cases (Optional[OneOrList[UseCase]]): one or more
            use case IDs (preferably from a task tracking system
            like Jira) that a particular test refers to
        workbenches (Optional[OneOrList[Workbench]]): one or more
            physical workbenches where the test should be triggered
        devices (Optional[OneOrList[Device]]): one or more device
            categories for which this test has been implemented
        suites (Optional[OneOrList[Suite]]): one or more arbitrary
            suite tags (should be used only if remaining tags don't
            provide enough possibilities to describe the context of
            the test script)
        functionality (Optional[OneOrList[FunctionalityItem]]): one
            or more functionality descriptors for the test script
    """
    id:            Optional[TestId] = None
    version:       Optional[TemplateVersion] = None
    use_cases:     Optional[OneOrList[UseCase]] = None
    workbenches:   Optional[OneOrList[Workbench]] = None
    devices:       Optional[OneOrList[Device]] = None
    suites:        Optional[OneOrList[Suite]] = None
    functionality: Optional[OneOrList[FunctionalityItem]] = None
    cli_config:    CliConfig = CliConfig()

    def __post_init__(self):
        frame = inspect.stack()[1]
        module = inspect.getmodule(frame[0])
        if module:
            module_path = os.path.abspath(module.__file__)
            SharedMetaElements(self.cli_config).resolve_interpolables(self, module_path)

    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Meta:
        block_spec = cli_config.spec.test
        meta_spec = cli_config.spec.meta
        params = dict(
            id=yaml_obj.get(block_spec.id, None),
            version=yaml_obj.get(meta_spec.template_version, None),
            use_cases=yaml_obj.get(meta_spec.related_usecase, None),
            workbenches=yaml_obj.get(meta_spec.workbench, None),
            devices=yaml_obj.get(block_spec.devices, None),
            suites=yaml_obj.get(block_spec.suites, None),
            functionality=yaml_obj.get(block_spec.functionality, None)
        )
        return Meta(
            **params
        )

Provides a specification outline for the Metadata tag in test templates.

Args: id (Optional[TestId]): unique test ID version (Optional[TemplateVersion]): template version use_cases (Optional[OneOrList[UseCase]]): one or more use case IDs (preferably from a task tracking system like Jira) that a particular test refers to workbenches (Optional[OneOrList[Workbench]]): one or more physical workbenches where the test should be triggered devices (Optional[OneOrList[Device]]): one or more device categories for which this test has been implemented suites (Optional[OneOrList[Suite]]): one or more arbitrary suite tags (should be used only if remaining tags don't provide enough possibilities to describe the context of the test script) functionality (Optional[OneOrList[FunctionalityItem]]): one or more functionality descriptors for the test script

#   Meta( id: Optional[str] = None, version: Optional[str] = None, use_cases: Union[str, List[str], NoneType] = None, workbenches: Union[str, List[str], NoneType] = None, devices: Union[str, List[str], NoneType] = None, suites: Union[str, List[str], NoneType] = None, functionality: Union[str, List[str], NoneType] = None, cli_config: kalash.config.CliConfig = CliConfig(file=None, log_dir='.', group_by=None, no_recurse=False, debug=False, no_log=False, no_log_echo=False, spec_path='spec.yaml', log_level=20, log_format='%(message)s', what_if=None, fail_fast=False) )
#   id: Optional[str] = None
#   version: Optional[str] = None
#   use_cases: Union[str, List[str], NoneType] = None
#   workbenches: Union[str, List[str], NoneType] = None
#   devices: Union[str, List[str], NoneType] = None
#   suites: Union[str, List[str], NoneType] = None
#   functionality: Union[str, List[str], NoneType] = None
#   cli_config: kalash.config.CliConfig = CliConfig(file=None, log_dir='.', group_by=None, no_recurse=False, debug=False, no_log=False, no_log_echo=False, spec_path='spec.yaml', log_level=20, log_format='%(message)s', what_if=None, fail_fast=False)
#  
@classmethod
def from_yaml_obj( cls, yaml_obj: Dict[str, Any], cli_config: kalash.config.CliConfig ) -> kalash.config.Meta:
View Source
    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Meta:
        block_spec = cli_config.spec.test
        meta_spec = cli_config.spec.meta
        params = dict(
            id=yaml_obj.get(block_spec.id, None),
            version=yaml_obj.get(meta_spec.template_version, None),
            use_cases=yaml_obj.get(meta_spec.related_usecase, None),
            workbenches=yaml_obj.get(meta_spec.workbench, None),
            devices=yaml_obj.get(block_spec.devices, None),
            suites=yaml_obj.get(block_spec.suites, None),
            functionality=yaml_obj.get(block_spec.functionality, None)
        )
        return Meta(
            **params
        )
Inherited Members
Base
get
dataclasses_jsonschema.JsonSchemaMixin
field_mapping
register_field_encoders
to_dict
from_dict
from_object
all_json_schemas
json_schema
from_json
to_json
#  
@dataclass
class Test(Meta, dataclasses_jsonschema.JsonSchemaMixin):
View Source
@dataclass
class Test(Meta, JsonSchemaMixin):
    """Provides a specification outline for a single category
    of tests that should be collected, e.g. by path, ID or any
    other parameter inherited from `Meta`.

    Args:
        path (Optional[OneOrList[TestPath]]): path to a test
            directory or a single test path
        id (Optional[OneOrList[TestId]]): one or more IDs to
            filter for
        no_recurse (Optional[Toggle]): if `True`, subfolders
            will not be searched for tests, intended for use with
            the `path` parameter
        last_result (Optional[LastResult]): if `OK` then filters
            out only the tests that have passed in the last run,
            if `NOK` then it only filters out those tests that
            have failed in the last run
        setup (Optional[AuxiliaryPath]): path to a setup script;
            runs once at the start of the test category run
        teardown (Optional[AuxiliaryPath]): path to a teardown
            script; runs once at the end of the test category
            run
    """
    path:          Optional[OneOrList[TestPath]] = None
    id:            Optional[OneOrList[TestId]] = None
    no_recurse:    Optional[Toggle] = None
    last_result:   Optional[LastResult] = None
    setup:         Optional[AuxiliaryPath] = None
    teardown:      Optional[AuxiliaryPath] = None
    cli_config:    CliConfig = CliConfig()

    def __post_init__(self):
        frame = inspect.stack()[1]
        module = inspect.getmodule(frame[0])
        if module:
            module_path = os.path.abspath(module.__file__)
            SharedMetaElements(self.cli_config).resolve_interpolables(self, module_path)

    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Test:
        """Loads `Test` blocks from a YAML object."""

        block_spec = cli_config.spec.test
        base_class_instance = super().from_yaml_obj(yaml_obj, cli_config)
        return Test(
            path=yaml_obj.get(block_spec.path, None),
            no_recurse=yaml_obj.get(block_spec.no_recurse, None),
            last_result=yaml_obj.get(block_spec.last_result, None),
            setup=yaml_obj.get(block_spec.setup_script, None),
            teardown=yaml_obj.get(block_spec.teardown_script, None),
            **base_class_instance.__dict__
        )

    @classproperty
    def _non_filters(cls):
        # ID is listed as non-filter beacuse it's handled
        # differently. A `Test` definition can filter for
        # multiple IDs. A `Meta` definition can only have
        # one ID (1 ID == 1 test case). Hence ID is handled
        # separately in the `apply_filters` function using
        # `match_id` helper
        return ['setup', 'teardown', 'path', 'id']

Provides a specification outline for a single category of tests that should be collected, e.g. by path, ID or any other parameter inherited from Meta.

Args: path (Optional[OneOrList[TestPath]]): path to a test directory or a single test path id (Optional[OneOrList[TestId]]): one or more IDs to filter for no_recurse (Optional[Toggle]): if True, subfolders will not be searched for tests, intended for use with the path parameter last_result (Optional[LastResult]): if OK then filters out only the tests that have passed in the last run, if NOK then it only filters out those tests that have failed in the last run setup (Optional[AuxiliaryPath]): path to a setup script; runs once at the start of the test category run teardown (Optional[AuxiliaryPath]): path to a teardown script; runs once at the end of the test category run

#   Test( id: Union[str, List[str], NoneType] = None, version: Optional[str] = None, use_cases: Union[str, List[str], NoneType] = None, workbenches: Union[str, List[str], NoneType] = None, devices: Union[str, List[str], NoneType] = None, suites: Union[str, List[str], NoneType] = None, functionality: Union[str, List[str], NoneType] = None, cli_config: kalash.config.CliConfig = CliConfig(file=None, log_dir='.', group_by=None, no_recurse=False, debug=False, no_log=False, no_log_echo=False, spec_path='spec.yaml', log_level=20, log_format='%(message)s', what_if=None, fail_fast=False), path: Union[str, List[str], NoneType] = None, no_recurse: Optional[bool] = None, last_result: Optional[str] = None, setup: Optional[str] = None, teardown: Optional[str] = None )
#   path: Union[str, List[str], NoneType] = None
#   id: Union[str, List[str], NoneType] = None
#   no_recurse: Optional[bool] = None
#   last_result: Optional[str] = None
#   setup: Optional[str] = None
#   teardown: Optional[str] = None
#   cli_config: kalash.config.CliConfig = CliConfig(file=None, log_dir='.', group_by=None, no_recurse=False, debug=False, no_log=False, no_log_echo=False, spec_path='spec.yaml', log_level=20, log_format='%(message)s', what_if=None, fail_fast=False)
#  
@classmethod
def from_yaml_obj( cls, yaml_obj: Dict[str, Any], cli_config: kalash.config.CliConfig ) -> kalash.config.Test:
View Source
    @classmethod
    def from_yaml_obj(cls, yaml_obj: ArbitraryYamlObj, cli_config: CliConfig) -> Test:
        """Loads `Test` blocks from a YAML object."""

        block_spec = cli_config.spec.test
        base_class_instance = super().from_yaml_obj(yaml_obj, cli_config)
        return Test(
            path=yaml_obj.get(block_spec.path, None),
            no_recurse=yaml_obj.get(block_spec.no_recurse, None),
            last_result=yaml_obj.get(block_spec.last_result, None),
            setup=yaml_obj.get(block_spec.setup_script, None),
            teardown=yaml_obj.get(block_spec.teardown_script, None),
            **base_class_instance.__dict__
        )

Loads Test blocks from a YAML object.

Inherited Members
Meta
version
use_cases
workbenches
devices
suites
functionality
Base
get
dataclasses_jsonschema.JsonSchemaMixin
field_mapping
register_field_encoders
to_dict
from_dict
from_object
all_json_schemas
json_schema
from_json
to_json
#  
@dataclass
class Config(Base, dataclasses_jsonschema.JsonSchemaMixin):
View Source
@dataclass
class Config(Base, JsonSchemaMixin):
    """Provides a specification outline for the runtime
    parameters. Where `Test` defines what tests to collect,
    this class defines global parameters determining how
    to run tests.

    Args:
        report (str): directory path where reports will
            be stored in XML format
        setup (Optional[AuxiliaryPath]): path to a setup script;
            runs once at the start of the complete run
        teardown (Optional[AuxiliaryPath]): path to a teardown
            script; runs once at the end of the complete run
    """
    report: str = './kalash_reports'
    setup: Optional[AuxiliaryPath] = None
    teardown: Optional[AuxiliaryPath] = None
    cli_config: CliConfig = CliConfig()

    def __post_init__(self):
        SharedMetaElements(self.cli_config).resolve_interpolables(self, __file__)

    @classmethod
    def from_yaml_obj(cls, yaml_obj: Optional[ArbitraryYamlObj], cli_config: CliConfig) -> Config:
        """Loads `Test` blocks from a YAML object."""
        config_spec = cli_config.spec.config
        if yaml_obj:
            return Config(
                yaml_obj.get(config_spec.report, None),
                yaml_obj.get(config_spec.one_time_setup_script, None),
                yaml_obj.get(config_spec.one_time_teardown_script, None)
            )
        else:
            return Config()

Provides a specification outline for the runtime parameters. Where Test defines what tests to collect, this class defines global parameters determining how to run tests.

Args: report (str): directory path where reports will be stored in XML format setup (Optional[AuxiliaryPath]): path to a setup script; runs once at the start of the complete run teardown (Optional[AuxiliaryPath]): path to a teardown script; runs once at the end of the complete run

#   Config( report: str = './kalash_reports', setup: Optional[str] = None, teardown: Optional[str] = None, cli_config: kalash.config.CliConfig = CliConfig(file=None, log_dir='.', group_by=None, no_recurse=False, debug=False, no_log=False, no_log_echo=False, spec_path='spec.yaml', log_level=20, log_format='%(message)s', what_if=None, fail_fast=False) )
#   report: str = './kalash_reports'
#   setup: Optional[str] = None
#   teardown: Optional[str] = None
#   cli_config: kalash.config.CliConfig = CliConfig(file=None, log_dir='.', group_by=None, no_recurse=False, debug=False, no_log=False, no_log_echo=False, spec_path='spec.yaml', log_level=20, log_format='%(message)s', what_if=None, fail_fast=False)
#  
@classmethod
def from_yaml_obj( cls, yaml_obj: Optional[Dict[str, Any]], cli_config: kalash.config.CliConfig ) -> kalash.config.Config:
View Source
    @classmethod
    def from_yaml_obj(cls, yaml_obj: Optional[ArbitraryYamlObj], cli_config: CliConfig) -> Config:
        """Loads `Test` blocks from a YAML object."""
        config_spec = cli_config.spec.config
        if yaml_obj:
            return Config(
                yaml_obj.get(config_spec.report, None),
                yaml_obj.get(config_spec.one_time_setup_script, None),
                yaml_obj.get(config_spec.one_time_teardown_script, None)
            )
        else:
            return Config()

Loads Test blocks from a YAML object.

Inherited Members
Base
get
dataclasses_jsonschema.JsonSchemaMixin
field_mapping
register_field_encoders
to_dict
from_dict
from_object
all_json_schemas
json_schema
from_json
to_json
#  
@dataclass
class Trigger(dataclasses_jsonschema.JsonSchemaMixin):
View Source
@dataclass
class Trigger(JsonSchemaMixin):
    """Main configuration class collecting all information for
    a test run, passed down throughout the whole call stack.

    Args:
        tests (List[Test]): list of `Test` categories, each
            describing a sliver of a test suite that shares certain
            test collection parameters
        config (Config): a `Config` object defining parameters
            telling Kalash *how* to run the tests
        cli_config (CliConfig): a `CliConfig` object representing
            command-line parameters used to trigger the test run
            modifying behavior of certain aspects of the application
            like logging or triggering speculative runs instead of
            real runs
    """
    tests:  List[Test] = field(default_factory=list)
    config: Config     = field(default_factory=lambda: Config())
    cli_config: CliConfig = field(default_factory=lambda: CliConfig())

    @classmethod
    def from_file(cls, file_path: str, cli_config: CliConfig):
        """Creates a `Trigger` instance from a YAML or JSON file."""
        with open(file_path, 'r') as f:
            yaml_obj: ArbitraryYamlObj = defaultdict(lambda: None, yaml.safe_load(f))
        list_blocks: List[ArbitraryYamlObj] = \
            yaml_obj[cli_config.spec.test.tests]
        cfg_section: ArbitraryYamlObj = yaml_obj[cli_config.spec.config.cfg]
        tests = [Test.from_yaml_obj(i, cli_config) for i in list_blocks]
        config = Config.from_yaml_obj(cfg_section, cli_config)
        return Trigger(tests, config, cli_config)

    def _resolve_interpolables(self, path: str):
        sm = SharedMetaElements(self.cli_config)
        for test in self.tests:
            sm.resolve_interpolables(test, path)
        sm.resolve_interpolables(self.config, path)

    @classmethod
    def infer_trigger(cls, cli_config: CliConfig, default_path: str = '.kalash.yaml'):
        """Creates the Trigger instance from a YAML file or
        a Python file.

        Args:
            path (str): path to the configuration file.

        Returns: `Tests` object
        """
        path = cli_config.file if cli_config.file else default_path
        if path.endswith('.yaml') or path.endswith('.json'):
            t = cls()
            t = Trigger.from_file(os.path.abspath(path), cli_config)
            t._resolve_interpolables(path)
            return t
        else:
            module = smuggle(os.path.abspath(path))
            for _, v in module.__dict__.items():
                if type(v) is cls:
                    v._resolve_interpolables(path)
                    return v
            else:
                raise ValueError(
                    f"No {cls.__name__} instance found in file {path}"
                )

Main configuration class collecting all information for a test run, passed down throughout the whole call stack.

Args: tests (List[Test]): list of Test categories, each describing a sliver of a test suite that shares certain test collection parameters config (Config): a Config object defining parameters telling Kalash how to run the tests cli_config (CliConfig): a CliConfig object representing command-line parameters used to trigger the test run modifying behavior of certain aspects of the application like logging or triggering speculative runs instead of real runs

#   Trigger( tests: List[kalash.config.Test] = <factory>, config: kalash.config.Config = <factory>, cli_config: kalash.config.CliConfig = <factory> )
#  
@classmethod
def from_file(cls, file_path: str, cli_config: kalash.config.CliConfig):
View Source
    @classmethod
    def from_file(cls, file_path: str, cli_config: CliConfig):
        """Creates a `Trigger` instance from a YAML or JSON file."""
        with open(file_path, 'r') as f:
            yaml_obj: ArbitraryYamlObj = defaultdict(lambda: None, yaml.safe_load(f))
        list_blocks: List[ArbitraryYamlObj] = \
            yaml_obj[cli_config.spec.test.tests]
        cfg_section: ArbitraryYamlObj = yaml_obj[cli_config.spec.config.cfg]
        tests = [Test.from_yaml_obj(i, cli_config) for i in list_blocks]
        config = Config.from_yaml_obj(cfg_section, cli_config)
        return Trigger(tests, config, cli_config)

Creates a Trigger instance from a YAML or JSON file.

#  
@classmethod
def infer_trigger( cls, cli_config: kalash.config.CliConfig, default_path: str = '.kalash.yaml' ):
View Source
    @classmethod
    def infer_trigger(cls, cli_config: CliConfig, default_path: str = '.kalash.yaml'):
        """Creates the Trigger instance from a YAML file or
        a Python file.

        Args:
            path (str): path to the configuration file.

        Returns: `Tests` object
        """
        path = cli_config.file if cli_config.file else default_path
        if path.endswith('.yaml') or path.endswith('.json'):
            t = cls()
            t = Trigger.from_file(os.path.abspath(path), cli_config)
            t._resolve_interpolables(path)
            return t
        else:
            module = smuggle(os.path.abspath(path))
            for _, v in module.__dict__.items():
                if type(v) is cls:
                    v._resolve_interpolables(path)
                    return v
            else:
                raise ValueError(
                    f"No {cls.__name__} instance found in file {path}"
                )

Creates the Trigger instance from a YAML file or a Python file.

Args: path (str): path to the configuration file.

Returns: Tests object

Inherited Members
dataclasses_jsonschema.JsonSchemaMixin
field_mapping
register_field_encoders
to_dict
from_dict
from_object
all_json_schemas
json_schema
from_json
to_json