Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

Store test-suite temporary files in temporary directories #422

Merged
merged 1 commit into from
Sep 20, 2022
Merged

Store test-suite temporary files in temporary directories #422

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 23 additions & 19 deletions tests/test_annotation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import os
import re
import tempfile
import unittest

import numpy as np
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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()

Expand All @@ -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):
Expand All @@ -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__":
Expand Down
149 changes: 81 additions & 68 deletions tests/test_record.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import datetime
import os
import shutil
import tempfile
import unittest

import numpy as np
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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):
Expand All @@ -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):
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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],
Expand All @@ -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__":
Expand Down