From aa06b7bd5bf7352c19df8c2ce21556151cab6a0f Mon Sep 17 00:00:00 2001
From: Ryan Ly <rly@lbl.gov>
Date: Tue, 3 Oct 2023 19:18:36 -0700
Subject: [PATCH] Fix warnings from tests (#1778)
---
setup.cfg | 1 +
src/pynwb/testing/mock/ophys.py | 30 ++-
test.py | 5 +-
tests/integration/hdf5/test_ecephys.py | 245 +++++++++++++++----------
tests/read_dandi/test_read_dandi.py | 90 +++++----
tests/unit/test_ecephys.py | 36 +++-
tests/unit/test_file.py | 5 +-
tests/unit/test_ophys.py | 26 ++-
tests/unit/test_resources.py | 12 +-
tests/validation/test_validate.py | 45 +++--
10 files changed, 315 insertions(+), 180 deletions(-)
diff --git a/setup.cfg b/setup.cfg
index d7ff8e67..d8b79d64 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -28,6 +28,7 @@ per-file-ignores =
tests/integration/__init__.py:F401
src/pynwb/testing/__init__.py:F401
src/pynwb/validate.py:T201
+ tests/read_dandi/test_read_dandi.py:T201
setup.py:T201
test.py:T201
scripts/*:T201
diff --git a/src/pynwb/testing/mock/ophys.py b/src/pynwb/testing/mock/ophys.py
index 5b43828f..d9ba0257 100644
--- a/src/pynwb/testing/mock/ophys.py
+++ b/src/pynwb/testing/mock/ophys.py
@@ -4,7 +4,7 @@ import numpy as np
from hdmf.common.table import DynamicTableRegion
-from ... import NWBFile
+from ... import NWBFile, ProcessingModule
from ...device import Device
from ...ophys import (
RoiResponseSeries,
@@ -272,6 +272,8 @@ def mock_RoiResponseSeries(
else:
n_rois = 5
+ plane_seg = plane_segmentation or mock_PlaneSegmentation(n_rois=n_rois, nwbfile=nwbfile)
+
roi_response_series = RoiResponseSeries(
name=name if name is not None else name_generator("RoiResponseSeries"),
data=data if data is not None else np.ones((30, n_rois)),
@@ -280,7 +282,7 @@ def mock_RoiResponseSeries(
or DynamicTableRegion(
name="rois",
description="rois",
- table=plane_segmentation or mock_PlaneSegmentation(n_rois=n_rois, nwbfile=nwbfile),
+ table=plane_seg,
data=list(range(n_rois)),
),
resolution=resolution,
@@ -298,6 +300,9 @@ def mock_RoiResponseSeries(
if "ophys" not in nwbfile.processing:
nwbfile.create_processing_module("ophys", "ophys")
+ if plane_seg.name not in nwbfile.processing["ophys"].data_interfaces:
+ nwbfile.processing["ophys"].add(plane_seg)
+
nwbfile.processing["ophys"].add(roi_response_series)
return roi_response_series
@@ -309,9 +314,9 @@ def mock_DfOverF(
nwbfile: Optional[NWBFile] = None
) -> DfOverF:
df_over_f = DfOverF(
- roi_response_series=roi_response_series or [mock_RoiResponseSeries(nwbfile=nwbfile)],
name=name if name is not None else name_generator("DfOverF"),
)
+ plane_seg = mock_PlaneSegmentation(nwbfile=nwbfile)
if nwbfile is not None:
if "ophys" not in nwbfile.processing:
@@ -319,6 +324,14 @@ def mock_DfOverF(
nwbfile.processing["ophys"].add(df_over_f)
+ else:
+ pm = ProcessingModule(name="ophys", description="ophys")
+ pm.add(plane_seg)
+ pm.add(df_over_f)
+
+ df_over_f.add_roi_response_series(
+ roi_response_series or mock_RoiResponseSeries(nwbfile=nwbfile, plane_segmentation=plane_seg)
+ )
return df_over_f
@@ -328,13 +341,22 @@ def mock_Fluorescence(
nwbfile: Optional[NWBFile] = None,
) -> Fluorescence:
fluorescence = Fluorescence(
- roi_response_series=roi_response_series or [mock_RoiResponseSeries(nwbfile=nwbfile)],
name=name if name is not None else name_generator("Fluorescence"),
)
+ plane_seg = mock_PlaneSegmentation(nwbfile=nwbfile)
if nwbfile is not None:
if "ophys" not in nwbfile.processing:
nwbfile.create_processing_module("ophys", "ophys")
+
nwbfile.processing["ophys"].add(fluorescence)
+ else:
+ pm = ProcessingModule(name="ophys", description="ophys")
+ pm.add(plane_seg)
+ pm.add(fluorescence)
+
+ fluorescence.add_roi_response_series(
+ roi_response_series or mock_RoiResponseSeries(nwbfile=nwbfile, plane_segmentation=plane_seg)
+ )
return fluorescence
diff --git a/test.py b/test.py
index 96b33445..16191ae3 100755
--- a/test.py
+++ b/test.py
@@ -163,7 +163,7 @@ def validate_nwbs():
def get_namespaces(nwbfile):
comp = run(["python", "-m", "pynwb.validate",
- "--list-namespaces", "--cached-namespace", nwb],
+ "--list-namespaces", nwbfile],
stdout=PIPE, stderr=STDOUT, universal_newlines=True, timeout=30)
if comp.returncode != 0:
@@ -179,14 +179,13 @@ def validate_nwbs():
cmds = []
cmds += [["python", "-m", "pynwb.validate", nwb]]
- cmds += [["python", "-m", "pynwb.validate", "--cached-namespace", nwb]]
cmds += [["python", "-m", "pynwb.validate", "--no-cached-namespace", nwb]]
for ns in namespaces:
# for some reason, this logging command is necessary to correctly printing the namespace in the
# next logging command
logging.info("Namespace found: %s" % ns)
- cmds += [["python", "-m", "pynwb.validate", "--cached-namespace", "--ns", ns, nwb]]
+ cmds += [["python", "-m", "pynwb.validate", "--ns", ns, nwb]]
for cmd in cmds:
logging.info("Validating with \"%s\"." % (" ".join(cmd[:-1])))
diff --git a/tests/integration/hdf5/test_ecephys.py b/tests/integration/hdf5/test_ecephys.py
index df6e81df..ff67d27c 100644
--- a/tests/integration/hdf5/test_ecephys.py
+++ b/tests/integration/hdf5/test_ecephys.py
@@ -1,4 +1,5 @@
from hdmf.common import DynamicTableRegion
+from pynwb import NWBFile
from pynwb.ecephys import (
ElectrodeGroup,
@@ -14,7 +15,7 @@ from pynwb.ecephys import (
)
from pynwb.device import Device
from pynwb.file import ElectrodeTable as get_electrode_table
-from pynwb.testing import NWBH5IOMixin, AcquisitionH5IOMixin, TestCase
+from pynwb.testing import NWBH5IOMixin, AcquisitionH5IOMixin, NWBH5IOFlexMixin, TestCase
class TestElectrodeGroupIO(NWBH5IOMixin, TestCase):
@@ -38,27 +39,36 @@ class TestElectrodeGroupIO(NWBH5IOMixin, TestCase):
return nwbfile.get_electrode_group(self.container.name)
-class TestElectricalSeriesIO(AcquisitionH5IOMixin, TestCase):
+def setup_electrode_table():
+ table = get_electrode_table()
+ dev1 = Device(name='dev1')
+ group = ElectrodeGroup(
+ name='tetrode1',
+ description='tetrode description',
+ location='tetrode location',
+ device=dev1
+ )
+ for i in range(4):
+ table.add_row(location='CA1', group=group, group_name='tetrode1')
+ return table, group, dev1
- @staticmethod
- def make_electrode_table(self):
- """ Make an electrode table, electrode group, and device """
- self.table = get_electrode_table()
- self.dev1 = Device(name='dev1')
- self.group = ElectrodeGroup(name='tetrode1',
- description='tetrode description',
- location='tetrode location',
- device=self.dev1)
- for i in range(4):
- self.table.add_row(location='CA1', group=self.group, group_name='tetrode1')
- def setUpContainer(self):
- """ Return the test ElectricalSeries to read/write """
- self.make_electrode_table(self)
+class TestElectricalSeriesIO(NWBH5IOFlexMixin, TestCase):
+
+ def getContainerType(self):
+ return "ElectricalSeries"
+
+ def addContainer(self):
+ """ Add the test ElectricalSeries and related objects to the given NWBFile """
+ table, group, dev1 = setup_electrode_table()
+ self.nwbfile.add_device(dev1)
+ self.nwbfile.add_electrode_group(group)
+ self.nwbfile.set_electrode_table(table)
+
region = DynamicTableRegion(name='electrodes',
data=[0, 2],
description='the first and third electrodes',
- table=self.table)
+ table=table)
data = list(zip(range(10), range(10, 20)))
timestamps = list(map(lambda x: x/10., range(10)))
channel_conversion = [1., 2., 3., 4.]
@@ -71,14 +81,11 @@ class TestElectricalSeriesIO(AcquisitionH5IOMixin, TestCase):
filtering=filtering,
timestamps=timestamps
)
- return es
- def addContainer(self, nwbfile):
- """ Add the test ElectricalSeries and related objects to the given NWBFile """
- nwbfile.add_device(self.dev1)
- nwbfile.add_electrode_group(self.group)
- nwbfile.set_electrode_table(self.table)
- nwbfile.add_acquisition(self.container)
+ self.nwbfile.add_acquisition(es)
+
+ def getContainer(self, nwbfile: NWBFile):
+ return nwbfile.acquisition['test_eS']
def test_eg_ref(self):
"""
@@ -92,58 +99,70 @@ class TestElectricalSeriesIO(AcquisitionH5IOMixin, TestCase):
self.assertIsInstance(row2.iloc[0]['group'], ElectrodeGroup)
-class MultiElectricalSeriesIOMixin(AcquisitionH5IOMixin):
- """
- Mixin class for methods to run a roundtrip test writing an NWB file with multiple ElectricalSeries.
+class TestLFPIO(NWBH5IOFlexMixin, TestCase):
+
+ def getContainerType(self):
+ return "LFP"
- The abstract method setUpContainer needs to be implemented by classes that include this mixin.
- def setUpContainer(self):
- # return a test Container to read/write
- """
+ def addContainer(self):
+ table, group, dev1 = setup_electrode_table()
+ self.nwbfile.add_device(dev1)
+ self.nwbfile.add_electrode_group(group)
+ self.nwbfile.set_electrode_table(table)
- def setUpTwoElectricalSeries(self):
- """ Return two test ElectricalSeries to read/write """
- TestElectricalSeriesIO.make_electrode_table(self)
region1 = DynamicTableRegion(name='electrodes',
data=[0, 2],
description='the first and third electrodes',
- table=self.table)
+ table=table)
region2 = DynamicTableRegion(name='electrodes',
data=[1, 3],
description='the second and fourth electrodes',
- table=self.table)
+ table=table)
data1 = list(zip(range(10), range(10, 20)))
data2 = list(zip(reversed(range(10)), reversed(range(10, 20))))
timestamps = list(map(lambda x: x/10., range(10)))
es1 = ElectricalSeries(name='test_eS1', data=data1, electrodes=region1, timestamps=timestamps)
es2 = ElectricalSeries(name='test_eS2', data=data2, electrodes=region2, channel_conversion=[4., .4],
timestamps=timestamps)
- return es1, es2
+ lfp = LFP()
+ self.nwbfile.add_acquisition(lfp)
+ lfp.add_electrical_series([es1, es2])
- def addContainer(self, nwbfile):
- """ Add the test ElectricalSeries and related objects to the given NWBFile """
- nwbfile.add_device(self.dev1)
- nwbfile.add_electrode_group(self.group)
- nwbfile.set_electrode_table(self.table)
- nwbfile.add_acquisition(self.container)
+ def getContainer(self, nwbfile: NWBFile):
+ return nwbfile.acquisition['LFP']
-class TestLFPIO(MultiElectricalSeriesIOMixin, TestCase):
+class TestFilteredEphysIO(NWBH5IOFlexMixin, TestCase):
- def setUpContainer(self):
- """ Return a test LFP to read/write """
- es = self.setUpTwoElectricalSeries()
- lfp = LFP(es)
- return lfp
+ def getContainerType(self):
+ return "FilteredEphys"
+ def addContainer(self):
+ table, group, dev1 = setup_electrode_table()
+ self.nwbfile.add_device(dev1)
+ self.nwbfile.add_electrode_group(group)
+ self.nwbfile.set_electrode_table(table)
-class TestFilteredEphysIO(MultiElectricalSeriesIOMixin, TestCase):
+ region1 = DynamicTableRegion(name='electrodes',
+ data=[0, 2],
+ description='the first and third electrodes',
+ table=table)
+ region2 = DynamicTableRegion(name='electrodes',
+ data=[1, 3],
+ description='the second and fourth electrodes',
+ table=table)
+ data1 = list(zip(range(10), range(10, 20)))
+ data2 = list(zip(reversed(range(10)), reversed(range(10, 20))))
+ timestamps = list(map(lambda x: x/10., range(10)))
+ es1 = ElectricalSeries(name='test_eS1', data=data1, electrodes=region1, timestamps=timestamps)
+ es2 = ElectricalSeries(name='test_eS2', data=data2, electrodes=region2, channel_conversion=[4., .4],
+ timestamps=timestamps)
+ fe = FilteredEphys()
+ self.nwbfile.add_acquisition(fe)
+ fe.add_electrical_series([es1, es2])
- def setUpContainer(self):
- """ Return a test FilteredEphys to read/write """
- es = self.setUpTwoElectricalSeries()
- fe = FilteredEphys(es)
- return fe
+ def getContainer(self, nwbfile: NWBFile):
+ return nwbfile.acquisition['FilteredEphys']
class TestClusteringIO(AcquisitionH5IOMixin, TestCase):
@@ -165,28 +184,35 @@ class TestClusteringIO(AcquisitionH5IOMixin, TestCase):
return super().roundtripExportContainer(cache_spec)
-class EventWaveformConstructor(AcquisitionH5IOMixin, TestCase):
+class EventWaveformConstructor(NWBH5IOFlexMixin, TestCase):
+
+ def getContainerType(self):
+ return "SpikeEventSeries"
+
+ def addContainer(self):
+ """ Add the test SpikeEventSeries and related objects to the given NWBFile """
+ table, group, dev1 = setup_electrode_table()
+ self.nwbfile.add_device(dev1)
+ self.nwbfile.add_electrode_group(group)
+ self.nwbfile.set_electrode_table(table)
- def setUpContainer(self):
- """ Return a test EventWaveform to read/write """
- TestElectricalSeriesIO.make_electrode_table(self)
region = DynamicTableRegion(name='electrodes',
data=[0, 2],
description='the first and third electrodes',
- table=self.table)
- sES = SpikeEventSeries(name='test_sES',
- data=((1, 1), (2, 2), (3, 3)),
- timestamps=[0., 1., 2.],
- electrodes=region)
- ew = EventWaveform(sES)
- return ew
+ table=table)
+ ses = SpikeEventSeries(
+ name='test_sES',
+ data=((1, 1), (2, 2), (3, 3)),
+ timestamps=[0., 1., 2.],
+ electrodes=region
+ )
- def addContainer(self, nwbfile):
- """ Add the test EventWaveform and related objects to the given NWBFile """
- nwbfile.add_device(self.dev1)
- nwbfile.add_electrode_group(self.group)
- nwbfile.set_electrode_table(self.table)
- nwbfile.add_acquisition(self.container)
+ ew = EventWaveform()
+ self.nwbfile.add_acquisition(ew)
+ ew.add_spike_event_series(ses)
+
+ def getContainer(self, nwbfile: NWBFile):
+ return nwbfile.acquisition['EventWaveform']
class ClusterWaveformsConstructor(AcquisitionH5IOMixin, TestCase):
@@ -220,51 +246,66 @@ class ClusterWaveformsConstructor(AcquisitionH5IOMixin, TestCase):
return super().roundtripExportContainer(cache_spec)
-class FeatureExtractionConstructor(AcquisitionH5IOMixin, TestCase):
+class FeatureExtractionConstructor(NWBH5IOFlexMixin, TestCase):
+
+ def getContainerType(self):
+ return "FeatureExtraction"
+
+ def addContainer(self):
+ """ Add the test FeatureExtraction and related objects to the given NWBFile """
+ table, group, dev1 = setup_electrode_table()
+ self.nwbfile.add_device(dev1)
+ self.nwbfile.add_electrode_group(group)
+ self.nwbfile.set_electrode_table(table)
- def setUpContainer(self):
- """ Return a test FeatureExtraction to read/write """
event_times = [1.9, 3.5]
- TestElectricalSeriesIO.make_electrode_table(self)
region = DynamicTableRegion(name='electrodes',
data=[0, 2],
description='the first and third electrodes',
- table=self.table)
+ table=table)
description = ['desc1', 'desc2', 'desc3']
features = [[[0., 1., 2.], [3., 4., 5.]], [[6., 7., 8.], [9., 10., 11.]]]
fe = FeatureExtraction(electrodes=region, description=description, times=event_times, features=features)
- return fe
- def addContainer(self, nwbfile):
- """ Add the test FeatureExtraction and related objects to the given NWBFile """
- nwbfile.add_device(self.dev1)
- nwbfile.add_electrode_group(self.group)
- nwbfile.set_electrode_table(self.table)
- nwbfile.add_acquisition(self.container)
+ self.nwbfile.add_acquisition(fe)
+ def getContainer(self, nwbfile: NWBFile):
+ return nwbfile.acquisition['FeatureExtraction']
-class EventDetectionConstructor(AcquisitionH5IOMixin, TestCase):
- def setUpContainer(self):
- """ Return a test EventDetection to read/write """
- TestElectricalSeriesIO.make_electrode_table(self)
+class EventDetectionConstructor(NWBH5IOFlexMixin, TestCase):
+
+ def getContainerType(self):
+ return "EventDetection"
+
+ def addContainer(self):
+ """ Add the test EventDetection and related objects to the given NWBFile """
+ table, group, dev1 = setup_electrode_table()
+ self.nwbfile.add_device(dev1)
+ self.nwbfile.add_electrode_group(group)
+ self.nwbfile.set_electrode_table(table)
+
region = DynamicTableRegion(name='electrodes',
data=[0, 2],
description='the first and third electrodes',
- table=self.table)
+ table=table)
data = list(range(10))
ts = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
- self.eS = ElectricalSeries(name='test_eS', data=data, electrodes=region, timestamps=ts)
- eD = EventDetection(detection_method='detection_method',
- source_electricalseries=self.eS,
- source_idx=(1, 2, 3),
- times=(0.1, 0.2, 0.3))
- return eD
+ eS = ElectricalSeries(
+ name='test_eS',
+ data=data,
+ electrodes=region,
+ timestamps=ts
+ )
+ eD = EventDetection(
+ detection_method='detection_method',
+ source_electricalseries=eS,
+ source_idx=(1, 2, 3),
+ times=(0.1, 0.2, 0.3)
+ )
- def addContainer(self, nwbfile):
- """ Add the test EventDetection and related objects to the given NWBFile """
- nwbfile.add_device(self.dev1)
- nwbfile.add_electrode_group(self.group)
- nwbfile.set_electrode_table(self.table)
- nwbfile.add_acquisition(self.eS)
- nwbfile.add_acquisition(self.container)
+ self.nwbfile.add_acquisition(eS)
+ self.nwbfile.add_acquisition(eD)
+
+ def getContainer(self, nwbfile: NWBFile):
+ return nwbfile.acquisition['EventDetection']
diff --git a/tests/read_dandi/test_read_dandi.py b/tests/read_dandi/test_read_dandi.py
index 84e9f3f6..0e0698d7 100644
--- a/tests/read_dandi/test_read_dandi.py
+++ b/tests/read_dandi/test_read_dandi.py
@@ -1,52 +1,62 @@
+"""Test reading NWB files from the DANDI Archive using ROS3."""
from dandi.dandiapi import DandiAPIClient
+import random
import sys
import traceback
from pynwb import NWBHDF5IO
-from pynwb.testing import TestCase
-class TestReadNWBDandisets(TestCase):
- """Test reading NWB files from the DANDI Archive using ROS3."""
+# NOTE: do not name the function with "test_" prefix, otherwise pytest
+# will try to run it as a test
+
+def read_first_nwb_asset():
+ """Test reading the first NWB asset from a random selection of 50 dandisets that uses NWB."""
+ num_dandisets_to_read = 50
+ client = DandiAPIClient()
+ dandisets = list(client.get_dandisets())
+ random.shuffle(dandisets)
+ dandisets_to_read = dandisets[:num_dandisets_to_read]
+ print("Reading NWB files from the following dandisets:")
+ print([d.get_raw_metadata()["identifier"] for d in dandisets_to_read])
+
+ failed_reads = dict()
+ for i, dandiset in enumerate(dandisets_to_read):
+ dandiset_metadata = dandiset.get_raw_metadata()
+
+ # skip any dandisets that do not use NWB
+ if not any(
+ data_standard["identifier"] == "RRID:SCR_015242" # this is the RRID for NWB
+ for data_standard in dandiset_metadata["assetsSummary"].get("dataStandard", [])
+ ):
+ continue
+
+ dandiset_identifier = dandiset_metadata["identifier"]
+ print("--------------")
+ print(f"{i}: {dandiset_identifier}")
+
+ # iterate through assets until we get an NWB file (it could be MP4)
+ assets = dandiset.get_assets()
+ first_asset = next(assets)
+ while first_asset.path.split(".")[-1] != "nwb":
+ first_asset = next(assets)
+ if first_asset.path.split(".")[-1] != "nwb":
+ print("No NWB files?!")
+ continue
- def test_read_first_nwb_asset(self):
- """Test reading the first NWB asset from each dandiset that uses NWB."""
- client = DandiAPIClient()
- dandisets = client.get_dandisets()
+ s3_url = first_asset.get_content_url(follow_redirects=1, strip_query=True)
- failed_reads = dict()
- for i, dandiset in enumerate(dandisets):
- dandiset_metadata = dandiset.get_raw_metadata()
+ try:
+ with NWBHDF5IO(path=s3_url, load_namespaces=True, driver="ros3") as io:
+ io.read()
+ except Exception as e:
+ print(traceback.format_exc())
+ failed_reads[dandiset] = e
- # skip any dandisets that do not use NWB
- if not any(
- data_standard["identifier"] == "RRID:SCR_015242" # this is the RRID for NWB
- for data_standard in dandiset_metadata["assetsSummary"].get("dataStandard", [])
- ):
- continue
+ if failed_reads:
+ print(failed_reads)
+ sys.exit(1)
- dandiset_identifier = dandiset_metadata["identifier"]
- print("--------------")
- print(f"{i}: {dandiset_identifier}")
- # iterate through assets until we get an NWB file (it could be MP4)
- assets = dandiset.get_assets()
- first_asset = next(assets)
- while first_asset.path.split(".")[-1] != "nwb":
- first_asset = next(assets)
- if first_asset.path.split(".")[-1] != "nwb":
- print("No NWB files?!")
- continue
-
- s3_url = first_asset.get_content_url(follow_redirects=1, strip_query=True)
-
- try:
- with NWBHDF5IO(path=s3_url, load_namespaces=True, driver="ros3") as io:
- io.read()
- except Exception as e:
- print(traceback.format_exc())
- failed_reads[dandiset] = e
-
- if failed_reads:
- print(failed_reads)
- sys.exit(1)
+if __name__ == "__main__":
+ read_first_nwb_asset()
diff --git a/tests/unit/test_ecephys.py b/tests/unit/test_ecephys.py
index 26320394..6f76a5e8 100644
--- a/tests/unit/test_ecephys.py
+++ b/tests/unit/test_ecephys.py
@@ -2,6 +2,7 @@ import warnings
import numpy as np
+from pynwb.base import ProcessingModule
from pynwb.ecephys import (
ElectricalSeries,
SpikeEventSeries,
@@ -217,7 +218,11 @@ class EventWaveformConstructor(TestCase):
table, region = self._create_table_and_region()
sES = SpikeEventSeries('test_sES', list(range(10)), list(range(10)), region)
- ew = EventWaveform(sES)
+ pm = ProcessingModule(name='test_module', description='a test module')
+ ew = EventWaveform()
+ pm.add(table)
+ pm.add(ew)
+ ew.add_spike_event_series(sES)
self.assertEqual(ew.spike_event_series['test_sES'], sES)
self.assertEqual(ew['test_sES'], ew.spike_event_series['test_sES'])
@@ -274,10 +279,25 @@ class LFPTest(TestCase):
)
return table, region
+ def test_init(self):
+ _, region = self._create_table_and_region()
+ eS = ElectricalSeries('test_eS', [0, 1, 2, 3], region, timestamps=[0.1, 0.2, 0.3, 0.4])
+ msg = (
+ "The linked table for DynamicTableRegion 'electrodes' does not share "
+ "an ancestor with the DynamicTableRegion."
+ )
+ with self.assertWarnsRegex(UserWarning, msg):
+ lfp = LFP(eS)
+ self.assertEqual(lfp.electrical_series.get('test_eS'), eS)
+ self.assertEqual(lfp['test_eS'], lfp.electrical_series.get('test_eS'))
+
def test_add_electrical_series(self):
lfp = LFP()
table, region = self._create_table_and_region()
eS = ElectricalSeries('test_eS', [0, 1, 2, 3], region, timestamps=[0.1, 0.2, 0.3, 0.4])
+ pm = ProcessingModule(name='test_module', description='a test module')
+ pm.add(table)
+ pm.add(lfp)
lfp.add_electrical_series(eS)
self.assertEqual(lfp.electrical_series.get('test_eS'), eS)
@@ -295,16 +315,24 @@ class FilteredEphysTest(TestCase):
return table, region
def test_init(self):
- table, region = self._create_table_and_region()
+ _, region = self._create_table_and_region()
eS = ElectricalSeries('test_eS', [0, 1, 2, 3], region, timestamps=[0.1, 0.2, 0.3, 0.4])
- fe = FilteredEphys(eS)
+ msg = (
+ "The linked table for DynamicTableRegion 'electrodes' does not share "
+ "an ancestor with the DynamicTableRegion."
+ )
+ with self.assertWarnsRegex(UserWarning, msg):
+ fe = FilteredEphys(eS)
self.assertEqual(fe.electrical_series.get('test_eS'), eS)
self.assertEqual(fe['test_eS'], fe.electrical_series.get('test_eS'))
def test_add_electrical_series(self):
- fe = FilteredEphys()
table, region = self._create_table_and_region()
eS = ElectricalSeries('test_eS', [0, 1, 2, 3], region, timestamps=[0.1, 0.2, 0.3, 0.4])
+ pm = ProcessingModule(name='test_module', description='a test module')
+ fe = FilteredEphys()
+ pm.add(table)
+ pm.add(fe)
fe.add_electrical_series(eS)
self.assertEqual(fe.electrical_series.get('test_eS'), eS)
self.assertEqual(fe['test_eS'], fe.electrical_series.get('test_eS'))
diff --git a/tests/unit/test_file.py b/tests/unit/test_file.py
index bb5c9c1e..c9bd98ad 100644
--- a/tests/unit/test_file.py
+++ b/tests/unit/test_file.py
@@ -563,9 +563,8 @@ class TestNoCacheSpec(TestCase):
with NWBHDF5IO(self.path, 'w') as io:
io.write(nwbfile, cache_spec=False)
- with self.assertWarnsWith(UserWarning, "No cached namespaces found in %s" % self.path):
- with NWBHDF5IO(self.path, 'r', load_namespaces=True) as reader:
- nwbfile = reader.read()
+ with NWBHDF5IO(self.path, 'r', load_namespaces=True) as reader:
+ nwbfile = reader.read()
class TestTimestampsRefDefault(TestCase):
diff --git a/tests/unit/test_ophys.py b/tests/unit/test_ophys.py
index 1ebb7c64..88bd2453 100644
--- a/tests/unit/test_ophys.py
+++ b/tests/unit/test_ophys.py
@@ -2,7 +2,7 @@ import warnings
import numpy as np
-from pynwb.base import TimeSeries
+from pynwb.base import TimeSeries, ProcessingModule
from pynwb.device import Device
from pynwb.image import ImageSeries
from pynwb.ophys import (
@@ -398,9 +398,15 @@ class RoiResponseSeriesConstructor(TestCase):
class DfOverFConstructor(TestCase):
def test_init(self):
+ pm = ProcessingModule(name='ophys', description="Optical physiology")
+
ps = create_plane_segmentation()
- rt_region = ps.create_roi_table_region(description='the second ROI', region=[1])
+ pm.add(ps)
+
+ dof = DfOverF()
+ pm.add(dof)
+ rt_region = ps.create_roi_table_region(description='the second ROI', region=[1])
rrs = RoiResponseSeries(
name='test_ts',
data=[1, 2, 3],
@@ -408,26 +414,32 @@ class DfOverFConstructor(TestCase):
unit='unit',
timestamps=[0.1, 0.2, 0.3]
)
+ dof.add_roi_response_series(rrs)
- dof = DfOverF(rrs)
self.assertEqual(dof.roi_response_series['test_ts'], rrs)
class FluorescenceConstructor(TestCase):
def test_init(self):
+ pm = ProcessingModule(name='ophys', description="Optical physiology")
+
ps = create_plane_segmentation()
- rt_region = ps.create_roi_table_region(description='the second ROI', region=[1])
+ pm.add(ps)
- ts = RoiResponseSeries(
+ ff = Fluorescence()
+ pm.add(ff)
+
+ rt_region = ps.create_roi_table_region(description='the second ROI', region=[1])
+ rrs = RoiResponseSeries(
name='test_ts',
data=[1, 2, 3],
rois=rt_region,
unit='unit',
timestamps=[0.1, 0.2, 0.3]
)
+ ff.add_roi_response_series(rrs)
- ff = Fluorescence(ts)
- self.assertEqual(ff.roi_response_series['test_ts'], ts)
+ self.assertEqual(ff.roi_response_series['test_ts'], rrs)
class ImageSegmentationConstructor(TestCase):
diff --git a/tests/unit/test_resources.py b/tests/unit/test_resources.py
index e04f5c65..108a7fd8 100644
--- a/tests/unit/test_resources.py
+++ b/tests/unit/test_resources.py
@@ -1,3 +1,5 @@
+import warnings
+
from pynwb.resources import HERD
from pynwb.testing import TestCase
@@ -7,5 +9,11 @@ class TestNWBContainer(TestCase):
"""
Test constructor
"""
- er = HERD()
- self.assertIsInstance(er, HERD)
+ with warnings.catch_warnings(record=True):
+ warnings.filterwarnings(
+ "ignore",
+ message=r"HERD is experimental .*",
+ category=UserWarning,
+ )
+ er = HERD()
+ self.assertIsInstance(er, HERD)
diff --git a/tests/validation/test_validate.py b/tests/validation/test_validate.py
index 813f8d4e..74ce0992 100644
--- a/tests/validation/test_validate.py
+++ b/tests/validation/test_validate.py
@@ -2,6 +2,7 @@ import subprocess
import re
from unittest.mock import patch
from io import StringIO
+import warnings
from pynwb.testing import TestCase
from pynwb import validate, NWBHDF5IO
@@ -29,8 +30,6 @@ class TestValidateCLI(TestCase):
"tests/back_compat/1.0.2_nwbfile.nwb"], capture_output=True)
stderr_regex = re.compile(
- r".*UserWarning: No cached namespaces found in tests/back_compat/1\.0\.2_nwbfile\.nwb\s*"
- r"warnings.warn\(msg\)\s*"
r"The file tests/back_compat/1\.0\.2_nwbfile\.nwb has no cached namespace information\. "
r"Falling back to PyNWB namespace information\.\s*"
)
@@ -47,8 +46,6 @@ class TestValidateCLI(TestCase):
"--ns", "notfound"], capture_output=True)
stderr_regex = re.compile(
- r".*UserWarning: No cached namespaces found in tests/back_compat/1\.0\.2_nwbfile\.nwb\s*"
- r"warnings.warn\(msg\)\s*"
r"The file tests/back_compat/1\.0\.2_nwbfile\.nwb has no cached namespace information\. "
r"Falling back to PyNWB namespace information\.\s*"
r"The namespace 'notfound' could not be found in PyNWB namespace information as only "
@@ -222,26 +219,44 @@ class TestValidateFunction(TestCase):
def test_validate_io_cached_extension(self):
"""Test that validating a file with cached spec against its cached namespaces succeeds."""
- with NWBHDF5IO('tests/back_compat/2.1.0_nwbfile_with_extension.nwb', 'r', load_namespaces=True) as io:
- errors = validate(io)
- self.assertEqual(errors, [])
+ with warnings.catch_warnings(record=True):
+ warnings.filterwarnings(
+ "ignore",
+ message=r"Ignoring cached namespace .*",
+ category=UserWarning,
+ )
+ with NWBHDF5IO('tests/back_compat/2.1.0_nwbfile_with_extension.nwb', 'r', load_namespaces=True) as io:
+ errors = validate(io)
+ self.assertEqual(errors, [])
def test_validate_io_cached_extension_pass_ns(self):
"""Test that validating a file with cached extension spec against the extension namespace succeeds."""
- with NWBHDF5IO('tests/back_compat/2.1.0_nwbfile_with_extension.nwb', 'r', load_namespaces=True) as io:
- errors = validate(io, 'ndx-testextension')
- self.assertEqual(errors, [])
+ with warnings.catch_warnings(record=True):
+ warnings.filterwarnings(
+ "ignore",
+ message=r"Ignoring cached namespace .*",
+ category=UserWarning,
+ )
+ with NWBHDF5IO('tests/back_compat/2.1.0_nwbfile_with_extension.nwb', 'r', load_namespaces=True) as io:
+ errors = validate(io, 'ndx-testextension')
+ self.assertEqual(errors, [])
def test_validate_io_cached_core_with_io(self):
"""
For back-compatability, test that validating a file with cached extension spec against the core
namespace succeeds when using the `io` + `namespace` keywords.
"""
- with NWBHDF5IO(
- path='tests/back_compat/2.1.0_nwbfile_with_extension.nwb', mode='r', load_namespaces=True
- ) as io:
- results = validate(io=io, namespace="core")
- self.assertEqual(results, [])
+ with warnings.catch_warnings(record=True):
+ warnings.filterwarnings(
+ "ignore",
+ message=r"Ignoring cached namespace .*",
+ category=UserWarning,
+ )
+ with NWBHDF5IO(
+ path='tests/back_compat/2.1.0_nwbfile_with_extension.nwb', mode='r', load_namespaces=True
+ ) as io:
+ results = validate(io=io, namespace="core")
+ self.assertEqual(results, [])
def test_validate_file_cached_extension(self):
"""
--
2.41.0