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).

Author: Benjamin Moody

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__":

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__":