Merge pull request #178 from vinferrer/txt_test

[TEST] Speed-ups/improvements to test_txt.py

phys2bids/tests/test_txt.py

@@ -3,108 +3,110 @@
 from phys2bids.interfaces import txt
 from pytest import raises
 import wget
+import pytest
 
 
-def test_read_header_and_channels():
-    url = 'https://osf.io/sdz4n/download'  # url to Test_belt_pulse_samefreq.txt
+@pytest.fixture
+def samefreq_acq_file():
+    url = 'https://osf.io/4yudk/download'  # url to Test_belt_pulse_samefreq.txt
     test_path = resource_filename('phys2bids', 'tests/data')
-    test_filename = 'Test_belt_pulse_samefreq.txt'
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
-    chtrig = 2
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
-    assert len(header) == 16  # check proper header lenght
-    assert len(channels) == 1336816  # check proper number of timepoints
-    assert len(header[-1]) == 6  # check extra line is deleted
-    os.remove(test_full_path)
-    # load file with comment
+    test_filename = 'Test_belt_pulse_samefreq_short.txt'
+    test_full_path1 = os.path.join(test_path, test_filename)
+    if not os.path.isfile(test_full_path1):
+        wget.download(url, test_full_path1)
+    # chtrig = 2
+    return test_full_path1
+
+
+@pytest.fixture
+def multi_lab_file():
     url = 'https://osf.io/q4x2f/download'
     # url to Test_2minRest_trig_multifreq_header_comment.txt
     test_filename = 'Test_2minRest_trig_multifreq_header_comment.txt'
     test_path = resource_filename('phys2bids', 'tests/data')
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
-    chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
-    assert len(channels[152109 - 9]) == 6  # check the comment has been eliminated
-    os.remove(test_full_path)
+    test_full_path2 = os.path.join(test_path, test_filename)
+    if not os.path.isfile(test_full_path2):
+        wget.download(url, test_full_path2)
+    # chtrig = 1
+    return test_full_path2
 
 
-def test_populate_phys_input():
-    # testing for AcqKnoledge files
-    url = 'https://osf.io/sdz4n/download'
-    test_filename = 'Test_belt_pulse_samefreq.txt'
+@pytest.fixture
+def no_time_file():
+    # use file without time column
+    # # url to the file Test2_trigger_CO2_O2_pulse_1000Hz_534TRs_no_time.txt
+    url = 'https://osf.io/u5dq8/download'
+    test_filename = 'Test2_trigger_CO2_O2_pulse_1000Hz_534TRs_no_time.txt'
     test_path = resource_filename('phys2bids', 'tests/data')
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
-    chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
-    assert 'acq' in header[0][0]
-    txt.populate_phys_input(test_full_path, chtrig)
-    os.remove(test_full_path)
+    test_full_path3 = os.path.join(test_path, test_filename)
+    if not os.path.isfile(test_full_path3):
+        wget.download(url, test_full_path3)
+    # chtrig = 0
+    return test_full_path3
+# header_T2ntime, channels_T2ntime = txt.read_header_and_channels(test_full_path3, chtrig)
+
+
+@pytest.fixture
+def test_read_header_and_channels(samefreq_acq_file, multi_lab_file):
+    header_acq, channels_acq = txt.read_header_and_channels(samefreq_acq_file, 2)
+    header_lab, channels_lab = txt.read_header_and_channels(multi_lab_file, 1)
+    assert len(header_acq) == 8  # check proper header lenght
+    assert len(channels_acq) == 1048560  # check proper number of timepoints
+    assert len(header_acq[-1]) == 2  # check extra line is deleted
+    # load file with comment
+    # url to Test_2minRest_trig_multifreq_header_comment.txt
+    assert len(channels_lab[152109 - 9]) == 6  # check the comment has been eliminated
+    return header_acq, channels_acq, header_lab, channels_lab
+
+
+def test_populate_phys_input(samefreq_acq_file, multi_lab_file, test_read_header_and_channels):
+    # testing for AcqKnoledge files
+    chtrig = 2
+    header_acq = test_read_header_and_channels[0]
+    assert 'acq' in header_acq[0][0]
+    txt.populate_phys_input(samefreq_acq_file, chtrig)
     # testing for labchart files
-    url = 'https://osf.io/q4x2f/download'
-    test_filename = 'Test_2minRest_trig_multifreq_header_comment.txt'
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
-    chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
     # check the printing output according to each format
-    assert 'Interval=' in header[0]
-    os.remove(test_full_path)
+    header_lab = test_read_header_and_channels[2]
+    assert 'Interval=' in header_lab[0]
+    chtrig = 1
+    txt.populate_phys_input(multi_lab_file, chtrig)
 
 
-def test_process_labchart():
-    url = 'https://osf.io/q4x2f/download'
-    test_path = resource_filename('phys2bids', 'tests/data')
-    test_filename = 'Test_2minRest_trig_multifreq_header_comment.txt'
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
+def test_process_labchart(test_read_header_and_channels, no_time_file):
     chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
+    header_time = test_read_header_and_channels[2]
+    channels_time = test_read_header_and_channels[3]
+    header_no_time, channels_no_time = txt.read_header_and_channels(no_time_file, 0)
     # test file with header and seconds as unit:
-    phys_obj = txt.process_labchart(channels, chtrig, header)
+    phys_obj = txt.process_labchart(channels_time, chtrig, header_time)
     assert phys_obj.freq[0] == 1000
     # test when units are min:
-    header[0][1] = '0.001 min'
-    phys_obj = txt.process_labchart(channels, chtrig, header)
+    header_time[0][1] = '0.001 min'
+    phys_obj = txt.process_labchart(channels_time, chtrig, header_time)
     assert phys_obj.freq[0] == 16.666666666666668
     # test when units are hr:
-    header[0][1] = '0.001 hr'
-    phys_obj = txt.process_labchart(channels, chtrig, header)
+    header_time[0][1] = '0.001 hr'
+    phys_obj = txt.process_labchart(channels_time, chtrig, header_time)
     assert phys_obj.freq[0] == 0.2777777777777778
     # test when units are ms:
-    header[0][1] = '1 ms'
-    phys_obj = txt.process_labchart(channels, chtrig, header)
+    header_time[0][1] = '1 ms'
+    phys_obj = txt.process_labchart(channels_time, chtrig, header_time)
     assert phys_obj.freq[0] == 1000
     # test when units are µs:
-    header[0][1] = '1000 µs'
-    phys_obj = txt.process_labchart(channels, chtrig, header)
+    header_time[0][1] = '1000 µs'
+    phys_obj = txt.process_labchart(channels_time, chtrig, header_time)
     assert phys_obj.freq[0] == 1000
-    os.remove(test_full_path)
-    # use file without time column
-    # url to the file Test2_trigger_CO2_O2_pulse_1000Hz_534TRs_no_time.txt
-    url = 'https://osf.io/u5dq8/download'
-    test_filename = 'Test2_trigger_CO2_O2_pulse_1000Hz_534TRs_no_time.txt'
-    test_path = resource_filename('phys2bids', 'tests/data')
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
     chtrig = 0
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
-    phys_obj = txt.process_labchart(channels, chtrig, header)
-    assert len(phys_obj.timeseries) == len(channels[0]) + 1
-    os.remove(test_full_path)
+    phys_obj = txt.process_labchart(channels_no_time, chtrig, header_no_time)
+    assert len(phys_obj.timeseries) == len(channels_no_time[0]) + 1
 
 
-def test_process_acq():
+def test_process_acq(test_read_header_and_channels):
+    header = test_read_header_and_channels[0]
+    channels = test_read_header_and_channels[1]
+    chtrig = 2
     # test file without header
-    url = 'https://osf.io/sdz4n/download'
-    test_path = resource_filename('phys2bids', 'tests/data')
-    test_filename = 'Test_belt_pulse_samefreq.txt'
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
-    chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
     with raises(AttributeError) as errorinfo:
         txt.process_acq(channels, chtrig)
     assert 'not supported' in str(errorinfo.value)
@@ -136,17 +138,17 @@ def test_process_acq():
     with raises(AttributeError) as errorinfo:
         phys_obj = txt.process_acq(channels, chtrig, header)
     assert 'not in a valid AcqKnowledge' in str(errorinfo.value)
-    os.remove(test_full_path)
 
 
-def test_raises():
+def test_raises(test_read_header_and_channels):
     # testing error for files without header for populate_phys_input
     # url to the file Test_belt_pulse_samefreq_no_header.txt
     url = 'https://osf.io/sre3h/download'
     test_filename = 'Test_belt_pulse_samefreq_no_header.txt'
     test_path = resource_filename('phys2bids', 'tests/data')
     test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
+    if not os.path.isfile(test_full_path):
+        wget.download(url, test_full_path)
     chtrig = 2
     with raises(AttributeError) as errorinfo:
         txt.populate_phys_input(test_full_path, chtrig)
@@ -159,13 +161,9 @@ def test_raises():
     os.remove(test_full_path)
     # now for labchart read
     # testing file already downloaded in the other tests
-    url = 'https://osf.io/q4x2f/download'
-    test_path = resource_filename('phys2bids', 'tests/data')
-    test_filename = 'Test_2minRest_trig_multifreq_header_comment.txt'
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
+    header = test_read_header_and_channels[2]
+    channels = test_read_header_and_channels[3]
     chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
     with raises(AttributeError) as errorinfo:
         txt.process_labchart(test_full_path, chtrig)
     assert 'not supported' in str(errorinfo.value)
@@ -174,18 +172,14 @@ def test_raises():
     with raises(AttributeError) as errorinfo:
         txt.process_labchart(channels, chtrig, header)
     assert 'not in a valid LabChart' in str(errorinfo.value)
-    os.remove(test_full_path)
 
 
-def test_multifreq():
-    url = 'https://osf.io/q4x2f/download'
-    test_path = resource_filename('phys2bids', 'tests/data')
-    test_filename = 'Test_2minRest_trig_multifreq_header_comment.txt'
-    test_full_path = os.path.join(test_path, test_filename)
-    wget.download(url, test_full_path)
+def test_multifreq(test_read_header_and_channels):
+    header = test_read_header_and_channels[2]
+    channels = test_read_header_and_channels[3]
     chtrig = 1
-    header, channels = txt.read_header_and_channels(test_full_path, chtrig)
     phys_obj = txt.process_labchart(channels, chtrig, header)
     new_freq = txt.check_multifreq(phys_obj.timeseries, [phys_obj.freq[0]] * len(phys_obj.freq))
     assert new_freq[-3] == 40
-    os.remove(test_full_path)
+    assert new_freq[-2] == 500
+    assert new_freq[-4] == 100