Skip to content

Commit

Permalink
cleanup/generalize
Browse files Browse the repository at this point in the history
  • Loading branch information
CodyCBakerPhD committed Sep 15, 2022
1 parent 5b750b6 commit 23760e6
Show file tree
Hide file tree
Showing 4 changed files with 64 additions and 43 deletions.
35 changes: 35 additions & 0 deletions nwbinspector/testing.py
Original file line number Diff line number Diff line change
@@ -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
21 changes: 20 additions & 1 deletion nwbinspector/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
23 changes: 4 additions & 19 deletions tests/test_inspector.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
from pathlib import Path
from unittest import TestCase
from datetime import datetime
from distutils.util import strtobool

import numpy as np
from pynwb import NWBFile, NWBHDF5IO, TimeSeries
Expand All @@ -25,25 +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

DANDI_TESTS_NONETWORK = os.environ.get("DANDI_TESTS_NONETWORK", "")
NO_NETWORK = strtobool(DANDI_TESTS_NONETWORK) if DANDI_TESTS_NONETWORK != "" else False
if not NO_NETWORK:
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 = NO_NETWORK or not HAVE_ROS3 or not HAVE_DANDI
DISABLE_STREAMING_TESTS, DISABLE_STREAMING_TESTS_REASON = check_streaming_tests_enabled()


def add_big_dataset_no_compression(nwbfile: NWBFile):
Expand Down Expand Up @@ -603,7 +588,7 @@ def test_dandiset_streaming_parallel():
)


@pytest.mark.skipif(DISABLE_STREAMING_TESTS, 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):
Expand Down
28 changes: 5 additions & 23 deletions tests/unit_tests/test_time_series.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,3 @@
import os
from distutils.util import strtobool

import numpy as np
import pynwb
import pytest
Expand All @@ -16,25 +13,10 @@
check_missing_unit,
check_resolution,
)
from nwbinspector.utils import get_package_version, robust_s3_read, is_module_installed

DANDI_TESTS_NONETWORK = os.environ.get("DANDI_TESTS_NONETWORK", "")
NO_NETWORK = strtobool(DANDI_TESTS_NONETWORK) if DANDI_TESTS_NONETWORK != "" else False
if not NO_NETWORK:
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
HAVE_DANDI = is_module_installed("dandi")
DISABLE_STREAMING_TESTS = NO_NETWORK or not HAVE_ROS3 or not HAVE_DANDI
from nwbinspector.testing import check_streaming_tests_enabled
from nwbinspector.utils import get_package_version, robust_s3_read

DISABLE_STREAMING_TESTS, DISABLE_STREAMING_TESTS_REASON = check_streaming_tests_enabled()


def test_check_regular_timestamps():
Expand Down Expand Up @@ -186,7 +168,7 @@ def test_check_unknown_resolution_pass():

@pytest.mark.skipif(
DISABLE_STREAMING_TESTS or get_package_version("hdmf") >= version.parse("3.3.1"),
reason="Needs h5py setup with ROS3, as well as 'hdmf<3.3.1'.",
reason=f"{DISABLE_STREAMING_TESTS_REASON}. Also needs 'hdmf<3.3.1'.",
)
def test_check_none_matnwb_resolution_pass():
"""
Expand Down

0 comments on commit 23760e6

Please sign in to comment.