Use pytest only and run PhysioNet checks as separate tests

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"
+```

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

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 = ""

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

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

tests/test_annotation.py

@@ -1,15 +1,13 @@
 import os
 import re
-import tempfile
-import unittest
 
 import numpy as np
 import pandas as pd
 
 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,43 +229,45 @@ 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
 
         Annotation file created by:
             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"]