forked from Inria-Empenn/narps_open_pipelines
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_team_C88N.py
134 lines (104 loc) · 4.94 KB
/
test_team_C88N.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
#!/usr/bin/python
# coding: utf-8
""" Tests of the 'narps_open.pipelines.team_C88N' module.
Launch this test with PyTest
Usage:
======
pytest -q test_team_C88N.py
pytest -q test_team_C88N.py -k <selected_test>
"""
from os.path import join
from pytest import helpers, mark
from numpy import isclose
from nipype import Workflow
from nipype.interfaces.base import Bunch
from narps_open.utils.configuration import Configuration
from narps_open.pipelines.team_C88N import PipelineTeamC88N
class TestPipelinesTeamC88N:
""" A class that contains all the unit tests for the PipelineTeamC88N class."""
@staticmethod
@mark.unit_test
def test_create():
""" Test the creation of a PipelineTeamC88N object """
pipeline = PipelineTeamC88N()
# 1 - check the parameters
assert pipeline.fwhm == 8.0
assert pipeline.team_id == 'C88N'
# 2 - check workflows
assert pipeline.get_preprocessing() is None
assert pipeline.get_run_level_analysis() is None
assert isinstance(pipeline.get_subject_level_analysis(), Workflow)
group_level = pipeline.get_group_level_analysis()
assert len(group_level) == 5
for sub_workflow in group_level:
assert isinstance(sub_workflow, Workflow)
@staticmethod
@mark.unit_test
def test_outputs():
""" Test the expected outputs of a PipelineTeamC88N object """
pipeline = PipelineTeamC88N()
# 1 - 1 subject outputs
pipeline.subject_list = ['001']
helpers.test_pipeline_outputs(pipeline, [0, 0, 8, 53, 18])
# 2 - 4 subjects outputs
pipeline.subject_list = ['001', '002', '003', '004']
helpers.test_pipeline_outputs(pipeline, [0, 0, 32, 53, 18])
@staticmethod
@mark.unit_test
def test_subject_information():
""" Test the get_subject_information method """
# Test with 'gain'
test_event_file = join(
Configuration()['directories']['test_data'], 'pipelines', 'events.tsv')
information = PipelineTeamC88N.get_subject_information(
[test_event_file, test_event_file],
'gain'
)[0]
assert isinstance(information, Bunch)
assert information.conditions == ['trial']
reference_durations = [[0.0, 0.0, 0.0, 0.0]]
assert len(reference_durations) == len(information.durations)
for reference_array, test_array in zip(reference_durations, information.durations):
assert isclose(reference_array, test_array).all()
reference_onsets = [[4.071, 11.834, 27.535, 36.435]]
assert len(reference_onsets) == len(information.onsets)
for reference_array, test_array in zip(reference_onsets, information.onsets):
assert isclose(reference_array, test_array).all()
paramateric_modulation = information.pmod[0]
assert isinstance(paramateric_modulation, Bunch)
assert paramateric_modulation.name == ['loss', 'gain']
assert paramateric_modulation.poly == [1, 1]
reference_param = [[6.0, 14.0, 15.0, 17.0], [14.0, 34.0, 10.0, 16.0]]
assert len(reference_param) == len(paramateric_modulation.param)
for reference_array, test_array in zip(reference_param, paramateric_modulation.param):
assert isclose(reference_array, test_array).all()
# Test with 'loss'
test_event_file = join(
Configuration()['directories']['test_data'], 'pipelines', 'events.tsv')
information = PipelineTeamC88N.get_subject_information(
[test_event_file, test_event_file],
'loss'
)[0]
assert isinstance(information, Bunch)
assert information.conditions == ['trial']
reference_durations = [[0.0, 0.0, 0.0, 0.0]]
assert len(reference_durations) == len(information.durations)
for reference_array, test_array in zip(reference_durations, information.durations):
assert isclose(reference_array, test_array).all()
reference_onsets = [[4.071, 11.834, 27.535, 36.435]]
assert len(reference_onsets) == len(information.onsets)
for reference_array, test_array in zip(reference_onsets, information.onsets):
assert isclose(reference_array, test_array).all()
paramateric_modulation = information.pmod[0]
assert isinstance(paramateric_modulation, Bunch)
assert paramateric_modulation.name == ['gain', 'loss']
assert paramateric_modulation.poly == [1, 1]
reference_param = [[14.0, 34.0, 10.0, 16.0], [6.0, 14.0, 15.0, 17.0]]
assert len(reference_param) == len(paramateric_modulation.param)
for reference_array, test_array in zip(reference_param, paramateric_modulation.param):
assert isclose(reference_array, test_array).all()
@staticmethod
@mark.pipeline_test
def test_execution():
""" Test the execution of a PipelineTeamC88N and compare results """
helpers.test_pipeline_evaluation('C88N')