Propagate download test with datalad on test/iotest

Author: samuelgarcia

neo/test/iotest/common_io_test.py

@@ -29,8 +29,9 @@
                             assert_sub_schema_is_lazy_loaded,
                             assert_children_empty)
 
-from neo.test.rawiotest.tools import (can_use_network, make_all_directories,
-                                   download_test_file, create_local_temp_dir)
+from neo.test.rawiotest.tools import can_use_network
+from neo.test.rawiotest.common_rawio_test import repo_for_test
+from neo.utils import download_dataset, get_local_testing_data_folder
 
 from neo.test.iotest.tools import (cleanup_test_file,
                                    close_object_safe, create_generic_io_object,
@@ -40,11 +41,8 @@
                                    iter_generic_readers, iter_read_objects,
                                    read_generic,
                                    write_generic)
-from neo.test.generate_datasets import generate_from_supported_objects
-
 
-# url_for_tests = "https://portal.g-node.org/neo/" #This is the old place
-url_for_tests = "https://web.gin.g-node.org/NeuralEnsemble/ephy_testing_data/raw/master/"
+from neo.test.generate_datasets import generate_from_supported_objects
 
 
 class BaseTestIO:
@@ -67,8 +65,8 @@ class BaseTestIO:
     # all IO test need to modify this:
     ioclass = None  # the IOclass to be tested
 
-    files_to_test = []  # list of files to test compliances
-    files_to_download = []  # when files are at G-Node
+    entities_to_test = []  # list of files to test compliances
+    entities_to_download = []  # when files are at gin
 
     # when reading then writing produces files with identical hashes
     hash_conserved_when_write_read = False
@@ -84,7 +82,6 @@ def setUp(self):
         '''
         Set up the test fixture.  This is run for every test
         '''
-        self.files_to_test = copy(self.__class__.files_to_test)
         self.higher = self.ioclass.supported_objects[0]
         self.shortname = self.ioclass.__name__.lower().rstrip('io')
         # these objects can both be written and read
@@ -93,11 +90,16 @@ def setUp(self):
         # these objects can be either written or read
         self.io_readorwrite = list(set(self.ioclass.readable_objects) |
                                    set(self.ioclass.writeable_objects))
-        self.create_local_dir_if_not_exists()
-        self.download_test_files_if_not_present()
+        
+        for remote_path in self.entities_to_download:
+            download_dataset(repo=repo_for_test, remote_path=remote_path)
+
         self.files_generated = []
         self.generate_files_for_io_able_to_write()
-        self.files_to_test.extend(self.files_generated)
+        
+        # be carefull self.entities_to_test is class attributes
+        self.files_to_test = [self.get_local_path(e) for e in self.entities_to_test]
+        self.entities_to_test += self.files_generated
 
     def create_local_dir_if_not_exists(self):
         '''
@@ -169,11 +171,33 @@ def able_to_write_or_read(self, writeread=False, readwrite=False):
 
         return True
 
+    #~ def get_filename_path(self, filename):
+        #~ '''
+        #~ Get the path to a filename in the current temporary file directory
+        #~ '''
+        #~ return os.path.join(self.local_test_dir, filename)
+
+    def get_local_base_folder(self):
+        return get_local_testing_data_folder()
+
+    def get_local_path(self, sub_path):
+        root_local_path = self.get_local_base_folder()
+        local_path = root_local_path / sub_path
+        # TODO later : remove the str when all IOs handle the Path stuff
+        local_path = str(local_path)
+        return local_path
+    
     def get_filename_path(self, filename):
-        '''
-        Get the path to a filename in the current temporary file directory
-        '''
-        return os.path.join(self.local_test_dir, filename)
+        # keep for backward compatibility
+        # will be removed soon
+        root_local_path = self.get_local_base_folder()
+        local_path = root_local_path / self.shortname / filename
+        # TODO later : remove the str when all IOs handle the Path stuff
+        local_path = str(local_path)
+        print('get_filename_path will be removed (use get_local_path() instead)', self.__class__.__name__, local_path)
+        return local_path
+
+
 
     def generic_io_object(self, filename=None, return_path=False, clean=False):
         '''

neo/test/iotest/test_alphaomegaio.py

@@ -9,9 +9,13 @@
 
 
 class TestAlphaOmegaIO(BaseTestIO, unittest.TestCase):
-    files_to_test = ['File_AlphaOmega_1.map',
-                     'File_AlphaOmega_2.map']
-    files_to_download = files_to_test
+    entities_to_download =[
+        'alphaomega'
+    ]
+    entities_to_test = [
+        'alphaomega/File_AlphaOmega_1.map',
+        'alphaomega/File_AlphaOmega_2.map'
+    ]
     ioclass = AlphaOmegaIO
 
 

neo/test/iotest/test_asciisignalio.py

@@ -16,11 +16,13 @@
 
 class TestAsciiSignalIOWithTestFiles(BaseTestIO, unittest.TestCase):
     ioclass = AsciiSignalIO
-    files_to_download = [  # 'File_asciisignal_1.asc',
-        'File_asciisignal_2.txt',
-        'File_asciisignal_3.txt',
+    entities_to_download = [
+        'asciisignal'
+    ]
+    entities_to_test = [
+        'asciisignal/File_asciisignal_2.txt',
+        'asciisignal/File_asciisignal_3.txt',
     ]
-    files_to_test = files_to_download
 
 
 class TestAsciiSignalIO(unittest.TestCase):

neo/test/iotest/test_asciispiketrainio.py

@@ -10,8 +10,12 @@
 
 class TestAsciiSpikeTrainIO(BaseTestIO, unittest.TestCase, ):
     ioclass = AsciiSpikeTrainIO
-    files_to_download = ['File_ascii_spiketrain_1.txt']
-    files_to_test = files_to_download
+    entities_to_download = [
+        'asciispiketrain'
+    ]
+    entities_to_test = [
+        'asciispiketrain/File_ascii_spiketrain_1.txt',
+    ]
 
 
 if __name__ == "__main__":

neo/test/iotest/test_axographio.py

@@ -11,27 +11,19 @@
 import numpy as np
 from numpy.testing import assert_equal
 import quantities as pq
-
+from neo.test.rawiotest.test_axographrawio import TestAxographRawIO
 
 class TestAxographIO(BaseTestIO, unittest.TestCase):
-    files_to_test = [
-        'AxoGraph_Graph_File',      # version 1 file, provided with AxoGraph
-        'AxoGraph_Digitized_File',  # version 2 file, provided with AxoGraph
-        'AxoGraph_X_File.axgx',     # version 5 file, provided with AxoGraph
-        'File_axograph.axgd',       # version 6 file
-        'episodic.axgd',
-        'events_and_epochs.axgx',
-        'written-by-axographio-with-linearsequence.axgx',
-        'written-by-axographio-without-linearsequence.axgx',
-        'corrupt-comment.axgx',
+    entities_to_download = [
+        'axograph'
     ]
-    files_to_download = files_to_test
+    entities_to_test = TestAxographRawIO.entities_to_test
     ioclass = AxographIO
 
     def test_version_1(self):
         """Test reading a version 1 AxoGraph file"""
 
-        filename = self.get_filename_path('AxoGraph_Graph_File')
+        filename = self.get_local_path('axograph/AxoGraph_Graph_File')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 1)
@@ -55,7 +47,7 @@ def test_version_1(self):
     def test_version_2(self):
         """Test reading a version 2 AxoGraph file"""
 
-        filename = self.get_filename_path('AxoGraph_Digitized_File')
+        filename = self.get_local_path('axograph/AxoGraph_Digitized_File')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 2)
@@ -85,7 +77,7 @@ def test_version_2(self):
     def test_version_5(self):
         """Test reading a version 5 AxoGraph file"""
 
-        filename = self.get_filename_path('AxoGraph_X_File.axgx')
+        filename = self.get_local_path('axograph/AxoGraph_X_File.axgx')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 5)
@@ -109,7 +101,7 @@ def test_version_5(self):
     def test_version_6(self):
         """Test reading a version 6 AxoGraph file"""
 
-        filename = self.get_filename_path('File_axograph.axgd')
+        filename = self.get_local_path('axograph/File_axograph.axgd')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 6)
@@ -133,7 +125,7 @@ def test_version_6(self):
     def test_file_written_by_axographio_package_with_linearsequence(self):
         """Test reading file written by axographio package with linearsequence time column"""
 
-        filename = self.get_filename_path('written-by-axographio-with-linearsequence.axgx')
+        filename = self.get_local_path('axograph/written-by-axographio-with-linearsequence.axgx')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 6)
@@ -157,7 +149,7 @@ def test_file_written_by_axographio_package_with_linearsequence(self):
     def test_file_written_by_axographio_package_without_linearsequence(self):
         """Test reading file written by axographio package without linearsequence time column"""
 
-        filename = self.get_filename_path('written-by-axographio-without-linearsequence.axgx')
+        filename = self.get_local_path('axograph/written-by-axographio-without-linearsequence.axgx')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 6)
@@ -181,7 +173,7 @@ def test_file_written_by_axographio_package_without_linearsequence(self):
     def test_file_with_corrupt_comment(self):
         """Test reading a file with a corrupt comment"""
 
-        filename = self.get_filename_path('corrupt-comment.axgx')
+        filename = self.get_local_path('axograph/corrupt-comment.axgx')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.annotations['format_ver'], 6)
@@ -205,7 +197,7 @@ def test_file_with_corrupt_comment(self):
     def test_multi_segment(self):
         """Test reading an episodic file into multiple Segments"""
 
-        filename = self.get_filename_path('episodic.axgd')
+        filename = self.get_local_path('axograph/episodic.axgd')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
 
@@ -228,7 +220,7 @@ def test_multi_segment(self):
     def test_force_single_segment(self):
         """Test reading an episodic file into one Segment"""
 
-        filename = self.get_filename_path('episodic.axgd')
+        filename = self.get_local_path('axograph/episodic.axgd')
         reader = AxographIO(filename=filename, force_single_segment=True)
         blk = reader.read_block(signal_group_mode='split-all')
 
@@ -251,7 +243,7 @@ def test_force_single_segment(self):
     def test_group_by_same_units(self):
         """Test reading with group-by-same-units"""
 
-        filename = self.get_filename_path('episodic.axgd')
+        filename = self.get_local_path('axograph/episodic.axgd')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='group-by-same-units')
 
@@ -274,7 +266,7 @@ def test_group_by_same_units(self):
     def test_events_and_epochs(self):
         """Test loading events and epochs"""
 
-        filename = self.get_filename_path('events_and_epochs.axgx')
+        filename = self.get_local_path('axograph/events_and_epochs.axgx')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
 
@@ -297,14 +289,14 @@ def test_rec_datetime(self):
 
         # file obtained in episodic acquisition mode has date and time on
         # separate lines of notes
-        filename = self.get_filename_path('episodic.axgd')
+        filename = self.get_local_path('axograph/episodic.axgd')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.rec_datetime, datetime(2018, 6, 7, 15, 11, 36))
 
         # file obtained in continuous acquisition mode has date and time in
         # single line of notes
-        filename = self.get_filename_path('events_and_epochs.axgx')
+        filename = self.get_local_path('axograph/events_and_epochs.axgx')
         reader = AxographIO(filename=filename)
         blk = reader.read_block(signal_group_mode='split-all')
         assert_equal(blk.rec_datetime, datetime(2019, 5, 25, 20, 16, 25))

neo/test/iotest/test_axonaio.py

@@ -16,13 +16,11 @@
 
 class TestAxonaIO(BaseTestIO, unittest.TestCase, ):
     ioclass = AxonaIO
-
-    files_to_download = [
-        'axona_raw.set',
-        'axona_raw.bin'
+    entities_to_download = [
+        'axona'
     ]
-    files_to_test = [
-        'axona_raw.set'
+    entities_to_test = [
+        'axona/axona_raw.set'
     ]
 
 

neo/test/iotest/test_axonio.py

@@ -7,30 +7,22 @@
 from neo.io import AxonIO
 from neo.test.iotest.common_io_test import BaseTestIO
 
+from neo.test.rawiotest.test_axonrawio import TestAxonRawIO
 
 class TestAxonIO(BaseTestIO, unittest.TestCase):
-    files_to_test = [
-        'File_axon_1.abf',
-        'File_axon_2.abf',
-        'File_axon_3.abf',
-        'File_axon_4.abf',
-        'File_axon_5.abf',
-        'File_axon_6.abf',
-        'File_axon_7.abf',
-        'test_file_edr3.abf',
-    ]
-    files_to_download = files_to_test
+    entities_to_download = TestAxonRawIO.entities_to_download
+    entities_to_test = TestAxonRawIO.entities_to_test
     ioclass = AxonIO
 
     def test_annotations(self):
-        reader = AxonIO(filename=self.get_filename_path('File_axon_2.abf'))
+        reader = AxonIO(filename=self.get_local_path('axon/File_axon_2.abf'))
         bl = reader.read_block()
         ev = bl.segments[0].events[0]
         assert 'comments' in ev.annotations
 
     def test_read_protocol(self):
-        for f in self.files_to_test:
-            filename = self.get_filename_path(f)
+        for f in self.entities_to_test:
+            filename = self.get_local_path(f)
             reader = AxonIO(filename=filename)
             bl = reader.read_block(lazy=True)
             if bl.annotations['abf_version'] >= 2.:

neo/test/iotest/test_bci2000.py

@@ -10,12 +10,14 @@
 
 class TestBCI2000IO(BaseTestIO, unittest.TestCase, ):
     ioclass = BCI2000IO
-    files_to_test = [
-        'eeg1_1.dat',
-        'eeg1_2.dat',
-        'eeg1_3.dat',
+    entities_to_download = [
+        'bci2000'
+    ]
+    entities_to_test = [
+        'bci2000/eeg1_1.dat',
+        'bci2000/eeg1_2.dat',
+        'bci2000/eeg1_3.dat',
     ]
-    files_to_download = files_to_test
 
 
 if __name__ == "__main__":