From bcd4adddb8285c42ea09a5dea3a0c5c7db16dba5 Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Wed, 28 Oct 2026 20:42:18 +0100 Subject: [PATCH 1/5] Convert tests from unittest to pytest Remove the unittest test framework from the test suite: - Test classes are plain pytest classes instead of unittest.TestCase subclasses. - self.assert*() calls are plain asserts (pytest.raises for assertRaises). - The temporary directory that used to be created in setUpClass() is a class-scoped fixture (temp_path) in conftest.py, and the CSV conversion test uses tmp_path instead of a "test_output" directory in the working directory. - assertAxesMatchSignal() in test_plot.py is a plain function. - The unittest.main() footers are removed. The tests are unchanged otherwise: the same 82 tests are collected. (unittest.mock is still used in test_processing.py.) --- tests/conftest.py | 7 ++ tests/io/test_convert.py | 68 ++++-------- tests/test_annotation.py | 53 ++++----- tests/test_plot.py | 131 +++++++++------------- tests/test_record.py | 231 +++++++++++++++++---------------------- tests/test_url.py | 61 +++++------ 6 files changed, 230 insertions(+), 321 deletions(-) create mode 100644 tests/conftest.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..da9e616a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,7 @@ +import pytest + + +@pytest.fixture(scope="class") +def temp_path(tmp_path_factory): + """Temporary directory (as a string) that is shared by all tests in a class.""" + return str(tmp_path_factory.mktemp("wfdb")) diff --git a/tests/io/test_convert.py b/tests/io/test_convert.py index 1453a5e2..6c5a042c 100644 --- a/tests/io/test_convert.py +++ b/tests/io/test_convert.py @@ -1,6 +1,4 @@ import os -import shutil -import unittest import numpy as np @@ -117,61 +115,43 @@ def test_edf_non_uniform(self): assert np.array_equal(test_results, target_results) -class TestCsvToWfdb(unittest.TestCase): +class TestCsvToWfdb: """ Tests for the io.convert.csv module. """ - def setUp(self): - """ - Create a temporary directory containing data for testing. - - Load 100.dat file for comparison to 100.csv file. - """ - self.test_dir = "test_output" - os.makedirs(self.test_dir, exist_ok=True) - - self.record_100_csv = "sample-data/100.csv" - self.record_100_dat = rdrecord("sample-data/100", physical=True) - - def tearDown(self): - """ - Remove the temporary directory after the test. - """ - if os.path.exists(self.test_dir): - shutil.rmtree(self.test_dir) - - def test_write_dir(self): + def test_write_dir(self, tmp_path): """ Call the function with the write_dir argument. """ + test_dir = str(tmp_path) + record_100_csv = "sample-data/100.csv" + # Load 100.dat file for comparison to 100.csv file + record_100_dat = rdrecord("sample-data/100", physical=True) + csv_to_wfdb( - file_name=self.record_100_csv, + file_name=record_100_csv, fs=360, units="mV", - write_dir=self.test_dir, + write_dir=test_dir, ) # Check if the output files are created in the specified directory - base_name = os.path.splitext(os.path.basename(self.record_100_csv))[0] - expected_dat_file = os.path.join(self.test_dir, f"{base_name}.dat") - expected_hea_file = os.path.join(self.test_dir, f"{base_name}.hea") + base_name = os.path.splitext(os.path.basename(record_100_csv))[0] + expected_dat_file = os.path.join(test_dir, f"{base_name}.dat") + expected_hea_file = os.path.join(test_dir, f"{base_name}.hea") - self.assertTrue(os.path.exists(expected_dat_file)) - self.assertTrue(os.path.exists(expected_hea_file)) + assert os.path.exists(expected_dat_file) + assert os.path.exists(expected_hea_file) # Check that newly written file matches the 100.dat file - record_write = rdrecord(os.path.join(self.test_dir, base_name)) - - self.assertEqual(record_write.fs, 360) - self.assertEqual(record_write.fs, self.record_100_dat.fs) - self.assertEqual(record_write.units, ["mV", "mV"]) - self.assertEqual(record_write.units, self.record_100_dat.units) - self.assertEqual(record_write.sig_name, ["MLII", "V5"]) - self.assertEqual(record_write.sig_name, self.record_100_dat.sig_name) - self.assertEqual(record_write.p_signal.size, 1300000) - self.assertEqual(record_write.p_signal.size, self.record_100_dat.p_signal.size) - - -if __name__ == "__main__": - unittest.main() + record_write = rdrecord(os.path.join(test_dir, base_name)) + + assert record_write.fs == 360 + assert record_write.fs == record_100_dat.fs + assert record_write.units == ["mV", "mV"] + assert record_write.units == record_100_dat.units + assert record_write.sig_name == ["MLII", "V5"] + assert record_write.sig_name == record_100_dat.sig_name + assert record_write.p_signal.size == 1300000 + assert record_write.p_signal.size == record_100_dat.p_signal.size diff --git a/tests/test_annotation.py b/tests/test_annotation.py index e70045f8..72f381b6 100644 --- a/tests/test_annotation.py +++ b/tests/test_annotation.py @@ -1,7 +1,5 @@ import os import re -import tempfile -import unittest import numpy as np import pandas as pd @@ -9,7 +7,7 @@ import wfdb -class TestAnnotation(unittest.TestCase): +class TestAnnotation: """ Testing read and write of WFDB annotations, including Physionet streaming. @@ -18,7 +16,7 @@ class TestAnnotation(unittest.TestCase): version 10.5.24 """ - def test_1(self): + def test_1(self, temp_path): """ Target file created with: rdann -r sample-data/100 -a atr > ann-1 @@ -87,9 +85,9 @@ def test_1(self): pn_annotation.create_label_map() # Test file writing - annotation.wrann(write_fs=True, write_dir=self.temp_path) + annotation.wrann(write_fs=True, write_dir=temp_path) write_annotation = wfdb.rdann( - os.path.join(self.temp_path, "100"), + os.path.join(temp_path, "100"), "atr", return_label_elements=["label_store", "symbol"], ) @@ -99,7 +97,7 @@ def test_1(self): assert annotation.__eq__(pn_annotation) assert annotation.__eq__(write_annotation) - def test_2(self): + def test_2(self, temp_path): """ Annotation file with many aux_note strings. @@ -161,9 +159,9 @@ def test_2(self): pn_annotation.create_label_map() # Test file writing - annotation.wrann(write_fs=True, write_dir=self.temp_path) + annotation.wrann(write_fs=True, write_dir=temp_path) write_annotation = wfdb.rdann( - os.path.join(self.temp_path, "12726"), + os.path.join(temp_path, "12726"), "anI", return_label_elements=["label_store", "symbol"], ) @@ -173,7 +171,7 @@ def test_2(self): assert annotation.__eq__(pn_annotation) assert annotation.__eq__(write_annotation) - def test_3(self): + def test_3(self, temp_path): """ Annotation file with custom annotation types @@ -236,9 +234,9 @@ def test_3(self): pn_annotation.create_label_map() # Test file writing - annotation.wrann(write_fs=True, write_dir=self.temp_path) + annotation.wrann(write_fs=True, write_dir=temp_path) write_annotation = wfdb.rdann( - os.path.join(self.temp_path, "1003"), + os.path.join(temp_path, "1003"), "atr", return_label_elements=["label_store", "symbol"], ) @@ -248,7 +246,7 @@ def test_3(self): assert annotation.__eq__(pn_annotation) assert annotation.__eq__(write_annotation) - def test_4(self): + def test_4(self, temp_path): """ Read and write annotations with large time skips @@ -256,14 +254,14 @@ def test_4(self): echo "xxxxxxxxx 10000000000 N 0 0 0" | wrann -r huge -a qrs """ annotation = wfdb.rdann("sample-data/huge", "qrs") - self.assertEqual(annotation.sample[0], 10000000000) - annotation.wrann(write_dir=self.temp_path) + assert annotation.sample[0] == 10000000000 + annotation.wrann(write_dir=temp_path) annotation1 = wfdb.rdann("sample-data/huge", "qrs") - annotation2 = wfdb.rdann(os.path.join(self.temp_path, "huge"), "qrs") - self.assertEqual(annotation1, annotation2) + annotation2 = wfdb.rdann(os.path.join(temp_path, "huge"), "qrs") + assert annotation1 == annotation2 - def test_5(self): + def test_5(self, temp_path): """ Write and read annotations with custom labels. """ @@ -284,20 +282,7 @@ def test_5(self): chan=ann_chan, custom_labels=ann_custom_labels, label_store=ann_label_store, - write_dir=self.temp_path, + write_dir=temp_path, ) - 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): - cls.temp_directory.cleanup() - - -if __name__ == "__main__": - unittest.main() + ann = wfdb.rdann(os.path.join(temp_path, "CustomLabel"), "atr") + assert ann.symbol == ["z", "l", "v", "r"] diff --git a/tests/test_plot.py b/tests/test_plot.py index c11ce518..97772b8d 100644 --- a/tests/test_plot.py +++ b/tests/test_plot.py @@ -1,5 +1,3 @@ -import unittest - import matplotlib.pyplot as plt import numpy as np @@ -7,70 +5,49 @@ from wfdb.plot import plot -class TestPlotWfdb(unittest.TestCase): +def assert_axes_match_signal(axes, signal, t_divisor=1): + """ + Check that axis limits are reasonable for plotting a signal array. + + Parameters + ---------- + axes : matplotlib.axes.Axes + An Axes object. + signal : numpy.ndarray + A one-dimensional array of sample values. + t_divisor : float, optional + The intended plotting resolution (number of samples of `signal` + per unit of the X axis.) + + """ + xmin, xmax = axes.get_xlim() + tmin = 0 + tmax = (len(signal) - 1) / t_divisor + # The range from tmin to tmax should fit within the plot. + assert xmin <= tmin, f"X range is [{xmin}, {xmax}]; expected [{tmin}, {tmax}]" + assert xmax >= tmax, f"X range is [{xmin}, {xmax}]; expected [{tmin}, {tmax}]" + # The padding on left and right sides should be approximately equal. + assert abs(xmin - tmin - (tmax - xmax)) <= (tmax - tmin) / 10 + 1 / t_divisor, ( + f"X range is [{xmin}, {xmax}]; expected [{tmin}, {tmax}]" + ) + + ymin, ymax = axes.get_ylim() + vmin = np.nanmin(signal) + vmax = np.nanmax(signal) + # The range from vmin to vmax should fit within the plot. + assert ymin <= vmin, f"Y range is [{ymin}, {ymax}]; expected [{vmin}, {vmax}]" + assert ymax >= vmax, f"Y range is [{ymin}, {ymax}]; expected [{vmin}, {vmax}]" + # The padding on top and bottom should be approximately equal. + assert abs(ymin - vmin - (vmax - ymax)) <= (vmax - vmin) / 10, ( + f"Y range is [{ymin}, {ymax}]; expected [{vmin}, {vmax}]" + ) + + +class TestPlotWfdb: """ Tests for the wfdb.plot_wfdb function """ - def assertAxesMatchSignal(self, axes, signal, t_divisor=1): - """ - Check that axis limits are reasonable for plotting a signal array. - - Parameters - ---------- - axes : matplotlib.axes.Axes - An Axes object. - signal : numpy.ndarray - A one-dimensional array of sample values. - t_divisor : float, optional - The intended plotting resolution (number of samples of `signal` - per unit of the X axis.) - - """ - xmin, xmax = axes.get_xlim() - tmin = 0 - tmax = (len(signal) - 1) / t_divisor - # The range from tmin to tmax should fit within the plot. - self.assertLessEqual( - xmin, - tmin, - msg=f"X range is [{xmin}, {xmax}]; expected [{tmin}, {tmax}]", - ) - self.assertGreaterEqual( - xmax, - tmax, - msg=f"X range is [{xmin}, {xmax}]; expected [{tmin}, {tmax}]", - ) - # The padding on left and right sides should be approximately equal. - self.assertAlmostEqual( - xmin - tmin, - tmax - xmax, - delta=(tmax - tmin) / 10 + 1 / t_divisor, - msg=f"X range is [{xmin}, {xmax}]; expected [{tmin}, {tmax}]", - ) - - ymin, ymax = axes.get_ylim() - vmin = np.nanmin(signal) - vmax = np.nanmax(signal) - # The range from vmin to vmax should fit within the plot. - self.assertLessEqual( - ymin, - vmin, - msg=f"Y range is [{ymin}, {ymax}]; expected [{vmin}, {vmax}]", - ) - self.assertGreaterEqual( - ymax, - vmax, - msg=f"Y range is [{ymin}, {ymax}]; expected [{vmin}, {vmax}]", - ) - # The padding on top and bottom should be approximately equal. - self.assertAlmostEqual( - ymin - vmin, - vmax - ymax, - delta=(vmax - vmin) / 10, - msg=f"Y range is [{ymin}, {ymax}]; expected [{vmin}, {vmax}]", - ) - def test_physical_smooth(self): """ Plot a record with physical, single-frequency data @@ -81,7 +58,7 @@ def test_physical_smooth(self): physical=True, smooth_frames=True, ) - self.assertIsNotNone(record.p_signal) + assert record.p_signal is not None annotation = wfdb.rdann("sample-data/100", "atr", sampto=1000) @@ -94,9 +71,9 @@ def test_physical_smooth(self): ) plt.close(fig) - self.assertEqual(len(fig.axes), record.n_sig) + assert len(fig.axes) == record.n_sig for ch in range(record.n_sig): - self.assertAxesMatchSignal(fig.axes[ch], record.p_signal[:, ch]) + assert_axes_match_signal(fig.axes[ch], record.p_signal[:, ch]) def test_digital_smooth(self): """ @@ -108,14 +85,14 @@ def test_digital_smooth(self): physical=False, smooth_frames=True, ) - self.assertIsNotNone(record.d_signal) + assert record.d_signal is not None fig = wfdb.plot_wfdb(record, time_units="seconds", return_fig=True) plt.close(fig) - self.assertEqual(len(fig.axes), record.n_sig) + assert len(fig.axes) == record.n_sig for ch in range(record.n_sig): - self.assertAxesMatchSignal(fig.axes[ch], record.d_signal[:, ch], record.fs) + assert_axes_match_signal(fig.axes[ch], record.d_signal[:, ch], record.fs) def test_physical_multifrequency(self): """ @@ -127,14 +104,14 @@ def test_physical_multifrequency(self): physical=True, smooth_frames=False, ) - self.assertIsNotNone(record.e_p_signal) + assert record.e_p_signal is not None fig = wfdb.plot_wfdb(record, time_units="seconds", return_fig=True) plt.close(fig) - self.assertEqual(len(fig.axes), record.n_sig) + assert len(fig.axes) == record.n_sig for ch in range(record.n_sig): - self.assertAxesMatchSignal( + assert_axes_match_signal( fig.axes[ch], record.e_p_signal[ch], record.fs * record.samps_per_frame[ch], @@ -150,21 +127,21 @@ def test_digital_multifrequency(self): physical=False, smooth_frames=False, ) - self.assertIsNotNone(record.e_d_signal) + assert record.e_d_signal is not None fig = wfdb.plot_wfdb(record, time_units="seconds", return_fig=True) plt.close(fig) - self.assertEqual(len(fig.axes), record.n_sig) + assert len(fig.axes) == record.n_sig for ch in range(record.n_sig): - self.assertAxesMatchSignal( + assert_axes_match_signal( fig.axes[ch], record.e_d_signal[ch], record.fs * record.samps_per_frame[ch], ) -class TestPlotInternal(unittest.TestCase): +class TestPlotInternal: """ Unit tests for internal wfdb.plot.plot functions """ @@ -202,7 +179,3 @@ def test_create_figure_multiple_subplots(self): assert fig is not None assert axes is not None assert len(axes) == n_subplots - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_record.py b/tests/test_record.py index cd2704db..cc95e224 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -1,15 +1,14 @@ import datetime import os -import tempfile -import unittest import numpy as np import pandas as pd +import pytest import wfdb -class TestRecord(unittest.TestCase): +class TestRecord: """ Test read and write of single segment WFDB records, including PhysioNet streaming. @@ -40,7 +39,7 @@ class TestRecord(unittest.TestCase): # ----------------------- 1. Basic Tests -----------------------# - def test_1a(self): + def test_1a(self, temp_path): """ Format 16, entire signal, digital. @@ -61,9 +60,9 @@ 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(write_dir=self.temp_path) + record_2.wrsamp(write_dir=temp_path) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "test01_00s"), + os.path.join(temp_path, "test01_00s"), physical=False, return_res=16, ) @@ -107,7 +106,7 @@ def test_1b(self): assert np.array_equal(sig, sig_pn) and fields == fields_pn assert np.array_equal(sig, sig_named) and fields == fields_named - def test_1c(self): + def test_1c(self, temp_path): """ Format 16, byte offset, selected duration, selected channels, digital, expanded format. @@ -140,9 +139,9 @@ def test_1c(self): ) # Test file writing - record.wrsamp(write_dir=self.temp_path, expanded=True) + record.wrsamp(write_dir=temp_path, expanded=True) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "a103l"), + os.path.join(temp_path, "a103l"), physical=False, smooth_frames=False, ) @@ -177,7 +176,7 @@ def test_1d(self): assert np.array_equal(sig_round, sig_target) assert np.array_equal(sig, sig_pn) and fields == fields_pn - def test_1e(self): + def test_1e(self, temp_path): """ Format 24, entire signal, digital. @@ -197,9 +196,9 @@ def test_1e(self): # Test file writing record_2 = wfdb.rdrecord("sample-data/n8_evoked_raw_95_F1_R9", physical=False) - record_2.wrsamp(write_dir=self.temp_path) + record_2.wrsamp(write_dir=temp_path) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "n8_evoked_raw_95_F1_R9"), + os.path.join(temp_path, "n8_evoked_raw_95_F1_R9"), physical=False, ) @@ -207,7 +206,7 @@ def test_1e(self): assert record.__eq__(record_pn) assert record_2.__eq__(record_write) - def test_1f(self): + def test_1f(self, temp_path): """ All binary formats, multiple signal files in one record. @@ -253,14 +252,14 @@ def test_1f(self): physical=False, channels=channels, ) - partial_record.wrsamp(write_dir=self.temp_path) + partial_record.wrsamp(write_dir=temp_path) converted_record = wfdb.rdrecord( - os.path.join(self.temp_path, "binformats"), + os.path.join(temp_path, "binformats"), physical=False, ) assert partial_record == converted_record - def test_read_write_flac(self): + def test_read_write_flac(self, temp_path): """ All FLAC formats, multiple signal files in one record. @@ -292,14 +291,14 @@ def test_read_write_flac(self): ) # Test file writing - record.wrsamp(write_dir=self.temp_path) + record.wrsamp(write_dir=temp_path) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "flacformats"), + os.path.join(temp_path, "flacformats"), physical=False, ) assert record == record_write - def test_read_write_flac_multifrequency(self): + def test_read_write_flac_multifrequency(self, temp_path): """ Format 516 with multiple signal files and variable samples per frame. """ @@ -309,17 +308,17 @@ def test_read_write_flac_multifrequency(self): physical=False, smooth_frames=False, ) - record.wrsamp(write_dir=self.temp_path, expanded=True) + record.wrsamp(write_dir=temp_path, expanded=True) # Check that result matches the original record = wfdb.rdrecord("sample-data/mixedsignals", smooth_frames=False) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "mixedsignals"), + os.path.join(temp_path, "mixedsignals"), smooth_frames=False, ) assert record == record_write - def test_unique_samps_per_frame_e_p_signal(self): + def test_unique_samps_per_frame_e_p_signal(self, temp_path): """ Test writing an e_p_signal with wfdb.io.wrsamp where the signals have different samples per frame. All other parameters which overlap between a Record object and wfdb.io.wrsamp are also checked. @@ -347,14 +346,14 @@ def test_unique_samps_per_frame_e_p_signal(self): baseline=record.baseline, adc_gain=record.adc_gain, fmt=record.fmt, - write_dir=self.temp_path, + write_dir=temp_path, ) # Check that the written record matches the original # Read in the original and written records record = wfdb.rdrecord("sample-data/mixedsignals", smooth_frames=False) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "mixedsignals"), + os.path.join(temp_path, "mixedsignals"), smooth_frames=False, ) @@ -397,7 +396,7 @@ def test_unique_samps_per_frame_e_p_signal(self): # Check that the arguments beyond the signals also match assert record_filtered == record_write_filtered - def test_unique_samps_per_frame_e_d_signal(self): + def test_unique_samps_per_frame_e_d_signal(self, temp_path): """ Test writing an e_d_signal with wfdb.io.wrsamp where the signals have different samples per frame. All other parameters which overlap between a Record object and wfdb.io.wrsamp are also checked. @@ -426,7 +425,7 @@ def test_unique_samps_per_frame_e_d_signal(self): baseline=record.baseline, adc_gain=record.adc_gain, fmt=record.fmt, - write_dir=self.temp_path, + write_dir=temp_path, ) # Check that the written record matches the original @@ -435,7 +434,7 @@ def test_unique_samps_per_frame_e_d_signal(self): "sample-data/mixedsignals", physical=False, smooth_frames=False ) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "mixedsignals"), + os.path.join(temp_path, "mixedsignals"), physical=False, smooth_frames=False, ) @@ -479,7 +478,7 @@ def test_unique_samps_per_frame_e_d_signal(self): # Check that the arguments beyond the signals also match assert record_filtered == record_write_filtered - def test_read_write_flac_many_channels(self): + def test_read_write_flac_many_channels(self, temp_path): """ Check we can read and write to format 516 with more than 8 channels. """ @@ -496,12 +495,12 @@ def test_read_write_flac_many_channels(self): d_signal=record.d_signal, adc_gain=record.adc_gain, baseline=record.baseline, - write_dir=self.temp_path, + write_dir=temp_path, ) # Check that signal matches the original record_fmt516 = wfdb.rdrecord( - os.path.join(self.temp_path, "s0010_re_fmt516"), + os.path.join(temp_path, "s0010_re_fmt516"), physical=False, ) assert (record.d_signal == record_fmt516.d_signal).all() @@ -550,7 +549,7 @@ def test_2a(self): assert np.array_equal(sig_round, sig_target) assert np.array_equal(sig, sig_pn) and fields == fields_pn - def test_2b(self): + def test_2b(self, temp_path): """ Format 212, selected duration, selected channel, digital. @@ -592,9 +591,9 @@ def test_2b(self): del record_named.comments[0] # Test file writing - record.wrsamp(write_dir=self.temp_path) + record.wrsamp(write_dir=temp_path) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "100"), + os.path.join(temp_path, "100"), physical=False, ) @@ -603,7 +602,7 @@ def test_2b(self): assert record.__eq__(record_named) assert record.__eq__(record_write) - def test_2c(self): + def test_2c(self, temp_path): """ Format 212, entire signal, physical, odd sampled record. @@ -616,8 +615,8 @@ def test_2c(self): # Test file writing record.d_signal = record.adc() - record.wrsamp(write_dir=self.temp_path) - record_write = wfdb.rdrecord(os.path.join(self.temp_path, "100_3chan")) + record.wrsamp(write_dir=temp_path) + record_write = wfdb.rdrecord(os.path.join(temp_path, "100_3chan")) record.d_signal = None assert np.array_equal(sig_round, sig_target) @@ -651,7 +650,7 @@ def test_2e(self): # --------------------- 3. Multi-dat records --------------------- # - def test_3a(self): + def test_3a(self, temp_path): """ Multi-dat, entire signal, digital Target file created with: @@ -665,9 +664,9 @@ def test_3a(self): record_pn = wfdb.rdrecord("s0010_re", physical=False, pn_dir="ptbdb/patient001") # Test file writing - record.wrsamp(write_dir=self.temp_path) + record.wrsamp(write_dir=temp_path) record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "s0010_re"), + os.path.join(temp_path, "s0010_re"), physical=False, ) @@ -705,7 +704,7 @@ def test_3b(self): # -------------- 4. Skew and multiple samples/frame -------------- # - def test_4a(self): + def test_4a(self, temp_path): """ Format 16, multi-samples per frame, skew, digital. @@ -729,17 +728,17 @@ def test_4a(self): smooth_frames=False, ignore_skew=True, ) - record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True) + record_no_skew.wrsamp(write_dir=temp_path, expanded=True) # Read the written record record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "test01_00s_skewframe"), + os.path.join(temp_path, "test01_00s_skewframe"), physical=False, ) assert np.array_equal(sig, sig_target) assert record.__eq__(record_write) - def test_4b(self): + def test_4b(self, temp_path): """ Format 12, multi-samples per frame, skew, entire signal, digital. @@ -766,10 +765,10 @@ def test_4b(self): smooth_frames=False, ignore_skew=True, ) - record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True) + record_no_skew.wrsamp(write_dir=temp_path, expanded=True) # Read the written record record_write = wfdb.rdrecord( - os.path.join(self.temp_path, "03700181"), + os.path.join(temp_path, "03700181"), physical=False, ) @@ -777,7 +776,7 @@ def test_4b(self): assert record.__eq__(record_pn) assert record.__eq__(record_write) - def test_4c(self): + def test_4c(self, temp_path): """ Format 12, multi-samples per frame, skew, selected suration, selected channels, physical. @@ -808,10 +807,10 @@ def test_4c(self): smooth_frames=False, ignore_skew=True, ) - record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True) + record_no_skew.wrsamp(write_dir=temp_path, expanded=True) # Read the written record writesig, writefields = wfdb.rdsamp( - os.path.join(self.temp_path, "03700181"), + os.path.join(temp_path, "03700181"), channels=[0, 2], sampfrom=1000, sampto=16000, @@ -841,7 +840,7 @@ def test_4d(self): assert np.array_equal(sig_round, sig_target) - def test_write_smoothed(self): + def test_write_smoothed(self, temp_path): """ Test writing a record after reading with smooth_frames """ @@ -850,9 +849,9 @@ def test_write_smoothed(self): physical=False, smooth_frames=True, ) - record.wrsamp(write_dir=self.temp_path) + record.wrsamp(write_dir=temp_path) record2 = wfdb.rdrecord( - os.path.join(self.temp_path, "drive02"), + os.path.join(temp_path, "drive02"), physical=False, ) np.testing.assert_array_equal(record.d_signal, record2.d_signal) @@ -861,12 +860,11 @@ def test_to_dataframe(self): record = wfdb.rdrecord("sample-data/test01_00s") df = record.to_dataframe() - self.assertEqual(record.sig_name, list(df.columns)) - self.assertEqual(len(df), record.sig_len) - self.assertEqual(df.index[0], pd.Timedelta(0)) - self.assertEqual( - df.index[-1], - pd.Timedelta(seconds=1 / record.fs * (record.sig_len - 1)), + assert record.sig_name == list(df.columns) + assert len(df) == record.sig_len + assert df.index[0] == pd.Timedelta(0) + assert df.index[-1] == pd.Timedelta( + seconds=1 / record.fs * (record.sig_len - 1) ) assert np.array_equal(record.p_signal, df.values) @@ -891,17 +889,8 @@ def test_header_with_non_utf8(self): ] assert record.units.__eq__(sig_units_target) - @classmethod - def setUpClass(cls): - cls.temp_directory = tempfile.TemporaryDirectory() - cls.temp_path = cls.temp_directory.name - @classmethod - def tearDownClass(cls): - cls.temp_directory.cleanup() - - -class TestMultiRecord(unittest.TestCase): +class TestMultiRecord: """ Test read and write of multi segment WFDB records, including PhysioNet streaming. @@ -999,7 +988,7 @@ def test_multi_fixed_d(self): # Sample values should match the output of rdsamp -H np.testing.assert_array_equal(sig, sig_target) # channel_names=[...] should give the same result as channels=[...] - self.assertEqual(record, record_named) + assert record == record_named def test_multi_variable_a(self): """ @@ -1133,7 +1122,7 @@ def test_multi_variable_d(self): assert record.__eq__(record_named) -class TestTimeConversion(unittest.TestCase): +class TestTimeConversion: """ Test cases for time conversion """ @@ -1153,18 +1142,21 @@ def test_single(self): # these time values should be equivalent n = 123 * header.fs t = datetime.timedelta(seconds=123) - self.assertEqual(header.get_frame_number(n), n) - self.assertEqual(header.get_frame_number(t), n) - self.assertEqual(header.get_elapsed_time(n), t) - self.assertEqual(header.get_elapsed_time(t), t) + assert header.get_frame_number(n) == n + assert header.get_frame_number(t) == n + assert header.get_elapsed_time(n) == t + assert header.get_elapsed_time(t) == t # record test01_00s has no base date, so absolute time conversions # should fail - self.assertIsNone(header.base_date) + assert header.base_date is None d = datetime.datetime(2001, 1, 1, 12, 0, 0) - self.assertRaises(ValueError, header.get_frame_number, d) - self.assertRaises(ValueError, header.get_absolute_time, n) - self.assertRaises(ValueError, header.get_absolute_time, t) + with pytest.raises(ValueError): + header.get_frame_number(d) + with pytest.raises(ValueError): + header.get_absolute_time(n) + with pytest.raises(ValueError): + header.get_absolute_time(t) def test_multisegment_with_date(self): """ @@ -1184,18 +1176,18 @@ def test_multisegment_with_date(self): n = 123 * header.fs t = datetime.timedelta(seconds=123) d = t + header.base_datetime - self.assertEqual(header.get_frame_number(n), n) - self.assertEqual(header.get_frame_number(t), n) - self.assertEqual(header.get_frame_number(d), n) - self.assertEqual(header.get_elapsed_time(n), t) - self.assertEqual(header.get_elapsed_time(t), t) - self.assertEqual(header.get_elapsed_time(d), t) - self.assertEqual(header.get_absolute_time(n), d) - self.assertEqual(header.get_absolute_time(t), d) - self.assertEqual(header.get_absolute_time(d), d) - - -class TestSignal(unittest.TestCase): + assert header.get_frame_number(n) == n + assert header.get_frame_number(t) == n + assert header.get_frame_number(d) == n + assert header.get_elapsed_time(n) == t + assert header.get_elapsed_time(t) == t + assert header.get_elapsed_time(d) == t + assert header.get_absolute_time(n) == d + assert header.get_absolute_time(t) == d + assert header.get_absolute_time(d) == d + + +class TestSignal: """ For lower level signal tests @@ -1219,7 +1211,7 @@ def test_infer_sig_len(self): assert record_2.__eq__(record) - def test_physical_conversion(self): + def test_physical_conversion(self, temp_path): n_sig = 3 adc_gain = [1.0, 1234.567, 765.4321] baseline = [10, 20, -30] @@ -1260,12 +1252,12 @@ def test_physical_conversion(self): ) e_d_signal_converted = record.adc(expanded=True, inplace=False) - self.assertEqual(len(e_d_signal_converted), n_sig) + assert len(e_d_signal_converted) == n_sig for x, y in zip(e_d_signal_converted, e_d_signal): np.testing.assert_array_equal(x, y) record.adc(expanded=True, inplace=True) - self.assertEqual(len(record.e_d_signal), n_sig) + assert len(record.e_d_signal) == n_sig for x, y in zip(record.e_d_signal, e_d_signal): np.testing.assert_array_equal(x, y) @@ -1280,16 +1272,16 @@ def test_physical_conversion(self): adc_gain=adc_gain, baseline=baseline, fmt=fmt, - write_dir=self.temp_path, + write_dir=temp_path, ) record = wfdb.rdrecord( - os.path.join(self.temp_path, "test_physical_conversion"), + os.path.join(temp_path, "test_physical_conversion"), physical=False, ) np.testing.assert_array_equal(record.d_signal, d_signal) record = wfdb.rdrecord( - os.path.join(self.temp_path, "test_physical_conversion"), + os.path.join(temp_path, "test_physical_conversion"), physical=True, ) for ch, gain in enumerate(adc_gain): @@ -1300,7 +1292,7 @@ def test_physical_conversion(self): atol=(0.05 / gain), ) - def test_adc_gain_max_boundary(self): + def test_adc_gain_max_boundary(self, temp_path): """ Exercise the MAX_I32 clamp path: use a tiny range around a huge negative signal so the computed baseline would exceed MAX_I32. Verify we hit that branch and @@ -1326,12 +1318,12 @@ def test_adc_gain_max_boundary(self): units=["mV"], p_signal=p_signal, fmt=["16"], - write_dir=self.temp_path, + write_dir=temp_path, ) # Read it back record = wfdb.rdrecord( - os.path.join(self.temp_path, "test_negative_signal"), + os.path.join(temp_path, "test_negative_signal"), physical=True, ) @@ -1344,13 +1336,13 @@ def test_adc_gain_max_boundary(self): ) # Verify baseline was clamped to MAX_I32 - self.assertEqual(record.baseline[0], 2147483647) # MAX_I32 + assert record.baseline[0] == 2147483647 # MAX_I32 # Confirm the formula is correct expected_gain = (2147483647 - (-32768)) / abs(base_value) np.testing.assert_allclose(record.adc_gain[0], expected_gain, rtol=1e-3) - def test_adc_gain_min_boundary(self): + def test_adc_gain_min_boundary(self, temp_path): """ Exercise the MIN_I32 clamp path: use a tiny range around a huge positive signal so the computed baseline would drop below MIN_I32. Verify we hit that branch and @@ -1376,12 +1368,12 @@ def test_adc_gain_min_boundary(self): units=["mV"], p_signal=p_signal, fmt=["16"], - write_dir=self.temp_path, + write_dir=temp_path, ) # Read it back record = wfdb.rdrecord( - os.path.join(self.temp_path, "test_positive_signal"), + os.path.join(temp_path, "test_positive_signal"), physical=True, ) @@ -1394,42 +1386,19 @@ def test_adc_gain_min_boundary(self): ) # Verify baseline was clamped to MIN_I32 - self.assertEqual(record.baseline[0], -2147483648) + assert record.baseline[0] == -2147483648 # Confirm the formula is correct expected_gain = (32767 - (-2147483648)) / (base_value + 1.5) np.testing.assert_allclose(record.adc_gain[0], expected_gain, rtol=1e-3) - @classmethod - def setUpClass(cls): - cls.temp_directory = tempfile.TemporaryDirectory() - cls.temp_path = cls.temp_directory.name - - @classmethod - def tearDownClass(cls): - cls.temp_directory.cleanup() - -class TestDownload(unittest.TestCase): +class TestDownload: # 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", self.temp_path, ["00735"]) + def test_dl_database_no_dat_file(self, temp_path): + wfdb.dl_database("afdb", 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", self.temp_path, ["04015"]) - - @classmethod - def setUpClass(cls): - cls.temp_directory = tempfile.TemporaryDirectory() - cls.temp_path = cls.temp_directory.name - - @classmethod - def tearDownClass(cls): - cls.temp_directory.cleanup() - - -if __name__ == "__main__": - unittest.main() - print("Everything passed!") + def test_dl_database_with_dat_file(self, temp_path): + wfdb.dl_database("afdb", temp_path, ["04015"]) diff --git a/tests/test_url.py b/tests/test_url.py index e96b6fa4..52e33764 100644 --- a/tests/test_url.py +++ b/tests/test_url.py @@ -1,12 +1,11 @@ import gzip import http.server import threading -import unittest import wfdb.io._url -class TestNetFiles(unittest.TestCase): +class TestNetFiles: """ Test accessing remote files. """ @@ -78,10 +77,10 @@ def _test_text(self, url, content, buffering): """ # read(-1), readable(), seekable() with wfdb.io._url.openurl(url, "r", buffering=buffering) as tf: - self.assertTrue(tf.readable()) - self.assertTrue(tf.seekable()) - self.assertEqual(tf.read(), content) - self.assertEqual(tf.read(), "") + assert tf.readable() + assert tf.seekable() + assert tf.read() == content + assert tf.read() == "" # read(10) with wfdb.io._url.openurl(url, "r", buffering=buffering) as tf: @@ -91,7 +90,7 @@ def _test_text(self, url, content, buffering): result += chunk if len(chunk) < 10: break - self.assertEqual(result, content) + assert result == content # readline(), seek(), tell() with wfdb.io._url.openurl(url, "r", buffering=buffering) as tf: @@ -104,7 +103,7 @@ def _test_text(self, url, content, buffering): result += chunk if len(chunk) == 0: break - self.assertEqual(result, content) + assert result == content def _test_binary(self, url, content, buffering): """ @@ -126,11 +125,11 @@ def _test_binary(self, url, content, buffering): """ # read(-1), readable(), seekable() with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: - self.assertTrue(bf.readable()) - self.assertTrue(bf.seekable()) - self.assertEqual(bf.read(), content) - self.assertEqual(bf.read(), b"") - self.assertEqual(bf.tell(), len(content)) + assert bf.readable() + assert bf.seekable() + assert bf.read() == content + assert bf.read() == b"" + assert bf.tell() == len(content) # read(10) with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: @@ -140,8 +139,8 @@ def _test_binary(self, url, content, buffering): result += chunk if len(chunk) < 10: break - self.assertEqual(result, content) - self.assertEqual(bf.tell(), len(content)) + assert result == content + assert bf.tell() == len(content) # readline() with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: @@ -151,13 +150,13 @@ def _test_binary(self, url, content, buffering): result += chunk if len(chunk) == 0: break - self.assertEqual(result, content) - self.assertEqual(bf.tell(), len(content)) + assert result == content + assert bf.tell() == len(content) # read1(10), seek(), tell() with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: bf.seek(0, 2) - self.assertEqual(bf.tell(), len(content)) + assert bf.tell() == len(content) bf.seek(0) result = b"" while True: @@ -168,8 +167,8 @@ def _test_binary(self, url, content, buffering): result += chunk if len(chunk) == 0: break - self.assertEqual(result, content) - self.assertEqual(bf.tell(), len(content)) + assert result == content + assert bf.tell() == len(content) # readinto(bytearray(10)) with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: @@ -180,8 +179,8 @@ def _test_binary(self, url, content, buffering): result += chunk[:count] if count < 10: break - self.assertEqual(result, content) - self.assertEqual(bf.tell(), len(content)) + assert result == content + assert bf.tell() == len(content) # readinto1(bytearray(10)) with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: @@ -192,11 +191,11 @@ def _test_binary(self, url, content, buffering): result += chunk[:count] if count == 0: break - self.assertEqual(result, content) - self.assertEqual(bf.tell(), len(content)) + assert result == content + assert bf.tell() == len(content) -class TestRemoteFLACFiles(unittest.TestCase): +class TestRemoteFLACFiles: """ Test reading FLAC files over HTTP. """ @@ -233,8 +232,8 @@ def test_whole_file(self): # Read the file using standard file I/O sf1 = soundfile.SoundFile(data_file_path) - self.assertEqual(sf1.format, expected_format) - self.assertEqual(sf1.subtype, expected_subtype) + assert sf1.format == expected_format + assert sf1.subtype == expected_subtype data1 = sf1.read() # Read the file using HTTP @@ -244,8 +243,8 @@ def test_whole_file(self): url = server.url("/foo.dat") file2 = wfdb.io._url.openurl(url, "rb") sf2 = soundfile.SoundFile(file2) - self.assertEqual(sf2.format, expected_format) - self.assertEqual(sf2.subtype, expected_subtype) + assert sf2.format == expected_format + assert sf2.subtype == expected_subtype data2 = sf2.read() # Check that results are equal @@ -387,7 +386,3 @@ def send_head(self): self.send_header(h, v) self.end_headers() return content - - -if __name__ == "__main__": - unittest.main() From 032263d65f878a30e92cf8afe5f421a1358bfdb7 Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Wed, 28 Oct 2026 20:43:45 +0100 Subject: [PATCH 2/5] Run local and PhysioNet checks as separate tests Tests that stream or download data from PhysioNet are now marked with "network". If such a test fails with a connection error, a timeout or an interrupted transfer, it is reported as skipped instead of failing (the reason is shown in the summary), so a slow or unreachable PhysioNet no longer breaks CI. Other failures, including failed assertions, still fail. 15 tests compared local sample data with the same data streamed from PhysioNet. They now take a "source" fixture that is parametrized with "local" and "physionet" (the latter marked "network"), so test_X[local] only uses local files and test_X[physionet] additionally streams from PhysioNet. The sequence of operations within each test is unchanged. A PhysioNet outage therefore only skips the [physionet] variants, and the local reading and writing checks always run (80 tests work offline instead of 65). Use `pytest -m "not network"` to run only these tests. --- DEVELOPING.md | 6 ++ pyproject.toml | 6 ++ tests/conftest.py | 34 +++++++ tests/test_annotation.py | 62 ++++++------ tests/test_record.py | 198 ++++++++++++++++++++++----------------- 5 files changed, 193 insertions(+), 113 deletions(-) diff --git a/DEVELOPING.md b/DEVELOPING.md index 84979b0c..d22dd3a1 100644 --- a/DEVELOPING.md +++ b/DEVELOPING.md @@ -82,3 +82,9 @@ pytest # https://github.com/pytest-dev/pytest-xdist pytest -n auto ``` + +Tests that stream or download data from PhysioNet are marked with `network`. Most of them also exist as a local variant (for example, `test_1a[local]` and `test_1a[physionet]`), so a slow or unreachable PhysioNet only affects the `physionet` variants. If PhysioNet cannot be reached, these tests are skipped instead of failing. To run only the tests that work offline: + +```sh +pytest -m "not network" +``` diff --git a/pyproject.toml b/pyproject.toml index c57ff58e..eea2a041 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,5 +66,11 @@ ignore = [ # Re-exports "__init__.py" = ["F401"] +[tool.pytest.ini_options] +addopts = "-r s" +markers = [ + "network: requires access to PhysioNet (skipped if it cannot be reached)", +] + [tool.uv.build-backend] module-root = "" diff --git a/tests/conftest.py b/tests/conftest.py index da9e616a..404f4727 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,41 @@ import pytest +import requests + +# Errors that indicate that PhysioNet is unreachable or too slow (and not a test failure) +NETWORK_ERRORS = ( + requests.exceptions.ConnectionError, + requests.exceptions.ChunkedEncodingError, + requests.exceptions.Timeout, +) @pytest.fixture(scope="class") def temp_path(tmp_path_factory): """Temporary directory (as a string) that is shared by all tests in a class.""" return str(tmp_path_factory.mktemp("wfdb")) + + +@pytest.fixture(params=["local", pytest.param("physionet", marks=pytest.mark.network)]) +def source(request): + """Run a test with local files, and (as a "network" test) with PhysioNet.""" + return request.param + + +@pytest.hookimpl(wrapper=True) +def pytest_runtest_makereport(item, call): + """Report tests marked as "network" as skipped if the network fails.""" + report = yield + if ( + call.when == "call" + and report.failed + and call.excinfo is not None + and call.excinfo.errisinstance(NETWORK_ERRORS) + and item.get_closest_marker("network") + ): + report.outcome = "skipped" + report.longrepr = ( + str(item.path), + item.location[1] + 1, + f"Skipped: network error: {call.excinfo.value}", + ) + return report diff --git a/tests/test_annotation.py b/tests/test_annotation.py index 72f381b6..9ee80794 100644 --- a/tests/test_annotation.py +++ b/tests/test_annotation.py @@ -16,7 +16,7 @@ class TestAnnotation: version 10.5.24 """ - def test_1(self, temp_path): + def test_1(self, source, temp_path): """ Target file created with: rdann -r sample-data/100 -a atr > ann-1 @@ -75,14 +75,15 @@ def test_1(self, temp_path): ] # Test file streaming - pn_annotation = wfdb.rdann( - "100", - "atr", - pn_dir="mitdb", - return_label_elements=["label_store", "symbol"], - ) - pn_annotation.aux_note[0] = "(N" - pn_annotation.create_label_map() + if source == "physionet": + pn_annotation = wfdb.rdann( + "100", + "atr", + pn_dir="mitdb", + return_label_elements=["label_store", "symbol"], + ) + pn_annotation.aux_note[0] = "(N" + pn_annotation.create_label_map() # Test file writing annotation.wrann(write_fs=True, write_dir=temp_path) @@ -94,10 +95,11 @@ def test_1(self, temp_path): write_annotation.create_label_map() assert comp == [True] * 6 - assert annotation.__eq__(pn_annotation) + if source == "physionet": + assert annotation.__eq__(pn_annotation) assert annotation.__eq__(write_annotation) - def test_2(self, temp_path): + def test_2(self, source, temp_path): """ Annotation file with many aux_note strings. @@ -150,13 +152,14 @@ def test_2(self, temp_path): annotation.aux_note == target_aux_note, ] # Test file streaming - pn_annotation = wfdb.rdann( - "12726", - "anI", - pn_dir="prcp", - return_label_elements=["label_store", "symbol"], - ) - pn_annotation.create_label_map() + if source == "physionet": + pn_annotation = wfdb.rdann( + "12726", + "anI", + pn_dir="prcp", + return_label_elements=["label_store", "symbol"], + ) + pn_annotation.create_label_map() # Test file writing annotation.wrann(write_fs=True, write_dir=temp_path) @@ -168,10 +171,11 @@ def test_2(self, temp_path): write_annotation.create_label_map() assert comp == [True] * 6 - assert annotation.__eq__(pn_annotation) + if source == "physionet": + assert annotation.__eq__(pn_annotation) assert annotation.__eq__(write_annotation) - def test_3(self, temp_path): + def test_3(self, source, temp_path): """ Annotation file with custom annotation types @@ -225,13 +229,14 @@ def test_3(self, temp_path): ] # Test file streaming - pn_annotation = wfdb.rdann( - "1003", - "atr", - pn_dir="challenge-2014/set-p2", - return_label_elements=["label_store", "symbol"], - ) - pn_annotation.create_label_map() + if source == "physionet": + pn_annotation = wfdb.rdann( + "1003", + "atr", + pn_dir="challenge-2014/set-p2", + return_label_elements=["label_store", "symbol"], + ) + pn_annotation.create_label_map() # Test file writing annotation.wrann(write_fs=True, write_dir=temp_path) @@ -243,7 +248,8 @@ def test_3(self, temp_path): write_annotation.create_label_map() assert comp == [True] * 6 - assert annotation.__eq__(pn_annotation) + if source == "physionet": + assert annotation.__eq__(pn_annotation) assert annotation.__eq__(write_annotation) def test_4(self, temp_path): diff --git a/tests/test_record.py b/tests/test_record.py index cc95e224..23c8e110 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -39,7 +39,7 @@ class TestRecord: # ----------------------- 1. Basic Tests -----------------------# - def test_1a(self, temp_path): + def test_1a(self, source, temp_path): """ Format 16, entire signal, digital. @@ -51,9 +51,10 @@ def test_1a(self, temp_path): sig_target = np.genfromtxt("tests/target-output/record-1a") # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "test01_00s", pn_dir="macecgdb", physical=False, return_res=16 - ) + if source == "physionet": + record_pn = wfdb.rdrecord( + "test01_00s", pn_dir="macecgdb", physical=False, return_res=16 + ) # Test file writing record_2 = wfdb.rdrecord( @@ -68,10 +69,11 @@ def test_1a(self, temp_path): ) assert np.array_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record_2.__eq__(record_write) - def test_1b(self): + def test_1b(self, source): """ Format 16, byte offset, selected duration, selected channels, physical. @@ -86,13 +88,14 @@ def test_1b(self): sig_target = np.genfromtxt("tests/target-output/record-1b") # Compare data streaming from Physionet - sig_pn, fields_pn = wfdb.rdsamp( - "a103l", - pn_dir="challenge-2015/training", - sampfrom=12500, - sampto=40000, - channels=[2, 0], - ) + if source == "physionet": + sig_pn, fields_pn = wfdb.rdsamp( + "a103l", + pn_dir="challenge-2015/training", + sampfrom=12500, + sampto=40000, + channels=[2, 0], + ) # Option of selecting channels by name sig_named, fields_named = wfdb.rdsamp( @@ -103,10 +106,11 @@ def test_1b(self): ) assert np.array_equal(sig_round, sig_target) - assert np.array_equal(sig, sig_pn) and fields == fields_pn + if source == "physionet": + assert np.array_equal(sig, sig_pn) and fields == fields_pn assert np.array_equal(sig, sig_named) and fields == fields_named - def test_1c(self, temp_path): + def test_1c(self, source, temp_path): """ Format 16, byte offset, selected duration, selected channels, digital, expanded format. @@ -129,14 +133,15 @@ def test_1c(self, temp_path): sig_target = np.genfromtxt("tests/target-output/record-1c") # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "a103l", - pn_dir="challenge-2015/training", - sampfrom=20000, - channels=[0, 1], - physical=False, - smooth_frames=False, - ) + if source == "physionet": + record_pn = wfdb.rdrecord( + "a103l", + pn_dir="challenge-2015/training", + sampfrom=20000, + channels=[0, 1], + physical=False, + smooth_frames=False, + ) # Test file writing record.wrsamp(write_dir=temp_path, expanded=True) @@ -147,10 +152,11 @@ def test_1c(self, temp_path): ) assert np.array_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record.__eq__(record_write) - def test_1d(self): + def test_1d(self, source): """ Format 80, selected duration, selected channels, physical @@ -165,18 +171,20 @@ def test_1d(self): sig_target = sig_target.reshape(len(sig_target), 1) # Compare data streaming from Physionet - sig_pn, fields_pn = wfdb.rdsamp( - "3000003_0003", - pn_dir="mimic3wdb/30/3000003/", - sampfrom=125, - sampto=1000, - channels=[1], - ) + if source == "physionet": + sig_pn, fields_pn = wfdb.rdsamp( + "3000003_0003", + pn_dir="mimic3wdb/30/3000003/", + sampfrom=125, + sampto=1000, + channels=[1], + ) assert np.array_equal(sig_round, sig_target) - assert np.array_equal(sig, sig_pn) and fields == fields_pn + if source == "physionet": + assert np.array_equal(sig, sig_pn) and fields == fields_pn - def test_1e(self, temp_path): + def test_1e(self, source, temp_path): """ Format 24, entire signal, digital. @@ -190,9 +198,10 @@ def test_1e(self, temp_path): sig_target[sig_target == -32768] = -(2**23) # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "n8_evoked_raw_95_F1_R9", physical=False, pn_dir="earndb/raw/N8" - ) + if source == "physionet": + record_pn = wfdb.rdrecord( + "n8_evoked_raw_95_F1_R9", physical=False, pn_dir="earndb/raw/N8" + ) # Test file writing record_2 = wfdb.rdrecord("sample-data/n8_evoked_raw_95_F1_R9", physical=False) @@ -203,7 +212,8 @@ def test_1e(self, temp_path): ) assert np.array_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record_2.__eq__(record_write) def test_1f(self, temp_path): @@ -529,7 +539,7 @@ def test_read_flac_longduration(self): # ------------------ 2. Special format records ------------------ # - def test_2a(self): + def test_2a(self, source): """ Format 212, entire signal, physical. @@ -541,15 +551,17 @@ def test_2a(self): sig_target = np.genfromtxt("tests/target-output/record-2a") # Compare data streaming from Physionet - sig_pn, fields_pn = wfdb.rdsamp("100", pn_dir="mitdb") + if source == "physionet": + sig_pn, fields_pn = wfdb.rdsamp("100", pn_dir="mitdb") # This comment line was manually added and is not present in the # original PhysioNet record del fields["comments"][0] assert np.array_equal(sig_round, sig_target) - assert np.array_equal(sig, sig_pn) and fields == fields_pn + if source == "physionet": + assert np.array_equal(sig, sig_pn) and fields == fields_pn - def test_2b(self, temp_path): + def test_2b(self, source, temp_path): """ Format 212, selected duration, selected channel, digital. @@ -568,14 +580,15 @@ def test_2b(self, temp_path): sig_target = sig_target.reshape(len(sig_target), 1) # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "100", - sampfrom=1, - sampto=10800, - channels=[1], - physical=False, - pn_dir="mitdb", - ) + if source == "physionet": + record_pn = wfdb.rdrecord( + "100", + sampfrom=1, + sampto=10800, + channels=[1], + physical=False, + pn_dir="mitdb", + ) # This comment line was manually added and is not present in the # original PhysioNet record del record.comments[0] @@ -598,7 +611,8 @@ def test_2b(self, temp_path): ) assert np.array_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record.__eq__(record_named) assert record.__eq__(record_write) @@ -650,7 +664,7 @@ def test_2e(self): # --------------------- 3. Multi-dat records --------------------- # - def test_3a(self, temp_path): + def test_3a(self, source, temp_path): """ Multi-dat, entire signal, digital Target file created with: @@ -661,7 +675,10 @@ def test_3a(self, temp_path): sig_target = np.genfromtxt("tests/target-output/record-3a") # Compare data streaming from Physionet - record_pn = wfdb.rdrecord("s0010_re", physical=False, pn_dir="ptbdb/patient001") + if source == "physionet": + record_pn = wfdb.rdrecord( + "s0010_re", physical=False, pn_dir="ptbdb/patient001" + ) # Test file writing record.wrsamp(write_dir=temp_path) @@ -671,10 +688,11 @@ def test_3a(self, temp_path): ) assert np.array_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record.__eq__(record_write) - def test_3b(self): + def test_3b(self, source): """ Multi-dat, selected duration, selected channels, physical. @@ -691,16 +709,18 @@ def test_3b(self): sig_target = np.genfromtxt("tests/target-output/record-3b") # Compare data streaming from Physionet - sig_pn, fields_pn = wfdb.rdsamp( - "s0010_re", - sampfrom=5000, - pn_dir="ptbdb/patient001", - sampto=38000, - channels=[13, 0, 4, 8, 3], - ) + if source == "physionet": + sig_pn, fields_pn = wfdb.rdsamp( + "s0010_re", + sampfrom=5000, + pn_dir="ptbdb/patient001", + sampto=38000, + channels=[13, 0, 4, 8, 3], + ) assert np.array_equal(sig_round, sig_target) - assert np.array_equal(sig, sig_pn) and fields == fields_pn + if source == "physionet": + assert np.array_equal(sig, sig_pn) and fields == fields_pn # -------------- 4. Skew and multiple samples/frame -------------- # @@ -738,7 +758,7 @@ def test_4a(self, temp_path): assert np.array_equal(sig, sig_target) assert record.__eq__(record_write) - def test_4b(self, temp_path): + def test_4b(self, source, temp_path): """ Format 12, multi-samples per frame, skew, entire signal, digital. @@ -755,7 +775,8 @@ def test_4b(self, temp_path): sig_target = np.genfromtxt("tests/target-output/record-4b") # Compare data streaming from Physionet - record_pn = wfdb.rdrecord("03700181", physical=False, pn_dir="mimicdb/037") + if source == "physionet": + record_pn = wfdb.rdrecord("03700181", physical=False, pn_dir="mimicdb/037") # Test file writing. Multiple samples per frame and skew. # Have to read all the samples in the record, ignoring skew @@ -773,10 +794,11 @@ def test_4b(self, temp_path): ) assert np.array_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record.__eq__(record_write) - def test_4c(self, temp_path): + def test_4c(self, source, temp_path): """ Format 12, multi-samples per frame, skew, selected suration, selected channels, physical. @@ -791,13 +813,14 @@ def test_4c(self, temp_path): sig_target = np.genfromtxt("tests/target-output/record-4c") # Compare data streaming from Physionet - sig_pn, fields_pn = wfdb.rdsamp( - "03700181", - pn_dir="mimicdb/037", - channels=[0, 2], - sampfrom=1000, - sampto=16000, - ) + if source == "physionet": + sig_pn, fields_pn = wfdb.rdsamp( + "03700181", + pn_dir="mimicdb/037", + channels=[0, 2], + sampfrom=1000, + sampto=16000, + ) # Test file writing. Multiple samples per frame and skew. # Have to read all the samples in the record, ignoring skew @@ -817,7 +840,8 @@ def test_4c(self, temp_path): ) assert np.array_equal(sig_round, sig_target) - assert np.array_equal(sig, sig_pn) and fields == fields_pn + if source == "physionet": + assert np.array_equal(sig, sig_pn) and fields == fields_pn assert np.array_equal(sig, writesig) and fields == writefields def test_4d(self): @@ -1078,7 +1102,7 @@ def test_multi_variable_c(self): np.testing.assert_equal(sig_round, sig_target) - def test_multi_variable_d(self): + def test_multi_variable_d(self, source): """ Multi-segment, variable layout, selected duration, selected channels, digital. There are two channels: PLETH, and II. Their @@ -1098,14 +1122,15 @@ def test_multi_variable_d(self): sig = record.d_signal # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "p000878-2137-10-26-16-57", - pn_dir="mimic3wdb/matched/p00/p000878/", - sampfrom=3550, - sampto=7500, - channels=[0, 1], - physical=False, - ) + if source == "physionet": + record_pn = wfdb.rdrecord( + "p000878-2137-10-26-16-57", + pn_dir="mimic3wdb/matched/p00/p000878/", + sampfrom=3550, + sampto=7500, + channels=[0, 1], + physical=False, + ) sig_target = np.genfromtxt("tests/target-output/record-multi-variable-d") # Option of selecting channels by name @@ -1118,7 +1143,8 @@ def test_multi_variable_d(self): ) np.testing.assert_equal(sig, sig_target) - assert record.__eq__(record_pn) + if source == "physionet": + assert record.__eq__(record_pn) assert record.__eq__(record_named) @@ -1396,9 +1422,11 @@ def test_adc_gain_min_boundary(self, temp_path): class TestDownload: # Test that we can download records with no "dat" file # Regression test for https://github.com/MIT-LCP/wfdb-python/issues/118 + @pytest.mark.network def test_dl_database_no_dat_file(self, temp_path): wfdb.dl_database("afdb", temp_path, ["00735"]) # Test that we can download records that *do* have a "dat" file. + @pytest.mark.network def test_dl_database_with_dat_file(self, temp_path): wfdb.dl_database("afdb", temp_path, ["04015"]) From 2dffc93634a719db1682d706efc9f18fb23ce9e5 Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Wed, 28 Oct 2026 20:43:54 +0100 Subject: [PATCH 3/5] Format section comments in test_record.py consistently Use the same width (72 columns) and spacing ("# --- title --- #") for all section banners. --- tests/test_record.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_record.py b/tests/test_record.py index 23c8e110..7eaac7b7 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -37,7 +37,7 @@ class TestRecord: "base_datetime", ] - # ----------------------- 1. Basic Tests -----------------------# + # ------------------------ 1. Basic Tests ------------------------ # def test_1a(self, source, temp_path): """ @@ -537,7 +537,7 @@ def test_read_flac_longduration(self): ) np.testing.assert_array_equal(record.p_signal, sig_target) - # ------------------ 2. Special format records ------------------ # + # ------------------ 2. Special format records ------------------- # def test_2a(self, source): """ From 549b829124ff628a121deb64d273f2e333654808 Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Wed, 28 Oct 2026 21:04:55 +0100 Subject: [PATCH 4/5] Skip network tests on HTTP server errors, too PhysioNet regularly answers with "502 Bad Gateway", which is raised as a NetFileError and was not covered by the skip hook (only connection errors, timeouts and interrupted transfers were), so the affected [physionet] tests still failed in CI. Treat NetFileError with a 5xx status code or 429 (too many requests) like the other network errors. Client errors such as 404 or 403 still fail. --- tests/conftest.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 404f4727..65c3ccb3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,8 @@ import pytest import requests +from wfdb.io._url import NetFileError + # Errors that indicate that PhysioNet is unreachable or too slow (and not a test failure) NETWORK_ERRORS = ( requests.exceptions.ConnectionError, @@ -9,6 +11,19 @@ ) +def is_network_error(exc): + """Whether an exception means that PhysioNet is not available right now.""" + if isinstance(exc, NETWORK_ERRORS): + return True + # Server errors (for example 502 Bad Gateway) and rate limiting, but not errors + # like 404 or 403 which indicate a real problem + return ( + isinstance(exc, NetFileError) + and exc.status_code is not None + and (exc.status_code >= 500 or exc.status_code == 429) + ) + + @pytest.fixture(scope="class") def temp_path(tmp_path_factory): """Temporary directory (as a string) that is shared by all tests in a class.""" @@ -29,7 +44,7 @@ def pytest_runtest_makereport(item, call): call.when == "call" and report.failed and call.excinfo is not None - and call.excinfo.errisinstance(NETWORK_ERRORS) + and is_network_error(call.excinfo.value) and item.get_closest_marker("network") ): report.outcome = "skipped" From 4bdae46d7e5e8972764ca700882982fc1498b27d Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Wed, 28 Oct 2026 21:19:23 +0100 Subject: [PATCH 5/5] Add timeouts to requests for remote files wfdb did not set a timeout for its HTTP requests, so a server that accepts the connection but does not respond (for example, PhysioNet when it is overloaded) blocked forever: scripts hung, and in the tests a [physionet] variant waited until the CI job timeout instead of being skipped. Use a timeout of 10 seconds to connect and 60 seconds to wait for data (not for the whole transfer, so slow but steady downloads still work). A requests.exceptions.Timeout is raised if the server does not respond, which the test suite already reports as a skip for tests marked "network". Add a delay option to the dummy HTTP server in the tests to test this. --- docs/changes.rst | 3 +++ tests/test_url.py | 22 ++++++++++++++++++++++ wfdb/io/_url.py | 10 +++++++++- 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/docs/changes.rst b/docs/changes.rst index c02048f7..fc7fc406 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -14,6 +14,9 @@ Unreleased **Raise minimum versions of dependencies** The minimum versions of ``soundfile`` (0.12.1), ``matplotlib`` (3.8.0) and ``requests`` (2.31.0) have been raised. The minimum versions of the other dependencies are unchanged. +**Add timeouts to requests for remote files** + Requests for files on PhysioNet (and other web servers) now time out if the server does not respond (10 seconds to connect, and 60 seconds to wait for data), instead of waiting indefinitely. A ``requests.exceptions.Timeout`` (``ConnectTimeout`` or ``ReadTimeout``) is raised in this case. + **Build with uv_build** The package is now built with ``uv_build`` instead of ``hatchling``. The version number is specified in ``pyproject.toml`` only, and ``wfdb.__version__`` is read from the installed package metadata (it is ``"unknown"`` if the package is run from a source tree without being installed). diff --git a/tests/test_url.py b/tests/test_url.py index 52e33764..680bf6a9 100644 --- a/tests/test_url.py +++ b/tests/test_url.py @@ -1,6 +1,10 @@ import gzip import http.server import threading +import time + +import pytest +import requests import wfdb.io._url @@ -10,6 +14,18 @@ class TestNetFiles: Test accessing remote files. """ + def test_timeout(self, monkeypatch): + """ + Test that a server that does not respond causes a timeout error. + + Without a timeout, the request would wait for the server forever. + """ + monkeypatch.setattr(wfdb.io._url, "TIMEOUT", (5, 0.2)) + with DummyHTTPServer({"/foo.txt": b"foo"}, delay=1.5) as server: + with pytest.raises(requests.exceptions.Timeout): + with wfdb.io._url.openurl(server.url("/foo.txt"), "rb") as f: + f.read() + def test_requests(self): """ Test reading a remote file using various APIs. @@ -273,6 +289,9 @@ class DummyHTTPServer(http.server.HTTPServer): True if the server should return partial responses (using 206 Partial Content and "Content-Range") when the client requests them (using "Range"). + delay : float, optional + Number of seconds that the server waits before it responds to a + request (to simulate a server that does not respond). server_address : tuple (str, int), optional A tuple specifying the address and port number where the server should listen for connections. If the port is 0, an @@ -286,12 +305,14 @@ def __init__( file_content, allow_gzip=True, allow_range=True, + delay=0, server_address=("127.0.0.1", 0), ): super().__init__(server_address, DummyHTTPRequestHandler) self.file_content = file_content self.allow_gzip = allow_gzip self.allow_range = allow_range + self.delay = delay def url(self, path="/"): """ @@ -342,6 +363,7 @@ def log_message(self, message, *args): pass def send_head(self): + time.sleep(self.server.delay) content = self.server.file_content.get(self.path) if content is None: self.send_error(404) diff --git a/wfdb/io/_url.py b/wfdb/io/_url.py index adc5245b..adccbcf5 100644 --- a/wfdb/io/_url.py +++ b/wfdb/io/_url.py @@ -16,6 +16,12 @@ # Default buffer size for remote files. DEFAULT_BUFFER_SIZE = 32768 +# Timeouts in seconds (connect, read) for requests. The read timeout is the +# maximum time to wait for the server to send data (not the total time of a +# transfer). Without a timeout, a server that accepts a connection but does +# not respond would block forever. +TIMEOUT = (10, 60) + # Logger for this module. _LOGGER = logging.getLogger(__name__) @@ -159,7 +165,9 @@ def __init__(self, url, start, end): } session = _get_session() - self._response = session.request(method, url, headers=headers, stream=True) + self._response = session.request( + method, url, headers=headers, stream=True, timeout=TIMEOUT + ) self._content_iter = self._response.iter_content(4096) try: self._parse_headers(method, self._response)