forked from smarie/python-pytest-harvest
-
Notifications
You must be signed in to change notification settings - Fork 0
/
results_session.py
614 lines (513 loc) · 27.7 KB
/
results_session.py
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
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
from distutils.version import LooseVersion
import pytest
import sys
from collections import OrderedDict, namedtuple
from itertools import chain
from six import string_types
pytest53 = LooseVersion(pytest.__version__) >= LooseVersion("5.3.0")
if pytest53:
def is_lazy_value_or_tupleitem_with_int_base(o):
return False
else:
# In this version of pytest, pytest-cases creates LazyValue objects that inherit from int, which makes pandas
# believe that their dtype should be int instead of object when creating a dataframe. We'll remove the int base here
try:
from pytest_cases.common_pytest_lazy_values import is_lazy
def is_lazy_value_or_tupleitem_with_int_base(o):
return is_lazy(o) and isinstance(o, int)
except ImportError: # noqa
def is_lazy_value_or_tupleitem_with_int_base(o):
return False
try: # python 3.5+
from typing import Union, Iterable, Mapping, Any
except ImportError:
pass
from pytest_harvest.common import HARVEST_PREFIX
from _pytest.doctest import DoctestItem
PYTEST_OBJ_NAME = 'pytest_obj'
def get_session_synthesis_dct(session_or_request,
test_id_format='full', # type: str
status_details=False, # type: bool
durations_in_ms=False, # type: bool
pytest_prefix=None, # type: bool
filter=None, # type: Any
filter_incomplete=True, # type: bool
flatten=False, # type: bool
fixture_store=None, # type: Union[Mapping[str, Any], Iterable[Mapping[str, Any]]]
flatten_more=None # type: Union[str, Iterable[str], Mapping[str, str]]
):
# type: (...) -> Mapping[str, Mapping[str, Any]]
"""
Returns a dictionary containing a synthesis of what is available currently in the provided `pytest` `session`
object.
For each entry, the key is the test id, and the value is a dictionary containing:
- `'pytest_obj'`: the object under test, typically a test function
- `'pytest_status'`: the overall status (`'failing'`, `'skipped'`, `'passed'`)
- `'pytest_duration'`: the duration of the `'call'` step. By default this is the pytest unit (s) but if you set
`durations_in_ms=True` it becomes (ms)
- `'pytest_status_details'`: a dictionary containing step-by-step status details for all pytest steps (`'setup'`,
`'call'`, `'teardown'`). This is only included if `status_details=True` (not by default)
It is possible to process the test id (the keys) using the `test_id_format` option. Let's assume that the id is
`pytest_steps/tests_raw/test_wrapped_in_class.py::TestX::()::test_easy[p1-p2]`. Here are the returned test ids
depending on the selected `test_id_format`
- `'function'` will return `test_easy[p1-p2]`
- `'class'` will return `TestX::()::test_easy[p1-p2]`
- `'module'` will return `test_wrapped_in_class.py::TestX::()::test_easy[...]`
- `'full'` will return the original id (this is the default behaviour)
In addition one can provide a custom string handling function that will be called for each test id to process.
The 'pytest' prefix in front of all these items (except `pytest_obj`) is by default added in non-flatten mode and
removed in flatten mode. To force one of these you can set `pytest_prefix` to `True` or `False`.
An optional `filter` can be provided, that can be a singleton or iterable of pytest objects (typically test
functions) and/or module names.
If this method is called before the end of the pytest session, some nodes might be incomplete, i.e. they will not
have data for the three stages (setup/call/teardown). By default these nodes are filtered out but you can set
`filter_incomplete=False` to make them appear. They will have a special 'pending' synthesis status.
An optional collection of storage objects can be provided, so as to merge them into the resulting dictionary.
Finally a `flatten_output` option allows users to get a flat dictionary output instead of nested status details,
parameters dict, and storage dicts.
:param session: a pytest session object.
:param test_id_format: one of 'function', 'class', 'module', or 'full' (default), or a custom test id processing
function.
:param status_details: a flag indicating if pytest status details per stage (setup/call/teardown) should be
included. Default=`False`: only the pytest status summary is provided.
:param durations_in_ms: by default `pytest` measures durations in seconds so they are outputed in this unit. You
can turn the flag to True to output milliseconds instead.
:param pytest_prefix: to add (True) or remove (False) the 'pytest_' prefix in front of status, duration and status
details. Typically useful in flatten mode when the names are not ambiguous. By default it is None, which
means =(not flatten)
:param filter: a singleton or iterable of pytest objects on which to filter the returned dict on (the returned
items will only by pytest nodes for which the pytest object is one of the ones provided). One can also use
module names.
:param filter_incomplete: a boolean indicating if incomplete nodes (without the three stages setup/call/teardown)
should appear in the results (False) or not (True, default).
:param flatten: a boolean (default `False`) indicating if the resulting dictionary should be flattened. If it
set to `True`, the 3 nested dictionaries (pytest status details, parameters, and optionally storages)
will have their contents directly copied in the first level (with a prefix added in case of pytest status
details).
:param fixture_store: a singleton or iterable containing dict-like fixture storage objects (see
`@saved_fixture` and `create_results_bag_fixture`). If flatten=`False` the contents of these dictionaries will
be added to the output in a dedicated 'fixtures' entry. If flatten=True all of their contents will be included
directly. A fixture name can also be provided.
:param flatten_more: a singleton, iterable or dictionary containing fixture names to flatten one level more in case
flatten=True. If a dictionary is provided, the key should be the fixture name, and the value should be a prefix
used for flattening its contents
:return: a dictionary where the keys are pytest node ids. Each value is also a dictionary, containing information
available from pytest concerning the test node, and optionally storage contents if `storage_dcts` is provided.
"""
res_dct = OrderedDict()
# extract session if needed
if hasattr(session_or_request, 'session') and session_or_request.session is not session_or_request:
request = session_or_request
session = request.session
else:
request = None
session = session_or_request
# Optional test id formatter
if test_id_format == 'function':
def test_id_format(test_id):
"""
from: path/to/test_file.py::TestClass::test_fun[param-param2]
to: test_fun[param-param2]
"""
# Old
# return test_id.split('::')[-1]
# New: resistant to '::' in param names
try:
# is there a bracket indicating parameters (therefore possibly custom ids)
_idx = test_id.index('[')
except ValueError:
return test_id.split('::')[-1]
else:
return test_id[:_idx].split('::')[-1] + test_id[_idx:]
elif test_id_format == 'class':
def test_id_format(test_id):
"""
from: path/to/test_file.py::TestClass::test_fun[param-param2]
to: TestClass::test_fun[param-param2]
note: if no class is there, this will be function
"""
return '::'.join(test_id.split('::')[1:])
elif test_id_format == 'module':
def test_id_format(test_id):
"""
from: path/to/test_file.py::TestClass::test_fun[param-param2]
to: test_file.py::TestClass::test_fun[param-param2]
"""
return test_id.replace('\\', '/').split('/')[-1]
elif test_id_format == 'full':
def test_id_format(test_id):
""" path/to/test_file.py::TestClass::test_fun[param-param2] """
return test_id
elif callable(test_id_format):
pass # use it directly
else:
raise ValueError("`test_id_format` should be one of {'function', 'class', 'module', 'full'} or be a custom "
"function. Found '%s'" % test_id_format)
# Optional 'pytest_' prefix in front of status and duration
if pytest_prefix is None:
pytest_prefix = not flatten
if pytest_prefix:
pytest_prefix = 'pytest_'
else:
pytest_prefix = ''
# Optional filter
filtered_items = filter_session_items(session, filter)
# fixture store check
if fixture_store is not None:
try:
fixture_store_items = fixture_store.items()
except AttributeError:
# not a dict: an iterable of dict
fixture_store_items = list(chain(store.items() for store in fixture_store))
# flatten_more check
if flatten_more is not None:
if isinstance(flatten_more, dict):
flatten_more_prefixes_dct = flatten_more.items()
elif isinstance(flatten_more, string_types):
# single name ?
flatten_more_prefixes_dct = {flatten_more: ''}
else:
# iterable ?
flatten_more_prefixes_dct = {k: '' for k in flatten_more}
# For each item add an entry
for item in filtered_items:
item_dct = OrderedDict()
# Fill the dictionary with information about this test node
# -- test object
item_dct[PYTEST_OBJ_NAME] = item.obj
# -- test status: this information is available thanks to our hook in plugin.py
(test_status, test_duration), status_dct = get_pytest_status(item, durations_in_ms=durations_in_ms,
current_request=request)
if test_status not in {'pending', 'unknown'} or not filter_incomplete:
# -- parameters (of tests and fixtures)
param_dct = get_pytest_params(item)
# Fill according to mode
item_dct[pytest_prefix + "status"] = test_status
item_dct[pytest_prefix + "duration_" + ('ms' if durations_in_ms else 's')] = test_duration
if flatten:
if status_details:
for k, v in status_dct.items():
item_dct[pytest_prefix + "status__" + k] = v
item_dct.update(param_dct)
else:
if status_details:
item_dct[pytest_prefix + "status_details"] = status_dct
item_dct[pytest_prefix + "params"] = param_dct
# -- fixture storages
# For info: https://docs.pytest.org/en/latest/_modules/_pytest/runner.html
# used_fixtures = sorted(item._fixtureinfo.name2fixturedefs.keys())
# if used_fixtures:
# tw.write(" (fixtures used: {})".format(", ".join(used_fixtures)))
if fixture_store is not None:
if not flatten:
item_dct['fixtures'] = OrderedDict()
for fixture_name, fixture_dct in fixture_store_items:
# if this fixture is available for this test
if item.nodeid in fixture_dct:
# get the fixture value for this test
fix_val = fixture_dct[item.nodeid]
# store it in the appropriate format
if flatten:
if flatten_more is not None and fixture_name in flatten_more_prefixes_dct:
prefix = flatten_more_prefixes_dct[fixture_name]
# flatten more
for k, v in fix_val.items():
item_dct[prefix + k] = v
else:
item_dct[fixture_name] = fix_val
else:
item_dct['fixtures'][fixture_name] = fix_val
# Finally store in the main dictionary
res_dct[test_id_format(item.nodeid)] = item_dct
return res_dct
def filter_session_items(session,
filter=None, # type: Any
):
"""
Filters pytest session item in the provided `session`.
An optional `filter` can be provided, that can be a singleton or iterable of pytest objects (typically test
functions) and/or module names.
Used in `get_session_synthesis_dct`.
:param session: a pytest session
:param filter: a singleton or iterable of pytest objects on which to filter the returned dict on (the returned
items will only by pytest nodes for which the pytest object is one of the ones provided). One can also use
module names.
:return: an iterable containing possibly filtered session items
"""
if filter is not None:
filterset = _get_filterset(filter)
filtered_items = tuple(item for item in session.items if _pytest_item_matches_filter(item, filterset))
else:
filtered_items = session.items
return filtered_items
def get_all_pytest_param_names(session,
filter=None, # type: Any
filter_incomplete=False, # type: bool
):
"""
Returns the list of all unique parameter names used in all items in the provided session, with given filter.
An optional `filter` can be provided, that can be a singleton or iterable of pytest objects (typically test
functions) and/or module names.
If this method is called before the end of the pytest session, some nodes might be incomplete, i.e. they will not
have data for the three stages (setup/call/teardown). By default these nodes are filtered out but you can set
`filter_incomplete=False` to make them appear. They will have a special 'pending' synthesis status.
:param session: a pytest session object.
:param filter: a singleton or iterable of pytest objects on which to filter the returned dict on (the returned
items will only by pytest nodes for which the pytest object is one of the ones provided). One can also use
modules.
:param filter_incomplete: a boolean indicating if incomplete nodes (without the three stages setup/call/teardown)
should appear in the results (False) or not (True). Note: by default incomplete nodes DO APPEAR (this is
different from get_session_synthesis_dct behaviour)
:return: a list of parameter names corresponding to the desired filters
"""
dset = set()
# relies on the fact that dset.add() always returns None
# thanks https://stackoverflow.com/questions/6197409/ordered-sets-python-2-7
return [k for item in filter_session_items(session, filter=filter)
for k in get_pytest_params(item)
if k not in dset and not (filter_incomplete and is_pytest_incomplete(item))
and not dset.add(k)]
def get_all_pytest_fixture_names(session,
filter=None, # type: Any
filter_incomplete=False, # type: bool
):
"""
Returns the list of all unique fixture names used in all items in the provided session, with given filter.
An optional `filter` can be provided, that can be a singleton or iterable of pytest objects (typically test
functions) and/or module names.
If this method is called before the end of the pytest session, some nodes might be incomplete, i.e. they will not
have data for the three stages (setup/call/teardown). By default these nodes are filtered out but you can set
`filter_incomplete=False` to make them appear. They will have a special 'pending' synthesis status.
:param session: a pytest session object.
:param filter: a singleton or iterable of pytest objects on which to filter the returned dict on (the returned
items will only by pytest nodes for which the pytest object is one of the ones provided). One can also use
modules.
:param filter_incomplete: a boolean indicating if incomplete nodes (without the three stages setup/call/teardown)
should appear in the results (False) or not (True). Note: by default incomplete nodes DO APPEAR (this is
different from get_session_synthesis_dct behaviour)
:return: a list of fixture names corresponding to the desired filters
"""
dset = set()
# relies on the fact that dset.add() always returns None
# thanks https://stackoverflow.com/questions/6197409/ordered-sets-python-2-7
return [k for item in filter_session_items(session, filter=filter)
for k in get_pytest_fixture_names(item)
if k not in dset and not (filter_incomplete and is_pytest_incomplete(item))
and not dset.add(k)]
# ------------ item-related -------------
def pytest_item_matches_filter(item, filter):
"""
Returns True if pytest session item `item` matches filter `filter`, `False` otherwise.
:param item: an item inside a pytest session
:param filter:
:return:
"""
filterset = _get_filterset(filter)
return _pytest_item_matches_filter(item, filterset)
def _pytest_item_matches_filter(item, filterset):
"""Internal method used to check if item matches filter set"""
item_obj = item.obj
if item_obj in filterset:
return True
# support class methods: the item object can be a bound method while the filter is maybe not
elif item_obj is None:
# This can happen with DoctestItem
return False
elif _is_unbound_present(item_obj, filterset):
return True
elif any(item_obj.__module__ == f for f in filterset):
return True
else:
return False
def _is_unbound_present(item_obj, filterset):
"""
Returns True if item_obj is a bound method and that its unbound version is in filterset
:param item_obj:
:param filterset:
:return:
"""
if sys.version_info >= (3,):
# Python 3
not_bound_fct = getattr(item_obj, '__func__', None)
if not_bound_fct is None:
return False
else:
return not_bound_fct in filterset
else:
# Python 2 has the concept of "unbound method" and behaves a bit diferently
# see https://stackoverflow.com/questions/14574641/python-get-unbound-class-method
not_bound_fct = getattr(item_obj, 'im_func', None)
if not_bound_fct is None:
return False
elif not_bound_fct in filterset:
return True
else:
# maybe the filterset contains a "truly unbound" method ?
return not_bound_fct in {getattr(f, 'im_func', None) for f in filterset}
def _get_pytest_status_keys(item):
return [k for k in vars(item) if k.startswith(HARVEST_PREFIX)]
def is_pytest_incomplete(item):
"""
Returns `True` if a pytest item is incomplete - in other words if at least one of the 3 steps (setup/call/teardown)
is missing from the available pytest report attached to this item.
:param item:
:return:
"""
return len(_get_pytest_status_keys(item)) < 3
def get_pytest_status(item, durations_in_ms=False, current_request=None):
"""
Returns a dictionary containing item's pytest status (success/skipped/failed, duration converted to ms) for
each pytest phase, and a tuple synthesizing the information.
The synthesis status contains the worst status of all phases (setup/call/teardown), or 'pending' if there are less
than 3 phases.
The synthesis duration is equal to the duration of the 'call' phase (not to the sum of all phases: indeed, we are
mostly interested in the test call itself).
:param item: a pytest session.item
:param durations_in_ms: by default `pytest` measures durations in seconds so they are outputed in this unit. You
can turn the flag to True to output milliseconds instead.
:param current_request: if a non-None `request` is provided and the item is precisely the one from the request,
then the status will be 'pending'
:return: a tuple ((test_status, test_duration), status_dct)
"""
# the status keys that have been stored by our plugin.py module
status_keys = _get_pytest_status_keys(item)
if len(status_keys) == 0:
if current_request is not None and current_request.node == item:
# do not raise a warning: it is normal that there is no information, the node is being called.
test_status = 'pending'
else:
# warn("[pytest-harvest] Test items status is not available. You should maybe install pytest-harvest with "
# "pip. If it is already the case, you case try to force-use it by adding "
# "`pytest_plugins = ['harvest']` to your conftest.py. But for normal use this should not be required,"
# "installing with pip should be enough.")
test_status = 'unknown'
test_duration = None
status_dct = dict()
else:
# adjust duration factor according to target unit
duration_factor = (1000 if durations_in_ms else 1)
# create the status dictionary for that item
status_dct = OrderedDict()
test_status = 'passed'
test_duration = None
for k in status_keys:
statusreport = getattr(item, k)
status_dct[statusreport.when] = (statusreport.outcome, statusreport.duration * duration_factor)
# update global test status
if test_status == 'passed' \
or (test_status == 'skipped' and statusreport.outcome != 'passed'):
test_status = statusreport.outcome
# global test duration is the duration of the "call" step only
if statusreport.when == "call":
test_duration = statusreport.duration * duration_factor
if len(status_keys) < 3:
# this is an incomplete test
test_status = 'pending'
return (test_status, test_duration), status_dct
def get_pytest_param_names(item):
""" Returns a list containing a pytest session item's parameters """
return list(get_pytest_params(item).keys())
def get_pytest_params(item):
""" Returns a dictionary containing a pytest session item's parameters """
if isinstance(item, _MinimalItem):
# Our special _MinimalItem object - when xdist is used and worker states have been saved + restored
return item.get_pytest_params()
elif isinstance(item, DoctestItem):
# No fixtures or parameters
return OrderedDict()
else:
param_dct = OrderedDict()
for param_name in item.fixturenames: # note: item.funcargnames gives the exact same list
if hasattr(item, 'callspec'):
if param_name in item.callspec.params:
param_value = item.callspec.params[param_name]
if is_lazy_value_or_tupleitem_with_int_base(param_value):
# remove the int base so that pandas does not interprete it as an int.
param_value = param_value.clone(remove_int_base=True)
if hasattr(pytest, "version_tuple") and pytest.version_tuple >= (8, 1):
fixturedefs = item.session._fixturemanager.getfixturedefs(param_name, item)
else:
fixturedefs = item.session._fixturemanager.getfixturedefs(param_name, item.nodeid)
if fixturedefs is not None:
# Fixture parameters have the same name than the fixtures themselves! change it
param_dct[param_name + '_param'] = param_value
else:
# Non-fixture parameter: ok
param_dct[param_name] = param_value
else:
# this is a non-parametrized fixture: it is not available by default in item, this is normal pytest
# behaviour (hence the @saved_fixture decorator)
pass
return param_dct
def get_pytest_fixture_names(item):
""" Returns a list containing a pytest session item's fixture names """
if isinstance(item, _MinimalItem):
# Our special _MinimalItem object - when xdist is used and worker states have been saved + restored
return item.get_pytest_fixture_names()
else:
# "normal" item
fixture_names = []
for param_name in item.fixturenames: # note: item.funcargnames gives the exact same list
# if hasattr(item, 'callspec'): # NO! it would only return fixtures when they are parametrized
# if param_name in item.callspec.params: NO ! it would only return fixtures when they are *directly* parametrized
if hasattr(pytest, "version_tuple") and pytest.version_tuple >= (8, 1):
fixturedefs = item.session._fixturemanager.getfixturedefs(param_name, item)
else:
fixturedefs = item.session._fixturemanager.getfixturedefs(param_name, item.nodeid)
if fixturedefs is not None:
fixture_names.append(param_name)
return fixture_names
# --- misc
def _get_filterset(filter):
"""
Always returns a set, even if the filter is a string (module name) or single object
:param filter:
:return:
"""
if isinstance(filter, string_types):
filter = {filter}
else:
try:
iter(filter)
filter = set(filter)
except TypeError:
# TypeError: '<....>' object is not iterable
filter = {filter}
return filter
def get_persistable_session_items(session):
"""
Returns a list containing minimal representation of session items, so that the `get_session_synthesis_dct`
function will be able to work if this list is available in place of session.items.
This method is used for persisted state across distributed pytest workers (e.g. xdist)
:param session:
:return:
"""
return [_MinimalItem(item) for item in session.items]
_MinimalCallSpec = namedtuple('_MinimalCallSpec', ('params',))
class _MinimalItem(object):
def __init__(self, item):
# all pytest attributes that we rely upon
self.obj = item.obj
self.nodeid = item.nodeid
if hasattr(item, 'callspec'):
# only keep the params
self.callspec = _MinimalCallSpec(params=item.callspec.params)
# convert these to simple tuples just in case pytest-cases is around and has messed with fixturenames.
try:
# This attribute does not seem to exist in latest pytest
self.funcargnames = tuple(item.funcargnames) # not needed but we can keep it
except AttributeError:
pass
self.fixturenames = tuple(item.fixturenames)
# We do not store the session object so everything that depends on it should be retrieved:
self._validated_pytest_params = get_pytest_params(item)
self._validated_fix_names = get_pytest_fixture_names(item)
# all pytest-harvest attributes
for k, v in vars(item).items():
if k.startswith(HARVEST_PREFIX):
# v is a TestReport object
setattr(self, k, v)
def get_pytest_params(self):
return self._validated_pytest_params
def get_pytest_fixture_names(self):
return self._validated_fix_names