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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
from __future__ import print_function
import os
import sys
import json
import logging
import argparse
from yatest_lib.ya import Ya
RECIPE_START_OPTION = "start"
RECIPE_STOP_OPTION = "stop"
ya = None
collect_cores = None
sanitizer_extra_checks = None
def _setup_logging(level=logging.DEBUG):
root_logger = logging.getLogger()
root_logger.setLevel(level)
log_format = '%(asctime)s - %(levelname)s - %(name)s - %(funcName)s: %(message)s'
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter(log_format)
stdout_handler.setFormatter(formatter)
root_logger.addHandler(stdout_handler)
def get_options():
parser = argparse.ArgumentParser()
parser.add_argument("--show-cwd", action="store_true", dest="show_cwd", default=False, help="show recipe cwd")
parser.add_argument("--test-debug", action="store_true", dest="test_debug", default=False, help="test debug mode")
parser.add_argument("--test-stderr", action="store_true", dest="test_stderr", default=False, help="test stderr")
parser.add_argument("--pdb", action="store_true", dest="pdb", default=False, help="run pdb on error")
parser.add_argument("--sanitizer-extra-checks", dest="sanitizer_extra_checks", action="store_true", default=False, help="enables extra checks for tests built with sanitizers")
parser.add_argument("--collect-cores", dest="collect_cores", action="store_true", default=False, help="allows core dump file recovering during test")
args, opts = parser.parse_known_args()
global ya, sanitizer_extra_checks, collect_cores
_setup_logging()
context = {
"test_stderr": args.test_stderr,
}
ya = Ya(context=context)
ya._data_root = "" # XXX remove
sanitizer_extra_checks = args.sanitizer_extra_checks
if sanitizer_extra_checks:
for envvar in ['LSAN_OPTIONS', 'ASAN_OPTIONS']:
if envvar in os.environ:
os.environ.pop(envvar)
if envvar + '_ORIGINAL' in os.environ:
os.environ[envvar] = os.environ[envvar + '_ORIGINAL']
collect_cores = args.collect_cores
for recipe_option in RECIPE_START_OPTION, RECIPE_STOP_OPTION:
if recipe_option in opts:
return args, opts[opts.index(recipe_option):]
def set_env(key, value):
with open(ya.env_file, "a") as f:
json.dump({key: value}, f)
f.write("\n")
def tty():
if os.isatty(1):
return
f = open('/dev/tty', 'w+')
fd = f.fileno()
os.dup2(fd, 0)
os.dup2(fd, 1)
os.dup2(fd, 2)
def declare_recipe(start, stop):
parsed_args, argv = get_options()
if parsed_args.show_cwd:
print("Recipe \"{} {}\" working dir is {}".format(sys.argv[0], " ".join(argv), os.getcwd()))
try:
if argv[0] == RECIPE_START_OPTION:
start(argv[1:])
elif argv[0] == RECIPE_STOP_OPTION:
stop(argv[1:])
except Exception:
if parsed_args.pdb:
tty()
import ipdb
ipdb.post_mortem()
else:
raise
|
