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