From d4977c04cd0ad068b394d437c0b8d5228ba4c1d6 Mon Sep 17 00:00:00 2001 From: Benjamin Moody Date: Tue, 20 Sep 2022 16:52:57 -0400 Subject: [PATCH] Write test output files to a temporary directory. When writing files as part of test cases, write them to a temporary directory (created in setUpClass and destroyed in tearDownClass) rather than writing them to the current working directory. This allows running tests without modifying the working tree, which means multiple test processes can run simultaneously from the same working tree (e.g., using pytest-xdist). --- tests/test_annotation.py | 42 ++++++----- tests/test_record.py | 149 +++++++++++++++++++++------------------ 2 files changed, 104 insertions(+), 87 deletions(-) diff --git a/tests/test_annotation.py b/tests/test_annotation.py index b282d269..e7d86b50 100644 --- a/tests/test_annotation.py +++ b/tests/test_annotation.py @@ -1,5 +1,6 @@ import os import re +import tempfile import unittest import numpy as np @@ -85,9 +86,11 @@ def test_1(self): pn_annotation.create_label_map() # Test file writing - annotation.wrann(write_fs=True) + annotation.wrann(write_fs=True, write_dir=self.temp_path) write_annotation = wfdb.rdann( - "100", "atr", return_label_elements=["label_store", "symbol"] + os.path.join(self.temp_path, "100"), + "atr", + return_label_elements=["label_store", "symbol"], ) write_annotation.create_label_map() @@ -156,9 +159,11 @@ def test_2(self): pn_annotation.create_label_map() # Test file writing - annotation.wrann(write_fs=True) + annotation.wrann(write_fs=True, write_dir=self.temp_path) write_annotation = wfdb.rdann( - "12726", "anI", return_label_elements=["label_store", "symbol"] + os.path.join(self.temp_path, "12726"), + "anI", + return_label_elements=["label_store", "symbol"], ) write_annotation.create_label_map() @@ -228,9 +233,11 @@ def test_3(self): pn_annotation.create_label_map() # Test file writing - annotation.wrann(write_fs=True) + annotation.wrann(write_fs=True, write_dir=self.temp_path) write_annotation = wfdb.rdann( - "1003", "atr", return_label_elements=["label_store", "symbol"] + os.path.join(self.temp_path, "1003"), + "atr", + return_label_elements=["label_store", "symbol"], ) write_annotation.create_label_map() @@ -247,10 +254,10 @@ def test_4(self): """ annotation = wfdb.rdann("sample-data/huge", "qrs") self.assertEqual(annotation.sample[0], 10000000000) - annotation.wrann() + annotation.wrann(write_dir=self.temp_path) annotation1 = wfdb.rdann("sample-data/huge", "qrs") - annotation2 = wfdb.rdann("huge", "qrs") + annotation2 = wfdb.rdann(os.path.join(self.temp_path, "huge"), "qrs") self.assertEqual(annotation1, annotation2) def test_5(self): @@ -274,22 +281,19 @@ def test_5(self): chan=ann_chan, custom_labels=ann_custom_labels, label_store=ann_label_store, + write_dir=self.temp_path, ) - ann = wfdb.rdann("CustomLabel", "atr") + ann = wfdb.rdann(os.path.join(self.temp_path, "CustomLabel"), "atr") self.assertEqual(ann.symbol, ["z", "l", "v", "r"]) + @classmethod + def setUpClass(cls): + cls.temp_directory = tempfile.TemporaryDirectory() + cls.temp_path = cls.temp_directory.name + @classmethod def tearDownClass(cls): - writefiles = [ - "100.atr", - "1003.atr", - "12726.anI", - "huge.qrs", - "CustomLabel.atr", - ] - for file in writefiles: - if os.path.isfile(file): - os.remove(file) + cls.temp_directory.cleanup() if __name__ == "__main__": diff --git a/tests/test_record.py b/tests/test_record.py index 7c7c5616..40a85f29 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -1,6 +1,7 @@ import datetime import os import shutil +import tempfile import unittest import numpy as np @@ -44,9 +45,11 @@ def test_1a(self): "sample-data/test01_00s", physical=False, return_res=16 ) record_2.sig_name = ["ECG_1", "ECG_2", "ECG_3", "ECG_4"] - record_2.wrsamp() + record_2.wrsamp(write_dir=self.temp_path) record_write = wfdb.rdrecord( - "test01_00s", physical=False, return_res=16 + os.path.join(self.temp_path, "test01_00s"), + physical=False, + return_res=16, ) assert np.array_equal(sig, sig_target) @@ -121,9 +124,11 @@ def test_1c(self): ) # Test file writing - record.wrsamp(expanded=True) + record.wrsamp(write_dir=self.temp_path, expanded=True) record_write = wfdb.rdrecord( - "a103l", physical=False, smooth_frames=False + os.path.join(self.temp_path, "a103l"), + physical=False, + smooth_frames=False, ) assert np.array_equal(sig, sig_target) @@ -180,8 +185,11 @@ def test_1e(self): record_2 = wfdb.rdrecord( "sample-data/n8_evoked_raw_95_F1_R9", physical=False ) - record_2.wrsamp() - record_write = wfdb.rdrecord("n8_evoked_raw_95_F1_R9", physical=False) + record_2.wrsamp(write_dir=self.temp_path) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "n8_evoked_raw_95_F1_R9"), + physical=False, + ) assert np.array_equal(sig, sig_target) assert record.__eq__(record_pn) @@ -251,8 +259,11 @@ def test_read_write_flac(self): ) # Test file writing - record.wrsamp() - record_write = wfdb.rdrecord("flacformats", physical=False) + record.wrsamp(write_dir=self.temp_path) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "flacformats"), + physical=False, + ) assert record == record_write def test_read_write_flac_multifrequency(self): @@ -265,11 +276,14 @@ def test_read_write_flac_multifrequency(self): physical=False, smooth_frames=False, ) - record.wrsamp(expanded=True) + record.wrsamp(write_dir=self.temp_path, expanded=True) # Check that result matches the original record = wfdb.rdrecord("sample-data/mixedsignals", smooth_frames=False) - record_write = wfdb.rdrecord("mixedsignals", smooth_frames=False) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "mixedsignals"), + smooth_frames=False, + ) assert record == record_write def test_read_flac_longduration(self): @@ -358,8 +372,11 @@ def test_2b(self): del record_named.comments[0] # Test file writing - record.wrsamp() - record_write = wfdb.rdrecord("100", physical=False) + record.wrsamp(write_dir=self.temp_path) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "100"), + physical=False, + ) assert np.array_equal(sig, sig_target) assert record.__eq__(record_pn) @@ -379,8 +396,8 @@ def test_2c(self): # Test file writing record.d_signal = record.adc() - record.wrsamp() - record_write = wfdb.rdrecord("100_3chan") + record.wrsamp(write_dir=self.temp_path) + record_write = wfdb.rdrecord(os.path.join(self.temp_path, "100_3chan")) record.d_signal = None assert np.array_equal(sig_round, sig_target) @@ -434,8 +451,11 @@ def test_3a(self): ) # Test file writing - record.wrsamp() - record_write = wfdb.rdrecord("s0010_re", physical=False) + record.wrsamp(write_dir=self.temp_path) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "s0010_re"), + physical=False, + ) assert np.array_equal(sig, sig_target) assert record.__eq__(record_pn) @@ -497,9 +517,12 @@ def test_4a(self): smooth_frames=False, ignore_skew=True, ) - record_no_skew.wrsamp(expanded=True) + record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True) # Read the written record - record_write = wfdb.rdrecord("test01_00s_skewframe", physical=False) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "test01_00s_skewframe"), + physical=False, + ) assert np.array_equal(sig, sig_target) assert record.__eq__(record_write) @@ -533,9 +556,12 @@ def test_4b(self): smooth_frames=False, ignore_skew=True, ) - record_no_skew.wrsamp(expanded=True) + record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True) # Read the written record - record_write = wfdb.rdrecord("03700181", physical=False) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "03700181"), + physical=False, + ) assert np.array_equal(sig, sig_target) assert record.__eq__(record_pn) @@ -572,10 +598,13 @@ def test_4c(self): smooth_frames=False, ignore_skew=True, ) - record_no_skew.wrsamp(expanded=True) + record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True) # Read the written record writesig, writefields = wfdb.rdsamp( - "03700181", channels=[0, 2], sampfrom=1000, sampto=16000 + os.path.join(self.temp_path, "03700181"), + channels=[0, 2], + sampfrom=1000, + sampto=16000, ) assert np.array_equal(sig_round, sig_target) @@ -639,38 +668,13 @@ def test_header_with_non_utf8(self): assert record.units.__eq__(sig_units_target) @classmethod - def tearDownClass(cls): - "Clean up written files" - writefiles = [ - "03700181.dat", - "03700181.hea", - "100.dat", - "100.hea", - "100_3chan.dat", - "100_3chan.hea", - "a103l.hea", - "a103l.mat", - "flacformats.d0", - "flacformats.d1", - "flacformats.d2", - "flacformats.hea", - "mixedsignals.hea", - "mixedsignals_e.dat", - "mixedsignals_p.dat", - "mixedsignals_r.dat", - "s0010_re.dat", - "s0010_re.hea", - "s0010_re.xyz", - "test01_00s.dat", - "test01_00s.hea", - "test01_00s_skewframe.hea", - "n8_evoked_raw_95_F1_R9.dat", - "n8_evoked_raw_95_F1_R9.hea", - ] + def setUpClass(cls): + cls.temp_directory = tempfile.TemporaryDirectory() + cls.temp_path = cls.temp_directory.name - for file in writefiles: - if os.path.isfile(file): - os.remove(file) + @classmethod + def tearDownClass(cls): + cls.temp_directory.cleanup() class TestMultiRecord(unittest.TestCase): @@ -1062,11 +1066,18 @@ def test_physical_conversion(self): adc_gain=adc_gain, baseline=baseline, fmt=["16", "16", "16"], + write_dir=self.temp_path, + ) + record = wfdb.rdrecord( + os.path.join(self.temp_path, "test_physical_conversion"), + physical=False, ) - record = wfdb.rdrecord("test_physical_conversion", physical=False) np.testing.assert_array_equal(record.d_signal, d_signal) - record = wfdb.rdrecord("test_physical_conversion", physical=True) + record = wfdb.rdrecord( + os.path.join(self.temp_path, "test_physical_conversion"), + physical=True, + ) for ch, gain in enumerate(adc_gain): np.testing.assert_allclose( record.p_signal[:, ch], @@ -1075,32 +1086,34 @@ def test_physical_conversion(self): atol=(0.05 / gain), ) + @classmethod + def setUpClass(cls): + cls.temp_directory = tempfile.TemporaryDirectory() + cls.temp_path = cls.temp_directory.name + @classmethod def tearDownClass(cls): - writefiles = [ - "test_physical_conversion.dat", - "test_physical_conversion.hea", - ] - for file in writefiles: - if os.path.isfile(file): - os.remove(file) + cls.temp_directory.cleanup() class TestDownload(unittest.TestCase): # Test that we can download records with no "dat" file # Regression test for https://github.com/MIT-LCP/wfdb-python/issues/118 def test_dl_database_no_dat_file(self): - wfdb.dl_database("afdb", "./download-tests/", ["00735"]) + wfdb.dl_database("afdb", self.temp_path, ["00735"]) # Test that we can download records that *do* have a "dat" file. def test_dl_database_with_dat_file(self): - wfdb.dl_database("afdb", "./download-tests/", ["04015"]) + wfdb.dl_database("afdb", self.temp_path, ["04015"]) + + @classmethod + def setUpClass(cls): + cls.temp_directory = tempfile.TemporaryDirectory() + cls.temp_path = cls.temp_directory.name - # Cleanup written files @classmethod - def tearDownClass(self): - if os.path.isdir("./download-tests/"): - shutil.rmtree("./download-tests/") + def tearDownClass(cls): + cls.temp_directory.cleanup() if __name__ == "__main__":