1import glob
2import os
3import re
4from collections import namedtuple
5
6import pytest
7import yaml
8
9from contrib.namespaces import LinuxNamespace
10
11Scenario = namedtuple("Scenario", ["path", "qmin", "config"])
12
13
14def config_sanity_check(config_dict, config_name):
15    """Checks if parsed configuration is valid"""
16    mandatory_keys = {'name', 'binary', 'templates', 'configs', 'additional'}
17    for cfg in config_dict['programs']:
18        missing_keys = mandatory_keys - set(cfg.keys())
19        assert not missing_keys, 'Mandatory fields in configuration are missing: %s' % missing_keys
20
21        # sanity check templates vs. configs
22        assert len(cfg['templates']) == len(cfg['configs']),\
23            ('Number of jinja2 template files is not equal '
24             'to number of config files to be generated for '
25             'program "%s" (%s), i.e. len(templates) != len(configs)'
26             % (cfg['name'], config_name))
27
28        for additional in cfg["additional"]:
29            assert isinstance(additional, str),\
30                "All additional arguments in yaml should be strings. (%s, %s)"\
31                % (cfg['name'], config_name)
32
33
34def get_qmin_config(path):
35    """Reads configuration from the *.rpl file and determines query-minimization setting."""
36    with open(path) as f:
37        for line in f:
38            if re.search(r"^CONFIG_END", line) or re.search(r"^SCENARIO_BEGIN", line):
39                return None
40            if re.search(r"^\s*query-minimization:\s*(on|yes)", line):
41                return True
42            if re.search(r"^\s*query-minimization:\s*(off|no)", line):
43                return False
44    return None
45
46
47def scenarios(paths, configs):
48    """Returns list of *.rpl files from given path and packs them with their minimization setting"""
49
50    assert len(paths) == len(configs),\
51        "Number of --config has to be equal to number of --scenarios arguments."
52
53    scenario_list = []
54
55    for path, config in zip(paths, configs):
56        with open(config) as f:
57            config_dict = yaml.load(f, yaml.SafeLoader)
58        config_sanity_check(config_dict, config)
59
60        if os.path.isfile(path):
61            filelist = [path]  # path to single file, accept it
62        else:
63            filelist = sorted(glob.glob(os.path.join(path, "*.rpl")))
64
65        if not filelist:
66            raise ValueError('no *.rpl files found in path "{}"'.format(path))
67
68        for file in filelist:
69            scenario_list.append(Scenario(file, get_qmin_config(file), config_dict))
70
71    return scenario_list
72
73
74def rpls(paths):
75    for path in paths:
76        if os.path.isfile(path):
77            filelist = [path]  # path to single file, accept it
78        else:
79            filelist = sorted(glob.glob(os.path.join(path, "*.rpl")))
80
81        return filelist
82
83
84def pytest_addoption(parser):
85    parser.addoption("--config", action="append", help="path to Deckard configuration .yaml file")
86    parser.addoption("--scenarios", action="append", help="directory with .rpl files")
87    parser.addoption("--retries", action="store", help=("number of retries per"
88                                                        "test when Deckard is under load"))
89
90
91def pytest_generate_tests(metafunc):
92    """This is pytest weirdness to parametrize the test over all the *.rpl files.
93    See https://docs.pytest.org/en/latest/parametrize.html#basic-pytest-generate-tests-example
94    for more info."""
95
96    if 'scenario' in metafunc.fixturenames:
97        if metafunc.config.option.config is None:
98            configs = []
99        else:
100            configs = metafunc.config.option.config
101
102        if metafunc.config.option.scenarios is None:
103            paths = ["sets/resolver"] * len(configs)
104        else:
105            paths = metafunc.config.option.scenarios
106
107        metafunc.parametrize("scenario", scenarios(paths, configs), ids=str)
108    if 'rpl_path' in metafunc.fixturenames:
109        paths = metafunc.config.option.scenarios
110        metafunc.parametrize("rpl_path", rpls(paths), ids=str)
111    if 'max_retries' in metafunc.fixturenames:
112        max_retries = metafunc.config.option.retries
113        if max_retries is None:
114            max_retries = 3
115        metafunc.parametrize("max_retries", [max_retries], ids=lambda id: "max-retries-"+str(id))
116
117
118def pytest_collection_modifyitems(items):
119    """We automatically mark test that need faking monotonic time and run them separately."""
120    for item in items:
121        if "monotonic" in item.nodeid:
122            item.add_marker(pytest.mark.monotonic)
123
124
125def pytest_runtest_setup(item):  # pylint: disable=unused-argument
126    LinuxNamespace("user").__enter__()
127