conftest.py 5.54 KB
Newer Older
1
from collections import namedtuple, OrderedDict
2
import glob
3
import logging
4 5 6 7
import os
import re
import yaml

8 9
import pytest

10

11
Scenario = namedtuple("Scenario", ["path", "qmin", "config"])
12 13


14 15 16 17 18 19 20 21 22
def ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
    """Make YaML load to OrderedDict.
    This is done to ensure compability with Python versions prior to 3.6.
    See docs.python.org/3.6/whatsnew/3.6.html#new-dict-implementation for more information.

    repr(config) is a part of testcase's name in pytest.
    We need to ensure that it is ordered in the same way.
    See https://github.com/pytest-dev/pytest/issues/1075.
    """
23
    class OrderedLoader(Loader):  # pylint: disable=too-many-ancestors
24 25 26 27 28 29 30 31 32 33 34 35 36
        pass

    def construct_mapping(loader, node):
        loader.flatten_mapping(node)
        return object_pairs_hook(loader.construct_pairs(node))

    OrderedLoader.add_constructor(
        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
        construct_mapping)

    return yaml.load(stream, OrderedLoader)


37 38 39 40 41 42 43 44 45 46 47 48 49 50 51
def config_sanity_check(config_dict, config_name):
    """Checks if parsed configuration is valid"""
    mandatory_keys = {'name', 'binary', 'templates', 'configs', 'additional'}
    for cfg in config_dict['programs']:
        missing_keys = mandatory_keys - set(cfg.keys())
        assert not missing_keys, 'Mandatory fields in configuration are missing: %s' % missing_keys

        # sanity check templates vs. configs
        assert len(cfg['templates']) == len(cfg['configs']),\
            ('Number of jinja2 template files is not equal '
             'to number of config files to be generated for '
             'program "%s" (%s), i.e. len(templates) != len(configs)'
             % (cfg['name'], config_name))

        for additional in cfg["additional"]:
52
            assert isinstance(additional, str),\
53 54 55 56 57 58 59 60 61 62 63 64 65 66
                "All additional arguments in yaml should be strings. (%s, %s)"\
                % (cfg['name'], config_name)


def get_qmin_config(path):
    """Reads configuration from the *.rpl file and determines query-minimization setting."""
    with open(path) as f:
        for line in f:
            if re.search(r"^CONFIG_END", line) or re.search(r"^SCENARIO_BEGIN", line):
                return None
            if re.search(r"^\s*query-minimization:\s*(on|yes)", line):
                return True
            if re.search(r"^\s*query-minimization:\s*(off|no)", line):
                return False
67
    return None
68 69


70
def scenarios(paths, configs):
71
    """Returns list of *.rpl files from given path and packs them with their minimization setting"""
72

73 74
    assert len(paths) == len(configs),\
        "Number of --config has to be equal to number of --scenarios arguments."
75 76 77 78

    scenario_list = []

    for path, config in zip(paths, configs):
79
        config_dict = ordered_load(open(config), yaml.SafeLoader)
80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95
        config_sanity_check(config_dict, config)

        if os.path.isfile(path):
            filelist = [path]  # path to single file, accept it
        else:
            filelist = sorted(glob.glob(os.path.join(path, "*.rpl")))

        if not filelist:
            raise ValueError('no *.rpl files found in path "{}"'.format(path))

        for file in filelist:
            scenario_list.append(Scenario(file, get_qmin_config(file), config_dict))

    return scenario_list


96 97 98 99 100 101 102 103 104 105
def rpls(paths):
    for path in paths:
        if os.path.isfile(path):
            filelist = [path]  # path to single file, accept it
        else:
            filelist = sorted(glob.glob(os.path.join(path, "*.rpl")))

        return filelist


106
def pytest_addoption(parser):
107 108
    parser.addoption("--config", action="append", help="path to Deckard configuration .yaml file")
    parser.addoption("--scenarios", action="append", help="directory with .rpl files")
109 110
    parser.addoption("--retries", action="store", help=("number of retries per"
                                                        "test when Deckard is under load"))
111 112 113


def pytest_generate_tests(metafunc):
114
    """This is pytest weirdness to parametrize the test over all the *.rpl files.
Štěpán Balážik's avatar
Štěpán Balážik committed
115 116 117
    See https://docs.pytest.org/en/latest/parametrize.html#basic-pytest-generate-tests-example
    for more info."""

118
    if 'scenario' in metafunc.fixturenames:
119 120
        if metafunc.config.option.config is None:
            configs = []
121
        else:
122
            configs = metafunc.config.option.config
123

124 125 126 127
        if metafunc.config.option.scenarios is None:
            paths = ["sets/resolver"] * len(configs)
        else:
            paths = metafunc.config.option.scenarios
128

129
        metafunc.parametrize("scenario", scenarios(paths, configs), ids=str)
130
    if 'rpl_path' in metafunc.fixturenames:
131
        paths = metafunc.config.option.scenarios
132
        metafunc.parametrize("rpl_path", rpls(paths), ids=str)
133 134 135 136 137
    if 'max_retries' in metafunc.fixturenames:
        max_retries = metafunc.config.option.retries
        if max_retries is None:
            max_retries = 3
        metafunc.parametrize("max_retries", [max_retries], ids=str)
138

Štěpán Balážik's avatar
Štěpán Balážik committed
139

140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
def check_log_level_xdist(level):
    if level < logging.ERROR:
        pytest.exit("Advanced logging not available while running with xdist "
                    "(try ommiting -n option)")


def pytest_configure(config):
    # This means pytest-xdist is installed and enabled
    if hasattr(config.option, "dist") and config.option.dist == "load":
        log_level = config.option.log_level
        if log_level is None:
            return
        try:
            log_level = int(log_level)
        except ValueError:
            log_level = logging.getLevelName(log_level)
        check_log_level_xdist(log_level)