Add environmental control over network-dependent tests

docs/developer_guide.rst

@@ -35,3 +35,17 @@ Then, all that is needed for this to be automatically included when you run the
 the modules flag ``-m`` or ``--modules`` along with the name of your module that contains the custom check. If using
 the library instead, you need only import the ``available_checks`` global variable from your own submodules, or
 otherwise import your check functions after importing the ``nwbinspector`` in your ``__init__.py``.
+
+
+Disable Tests That Require Network Connection
+---------------------------------------------
+
+Some of the tests in the suite require internet connectivity both to and from the DANDI archive S3 bucket.
+If this is failing for some reason, you can explicitly control all related tests by setting the environment variable
+``NWBI_SKIP_NETWORK_TESTS`` to some value able to be parsed by ``distutils.util.str2tool``. For example, to disable them on
+a linux system, run
+
+.. code-block::
+    export NWBI_SKIP_NETWORK_TESTS=1
+
+in your environment before running ``pytest``.

nwbinspector/testing.py

@@ -0,0 +1,35 @@
+"""Helper functions for internal use across the testing suite."""
+import os
+from distutils.util import strtobool
+from typing import Tuple, Optional
+
+from .tools import check_streaming_enabled
+from .utils import is_module_installed
+
+
+def check_streaming_tests_enabled() -> Tuple[bool, Optional[str]]:
+    """
+    General purpose helper for determining if the testing environment can support S3 DANDI streaming.
+
+    Returns the boolean status of the check and, if False, provides a string reason for the failure for the user to
+    utilize as they please (raise an error or warning with that message, print it, or ignore it).
+    """
+    failure_reason = ""
+
+    environment_skip_flag = os.environ.get("NWBI_SKIP_NETWORK_TESTS", "")
+    environment_skip_flag_bool = (
+        strtobool(os.environ.get("NWBI_SKIP_NETWORK_TESTS", "")) if environment_skip_flag != "" else False
+    )
+    if environment_skip_flag_bool:
+        failure_reason += "Environmental variable set to skip network tets."
+
+    streaming_enabled, streaming_failure_reason = check_streaming_enabled()
+    if not streaming_enabled:
+        failure_reason += streaming_failure_reason
+
+    have_dandi = is_module_installed("dandi")
+    if not have_dandi:
+        failure_reason += "The DANDI package is not installed on the system."
+
+    failure_reason = None if failure_reason == "" else failure_reason
+    return streaming_enabled and not environment_skip_flag_bool and have_dandi, failure_reason

nwbinspector/tools.py

@@ -2,9 +2,12 @@
 import re
 from uuid import uuid4
 from datetime import datetime
-from typing import Optional, Dict
+from typing import Optional, Dict, Tuple
 from concurrent.futures import ProcessPoolExecutor, as_completed
+from urllib import request
+from warnings import warn
 
+import h5py
 from pynwb import NWBFile
 
 from .utils import is_module_installed, calculate_number_of_cpu
@@ -75,3 +78,19 @@ def _get_content_url_and_path(asset, follow_redirects: int = 1, strip_query: boo
     Must be globally defined (not as a part of get_s3_urls..) in order to be pickled.
     """
     return {asset.get_content_url(follow_redirects=1, strip_query=True): asset.path}
+
+
+def check_streaming_enabled() -> Tuple[bool, Optional[str]]:
+    """
+    General purpose helper for determining if the environment can support S3 DANDI streaming.
+
+    Returns the boolean status of the check and, if False, provides a string reason for the failure for the user to
+    utilize as they please (raise an error or warning with that message, print it, or ignore it).
+    """
+    try:
+        request.urlopen("https://dandiarchive.s3.amazonaws.com/ros3test.nwb", timeout=1)
+    except request.URLError:
+        return False, "Internet access to DANDI failed."
+    if "ros3" not in h5py.registered_drivers():
+        return False, "ROS3 driver not installed."
+    return True, None

tests/test_inspector.py

@@ -24,22 +24,11 @@
 )
 from nwbinspector import inspect_all, inspect_nwb
 from nwbinspector.register_checks import Severity, InspectorMessage, register_check
-from nwbinspector.utils import FilePathType, is_module_installed
+from nwbinspector.testing import check_streaming_tests_enabled
 from nwbinspector.tools import make_minimal_nwbfile
+from nwbinspector.utils import FilePathType
 
-
-try:
-    with NWBHDF5IO(
-        path="https://dandiarchive.s3.amazonaws.com/blobs/11e/c89/11ec8933-1456-4942-922b-94e5878bb991",
-        mode="r",
-        load_namespaces=True,
-        driver="ros3",
-    ) as io:
-        nwbfile = io.read()
-    HAVE_ROS3 = True
-except ValueError:  # ValueError: h5py was built without ROS3 support, can't use ros3 driver
-    HAVE_ROS3 = False
-HAVE_DANDI = is_module_installed("dandi")
+DISABLE_STREAMING_TESTS, DISABLE_STREAMING_TESTS_REASON = check_streaming_tests_enabled()
 
 
 def add_big_dataset_no_compression(nwbfile: NWBFile):
@@ -571,7 +560,7 @@ def test_inspect_nwb_dandi_config(self):
         self.assertCountEqual(first=test_results, second=true_results)
 
 
-@pytest.mark.skipif(not HAVE_ROS3 or not HAVE_DANDI, reason="Needs h5py setup with ROS3.")
+@pytest.mark.skipif(DISABLE_STREAMING_TESTS, reason="Needs h5py setup with ROS3.")
 def test_dandiset_streaming():
     messages = list(inspect_all(path="000126", select=["check_subject_species_exists"], stream=True))
     assert messages[0] == InspectorMessage(
@@ -585,7 +574,7 @@ def test_dandiset_streaming():
     )
 
 
-@pytest.mark.skipif(not HAVE_ROS3 or not HAVE_DANDI, reason="Needs h5py setup with ROS3.")
+@pytest.mark.skipif(DISABLE_STREAMING_TESTS, reason="Needs h5py setup with ROS3.")
 def test_dandiset_streaming_parallel():
     messages = list(inspect_all(path="000126", select=["check_subject_species_exists"], stream=True, n_jobs=2))
     assert messages[0] == InspectorMessage(
@@ -599,7 +588,7 @@ def test_dandiset_streaming_parallel():
     )
 
 
-@pytest.mark.skipif(not HAVE_ROS3 or not HAVE_DANDI, reason="Needs h5py setup with ROS3.")
+@pytest.mark.skipif(DISABLE_STREAMING_TESTS, reason=DISABLE_STREAMING_TESTS_REASON or "")
 class TestStreamingCLI(TestCase):
     @classmethod
     def setUpClass(cls):

tests/unit_tests/test_time_series.py

@@ -1,9 +1,7 @@
-from packaging import version
-from time import sleep
-
 import numpy as np
 import pynwb
 import pytest
+from packaging import version
 
 from nwbinspector import (
     InspectorMessage,
@@ -15,20 +13,10 @@
     check_missing_unit,
     check_resolution,
 )
+from nwbinspector.testing import check_streaming_tests_enabled
 from nwbinspector.utils import get_package_version, robust_s3_read
 
-try:
-    # Test ros3 on sub-YutaMouse54/sub-YutaMouse54_ses-YutaMouse54-160630_behavior+ecephys.nwb from #3
-    with pynwb.NWBHDF5IO(
-        path="https://dandiarchive.s3.amazonaws.com/blobs/f03/18e/f0318e30-4f4f-466d-a8e9-a962863e3081",
-        mode="r",
-        load_namespaces=True,
-        driver="ros3",
-    ) as io:
-        nwbfile = io.read()
-    HAVE_ROS3 = True
-except ValueError:  # ValueError: h5py was built without ROS3 support, can't use ros3 driver
-    HAVE_ROS3 = False
+DISABLE_STREAMING_TESTS, DISABLE_STREAMING_TESTS_REASON = check_streaming_tests_enabled()
 
 
 def test_check_regular_timestamps():
@@ -179,8 +167,8 @@ def test_check_unknown_resolution_pass():
 
 
 @pytest.mark.skipif(
-    not HAVE_ROS3 or get_package_version("hdmf") >= version.parse("3.3.1"),
-    reason="Needs h5py setup with ROS3, as well as 'hdmf<3.3.1'.",
+    DISABLE_STREAMING_TESTS or get_package_version("hdmf") >= version.parse("3.3.1"),
+    reason=f"{DISABLE_STREAMING_TESTS_REASON}. Also needs 'hdmf<3.3.1'.",
 )
 def test_check_none_matnwb_resolution_pass():
     """