1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
|
import logging
import os
import subprocess
import random
import sys
import time
import pytest
import deckard
def set_coverage_env(path, qmin):
"""Sets up enviroment variables so code coverage utility can work."""
if os.environ.get("COVERAGE"):
exports = subprocess.check_output([os.environ["COVERAGE_ENV_SCRIPT"],
os.environ["DAEMONSRCDIR"],
os.environ["COVERAGE_STATSDIR"],
path + "-qmin-" + str(qmin)]).decode()
for export in exports.split():
key, value = export.split("=", 1)
value = value.strip('"')
os.environ[key] = value
def check_platform():
if sys.platform == 'windows':
pytest.exit('Not supported at all on Windows')
# Suppress extensive Augeas logging
logging.getLogger("augeas").setLevel(logging.ERROR)
check_platform()
def run_test(path, qmin, config, max_retries, retries=0):
set_coverage_env(path, qmin)
try:
del os.environ["SOCKET_WRAPPER_DIR"]
except KeyError:
pass
try:
deckard.process_file(path, qmin, config)
except deckard.DeckardUnderLoadError as e:
if retries < max_retries:
logging.error("Deckard under load. Retrying…")
# Exponential backoff
time.sleep((2 ** retries) + random.random())
run_test(path, qmin, config, max_retries, retries + 1)
else:
raise e
def test_passes_qmin_on(scenario, max_retries):
if scenario.qmin is True or scenario.qmin is None:
run_test(scenario.path, True, scenario.config, max_retries)
else:
pytest.skip("Query minimization is off in test config")
def test_passes_qmin_off(scenario, max_retries):
if scenario.qmin is False or scenario.qmin is None:
run_test(scenario.path, False, scenario.config, max_retries)
else:
pytest.skip("Query minimization is on in test config")
|
