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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
|
import errno
import functools
import json
import os
import threading
import six
_lock = threading.Lock()
_config = None
def _set_ya_config(config=None, ya=None):
global _config
if config:
_config = config
elif ya:
class Config:
def __init__(self):
self.ya = None
_config = Config()
_config.ya = ya
def _get_ya_config():
if _config:
return _config
else:
try:
import pytest
return pytest.config
except (ImportError, AttributeError):
raise NotImplementedError("yatest.common.* is only available from the testing runtime")
def _get_ya_plugin_instance():
return _get_ya_config().ya
def _norm_path(path):
if path is None:
return None
assert isinstance(path, six.string_types)
if "\\" in path:
raise AssertionError("path {} contains Windows seprators \\ - replace them with '/'".format(path))
return os.path.normpath(path)
def _join_path(main_path, path):
if not path:
return main_path
return os.path.join(main_path, _norm_path(path))
def not_test(func):
"""
Mark any function as not a test for py.test
:param func:
:return:
"""
@functools.wraps(func)
def wrapper(*args, **kwds):
return func(*args, **kwds)
setattr(wrapper, '__test__', False)
return wrapper
def source_path(path=None):
"""
Get source path inside arcadia
:param path: path arcadia relative, e.g. yatest.common.source_path('devtools/ya')
:return: absolute path to the source folder
"""
return _join_path(_get_ya_plugin_instance().source_root, path)
def build_path(path=None):
"""
Get path inside build directory
:param path: path relative to the build directory, e.g. yatest.common.build_path('devtools/ya/bin')
:return: absolute path inside build directory
"""
return _join_path(_get_ya_plugin_instance().build_root, path)
def java_path():
"""
[DEPRECATED] Get path to java
:return: absolute path to java
"""
from . import runtime_java
return runtime_java.get_java_path(binary_path(os.path.join('contrib', 'tools', 'jdk')))
def java_home():
"""
Get jdk directory path
"""
from . import runtime_java
jdk_dir = runtime_java.get_build_java_dir(binary_path('jdk'))
if not jdk_dir:
raise Exception("Cannot find jdk - make sure 'jdk' is added to the DEPENDS section and exists for the current platform")
return jdk_dir
def java_bin():
"""
Get path to the java binary
Requires DEPENDS(jdk)
"""
return os.path.join(java_home(), "bin", "java")
def data_path(path=None):
"""
Get path inside arcadia_tests_data directory
:param path: path relative to the arcadia_tests_data directory, e.g. yatest.common.data_path("pers/rerank_service")
:return: absolute path inside arcadia_tests_data
"""
return _join_path(_get_ya_plugin_instance().data_root, path)
def output_path(path=None):
"""
Get path inside the current test suite output dir.
Placing files to this dir guarantees that files will be accessible after the test suite execution.
:param path: path relative to the test suite output dir
:return: absolute path inside the test suite output dir
"""
return _join_path(_get_ya_plugin_instance().output_dir, path)
def ram_drive_path(path=None):
"""
:param path: path relative to the ram drive.
:return: absolute path inside the ram drive directory or None if no ram drive was provided by environment.
"""
if 'YA_TEST_RAM_DRIVE_PATH' in os.environ:
return _join_path(os.environ['YA_TEST_RAM_DRIVE_PATH'], path)
elif get_param("ram_drive_path"):
return _join_path(get_param("ram_drive_path"), path)
def output_ram_drive_path(path=None):
"""
Returns path inside ram drive directory which will be saved in the testing_out_stuff directory after testing.
Returns None if no ram drive was provided by environment.
:param path: path relative to the output ram drive directory
"""
if 'YA_TEST_OUTPUT_RAM_DRIVE_PATH' in os.environ:
return _join_path(os.environ['YA_TEST_OUTPUT_RAM_DRIVE_PATH'], path)
elif _get_ya_plugin_instance().get_context("test_output_ram_drive_path"):
return _join_path(_get_ya_plugin_instance().get_context("test_output_ram_drive_path"), path)
def binary_path(path=None):
"""
Get path to the built binary
:param path: path to the binary relative to the build directory e.g. yatest.common.binary_path('devtools/ya/bin/ya-bin')
:return: absolute path to the binary
"""
path = _norm_path(path)
return _get_ya_plugin_instance().get_binary(path)
def work_path(path=None):
"""
Get path inside the current test suite working directory. Creating files in the work directory does not guarantee
that files will be accessible after the test suite execution
:param path: path relative to the test suite working dir
:return: absolute path inside the test suite working dir
"""
return _join_path(
os.environ.get("TEST_WORK_PATH") or
_get_ya_plugin_instance().get_context("work_path") or
os.getcwd(),
path)
def python_path():
"""
Get path to the arcadia python.
Warn: if you are using build with system python (-DUSE_SYSTEM_PYTHON=X) beware that some python bundles
are built in a stripped-down form that is needed for building, not running tests.
See comments in the file below to find out which version of python is compatible with tests.
https://a.yandex-team.ru/arc/trunk/arcadia/build/platform/python/resources.inc
:return: absolute path to python
"""
return _get_ya_plugin_instance().python_path
def valgrind_path():
"""
Get path to valgrind
:return: absolute path to valgrind
"""
return _get_ya_plugin_instance().valgrind_path
def get_param(key, default=None):
"""
Get arbitrary parameter passed via command line
:param key: key
:param default: default value
:return: parameter value or the default
"""
return _get_ya_plugin_instance().get_param(key, default)
def get_param_dict_copy():
"""
Return copy of dictionary with all parameters. Changes to this dictionary do *not* change parameters.
:return: copy of dictionary with all parameters
"""
return _get_ya_plugin_instance().get_param_dict_copy()
@not_test
def test_output_path(path=None):
"""
Get dir in the suite output_path for the current test case
"""
test_out_dir = os.path.splitext(_get_ya_config().current_test_log_path)[0]
try:
os.makedirs(test_out_dir)
except OSError as e:
if e.errno != errno.EEXIST:
raise
return _join_path(test_out_dir, path)
def project_path(path=None):
"""
Get path in build root relating to build_root/project path
"""
return _join_path(os.path.join(build_path(), context.project_path), path)
def gdb_path():
"""
Get path to the gdb
"""
return _get_ya_plugin_instance().gdb_path
def c_compiler_path():
"""
Get path to the gdb
"""
return os.environ.get("YA_CC")
def get_yt_hdd_path(path=None):
if 'HDD_PATH' in os.environ:
return _join_path(os.environ['HDD_PATH'], path)
def cxx_compiler_path():
"""
Get path to the gdb
"""
return os.environ.get("YA_CXX")
def global_resources():
try:
if "YA_GLOBAL_RESOURCES" in os.environ:
return json.loads(os.environ.get("YA_GLOBAL_RESOURCES"))
else:
return _get_ya_plugin_instance().get_context("ya_global_resources")
except (TypeError, ValueError):
return {}
def _register_core(name, binary_path, core_path, bt_path, pbt_path):
config = _get_ya_config()
with _lock:
if not hasattr(config, 'test_cores_count'):
config.test_cores_count = 0
config.test_cores_count += 1
count_str = '' if config.test_cores_count == 1 else str(config.test_cores_count)
log_entry = config.test_logs[config.current_item_nodeid]
if binary_path:
log_entry['{} binary{}'.format(name, count_str)] = binary_path
if core_path:
log_entry['{} core{}'.format(name, count_str)] = core_path
if bt_path:
log_entry['{} backtrace{}'.format(name, count_str)] = bt_path
if pbt_path:
log_entry['{} backtrace html{}'.format(name, count_str)] = pbt_path
@not_test
def test_source_path(path=None):
return _join_path(os.path.join(source_path(), context.project_path), path)
class Context(object):
"""
Runtime context
"""
@property
def build_type(self):
return _get_ya_plugin_instance().get_context("build_type")
@property
def project_path(self):
return _get_ya_plugin_instance().get_context("project_path")
@property
def test_stderr(self):
return _get_ya_plugin_instance().get_context("test_stderr")
@property
def test_debug(self):
return _get_ya_plugin_instance().get_context("test_debug")
@property
def test_traceback(self):
return _get_ya_plugin_instance().get_context("test_traceback")
@property
def test_name(self):
return _get_ya_config().current_test_name
@property
def test_tool_path(self):
return _get_ya_plugin_instance().get_context("test_tool_path")
@property
def sanitize(self):
"""
Detect if current test run is under sanitizer
:return: one of `None`, 'address', 'memory', 'thread', 'undefined'
"""
return _get_ya_plugin_instance().get_context("sanitize")
@property
def flags(self):
_flags = _get_ya_plugin_instance().get_context("flags")
if _flags:
_flags_dict = dict()
for f in _flags:
key, value = f.split('=', 1)
_flags_dict[key] = value
return _flags_dict
else:
return dict()
def get_context_key(self, key):
return _get_ya_plugin_instance().get_context(key)
context = Context()
|