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/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/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 new file mode 100644 index 00000000..65c3ccb3 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,56 @@ +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, + requests.exceptions.ChunkedEncodingError, + requests.exceptions.Timeout, +) + + +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.""" + 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 is_network_error(call.excinfo.value) + 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/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..9ee80794 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, source, temp_path): """ Target file created with: rdann -r sample-data/100 -a atr > ann-1 @@ -77,29 +75,31 @@ def test_1(self): ] # 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=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"], ) 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): + def test_2(self, source, temp_path): """ Annotation file with many aux_note strings. @@ -152,28 +152,30 @@ def test_2(self): 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=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"], ) 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): + def test_3(self, source, temp_path): """ Annotation file with custom annotation types @@ -227,28 +229,30 @@ def test_3(self): ] # 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=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"], ) 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): + def test_4(self, temp_path): """ Read and write annotations with large time skips @@ -256,14 +260,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 +288,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..7eaac7b7 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. @@ -38,9 +37,9 @@ class TestRecord(unittest.TestCase): "base_datetime", ] - # ----------------------- 1. Basic Tests -----------------------# + # ------------------------ 1. Basic Tests ------------------------ # - def test_1a(self): + def test_1a(self, source, temp_path): """ Format 16, entire signal, digital. @@ -52,27 +51,29 @@ def test_1a(self): 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( "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, ) 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. @@ -87,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( @@ -104,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): + def test_1c(self, source, temp_path): """ Format 16, byte offset, selected duration, selected channels, digital, expanded format. @@ -130,28 +133,30 @@ def test_1c(self): 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=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, ) 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 @@ -166,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): + def test_1e(self, source, temp_path): """ Format 24, entire signal, digital. @@ -191,23 +198,25 @@ def test_1e(self): 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) - 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, ) 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): + def test_1f(self, temp_path): """ All binary formats, multiple signal files in one record. @@ -253,14 +262,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 +301,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 +318,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 +356,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 +406,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 +435,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 +444,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 +488,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 +505,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() @@ -528,9 +537,9 @@ 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): + def test_2a(self, source): """ Format 212, entire signal, physical. @@ -542,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): + def test_2b(self, source, temp_path): """ Format 212, selected duration, selected channel, digital. @@ -569,14 +580,15 @@ def test_2b(self): 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] @@ -592,18 +604,19 @@ 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, ) 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) - def test_2c(self): + def test_2c(self, temp_path): """ Format 212, entire signal, physical, odd sampled record. @@ -616,8 +629,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 +664,7 @@ def test_2e(self): # --------------------- 3. Multi-dat records --------------------- # - def test_3a(self): + def test_3a(self, source, temp_path): """ Multi-dat, entire signal, digital Target file created with: @@ -662,20 +675,24 @@ def test_3a(self): 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=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, ) 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. @@ -692,20 +709,22 @@ 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 -------------- # - def test_4a(self): + def test_4a(self, temp_path): """ Format 16, multi-samples per frame, skew, digital. @@ -729,17 +748,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, source, temp_path): """ Format 12, multi-samples per frame, skew, entire signal, digital. @@ -756,7 +775,8 @@ def test_4b(self): 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 @@ -766,18 +786,19 @@ 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, ) 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): + def test_4c(self, source, temp_path): """ Format 12, multi-samples per frame, skew, selected suration, selected channels, physical. @@ -792,13 +813,14 @@ def test_4c(self): 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 @@ -808,17 +830,18 @@ 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, ) 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): @@ -841,7 +864,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 +873,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 +884,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 +913,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 +1012,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): """ @@ -1089,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 @@ -1109,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 @@ -1129,11 +1143,12 @@ 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) -class TestTimeConversion(unittest.TestCase): +class TestTimeConversion: """ Test cases for time conversion """ @@ -1153,18 +1168,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 +1202,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 +1237,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 +1278,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 +1298,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 +1318,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 +1344,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 +1362,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 +1394,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 +1412,21 @@ 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"]) + @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. - 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!") + @pytest.mark.network + 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..680bf6a9 100644 --- a/tests/test_url.py +++ b/tests/test_url.py @@ -1,16 +1,31 @@ import gzip import http.server import threading -import unittest +import time + +import pytest +import requests import wfdb.io._url -class TestNetFiles(unittest.TestCase): +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. @@ -78,10 +93,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 +106,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 +119,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 +141,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 +155,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 +166,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 +183,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 +195,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 +207,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 +248,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 +259,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 @@ -274,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 @@ -287,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="/"): """ @@ -343,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) @@ -387,7 +408,3 @@ def send_head(self): self.send_header(h, v) self.end_headers() return content - - -if __name__ == "__main__": - unittest.main() 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)