-
Notifications
You must be signed in to change notification settings - Fork 50
/
conftest.py
118 lines (100 loc) · 4.71 KB
/
conftest.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
""" Example from pytest documentation
https://pytest.org/en/stable/example/simple.html#incremental-testing-test-steps
"""
from typing import Dict, Tuple
import pytest
import os.path as op
import hnn_core
from hnn_core import read_params, default_network, simulate_dipole
from hnn_core import MPIBackend, JoblibBackend
# store history of failures per test class name and per index in parametrize
# (if parametrize used)
_test_failed_incremental: Dict[str, Dict[Tuple[int, ...], str]] = {}
def pytest_runtest_makereport(item, call):
if "incremental" in item.keywords:
# incremental marker is used
# The following condition was modifed from the example linked above.
# We don't want to step out of the incremental testing block if
# a previous test was marked "Skipped". For instance if MPI tests
# are skipped because mpi4py is not installed, still continue with
# all other tests that do not require mpi4py
if call.excinfo is not None and not call.excinfo.typename == "Skipped":
# the test has failed, but was not skiped
# retrieve the class name of the test
cls_name = str(item.cls)
# retrieve the index of the test (if parametrize is used in
# combination with incremental)
parametrize_index = (
tuple(item.callspec.indices.values())
if hasattr(item, "callspec")
else ()
)
# retrieve the name of the test function
test_name = item.originalname or item.name
# store in _test_failed_incremental the original name of the
# failed test
_test_failed_incremental.setdefault(cls_name, {}).setdefault(
parametrize_index, test_name
)
def pytest_runtest_setup(item):
if "incremental" in item.keywords:
# retrieve the class name of the test
cls_name = str(item.cls)
# check if a previous test has failed for this class
if cls_name in _test_failed_incremental:
# retrieve the index of the test (if parametrize is used in
# combination with incremental)
parametrize_index = (
tuple(item.callspec.indices.values())
if hasattr(item, "callspec")
else ()
)
# retrieve the name of the first test function to fail for this
# class name and index
test_name = _test_failed_incremental[cls_name].get(
parametrize_index, None)
# if name found, test has failed for the combination of class name
# and test name
if test_name is not None:
pytest.xfail("previous test failed ({})".format(test_name))
@pytest.fixture(scope='module')
def run_hnn_core_fixture():
def _run_hnn_core_fixture(backend=None, n_procs=None, n_jobs=1,
reduced=False, record_vsoma=False,
record_isoma=False, postproc=False,
electrode_array=None):
hnn_core_root = op.dirname(hnn_core.__file__)
# default params
params_fname = op.join(hnn_core_root, 'param', 'default.json')
params = read_params(params_fname)
if reduced:
params.update({'N_pyr_x': 3,
'N_pyr_y': 3,
'tstop': 40,
't_evprox_1': 5,
't_evdist_1': 10,
't_evprox_2': 20,
'N_trials': 2})
net = default_network(params, add_drives_from_params=True)
if electrode_array is not None:
for name, positions in electrode_array.items():
net.add_electrode_array(name, positions)
# number of trials simulated
for drive in net.external_drives.values():
assert len(drive['events']) == params['N_trials']
if backend == 'mpi':
with MPIBackend(n_procs=n_procs, mpi_cmd='mpiexec'):
dpls = simulate_dipole(net, record_vsoma=record_isoma,
record_isoma=record_vsoma,
postproc=postproc)
elif backend == 'joblib':
with JoblibBackend(n_jobs=n_jobs):
dpls = simulate_dipole(net, record_vsoma=record_isoma,
record_isoma=record_vsoma,
postproc=postproc)
else:
dpls = simulate_dipole(net, record_vsoma=record_isoma,
record_isoma=record_vsoma,
postproc=postproc)
return dpls, net
return _run_hnn_core_fixture