register_new_session test util

ibllib/tests/fixtures/utils.py

@@ -3,10 +3,17 @@
 # @Author: Niccolò Bonacchi
 # @Date: Friday, October 9th 2020, 12:02:56 pm
 import json
+import random
+import string
+import logging
 from pathlib import Path
 
+from one.registration import RegistrationClient
+
 from ibllib.io import session_params
 
+_logger = logging.getLogger(__name__)
+
 
 def create_fake_session_folder(
     root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True
@@ -374,3 +381,36 @@ def create_fake_ephys_recording_bad_passive_transfer_sessions(
     populate_task_settings(fpath, passive_settings)
 
     return session_path, passive_session_path
+
+
+def register_new_session(one, subject=None, date=None):
+    """
+    Register a new test session.
+
+    NB: This creates the session path on disk, using `one.cache_dir`.
+
+    Parameters
+    ----------
+    one : one.api.OneAlyx
+        An instance of ONE.
+    subject : str
+        The subject name. If None, a new random subject is created.
+    date : str
+        An ISO date string. If None, a random one is created.
+
+    Returns
+    -------
+    pathlib.Path
+        New local session path.
+    uuid.UUID
+        The experiment UUID.
+    """
+    if not date:
+        date = f'20{random.randint(0, 99):02}-{random.randint(1, 12):02}-{random.randint(1, 28):02}'
+    if not subject:
+        subject = ''.join(random.choices(string.ascii_letters, k=10))
+        one.alyx.rest('subjects', 'create', data={'lab': 'mainenlab', 'nickname': subject})
+
+    session_path, eid = RegistrationClient(one).create_new_session(subject, date=str(date)[:10])
+    _logger.debug('Registered session %s with eid %s', session_path, eid)
+    return session_path, eid

ibllib/tests/qc/test_alignment_qc.py

@@ -7,17 +7,16 @@
 import copy
 import random
 import string
-import datetime
 
 from one.api import ONE
 from neuropixel import trace_header
 
 from ibllib.tests import TEST_DB
+from ibllib.tests.fixtures.utils import register_new_session
 from iblatlas.atlas import AllenAtlas
 from ibllib.pipes.misc import create_alyx_probe_insertions
 from ibllib.qc.alignment_qc import AlignmentQC
 from ibllib.pipes.histology import register_track, register_chronic_track
-from one.registration import RegistrationClient
 
 EPHYS_SESSION = 'b1c968ad-4874-468d-b2e4-5ffa9b9964e9'
 one = ONE(**TEST_DB)
@@ -34,11 +33,9 @@ class TestTracingQc(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls) -> None:
-        rng = np.random.default_rng()
         probe = [''.join(random.choices(string.ascii_letters, k=5)),
                  ''.join(random.choices(string.ascii_letters, k=5))]
-        date = str(datetime.date(2019, rng.integers(1, 12), rng.integers(1, 28)))
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(one, subject='ZM_1150')
         cls.eid = str(eid)
         # Currently the task protocol of a session must contain 'ephys' in order to create an insertion!
         one.alyx.rest('sessions', 'partial_update', id=cls.eid, data={'task_protocol': 'ephys'})
@@ -75,13 +72,11 @@ def tearDownClass(cls) -> None:
 class TestChronicTracingQC(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:
-        rng = np.random.default_rng()
         probe = ''.join(random.choices(string.ascii_letters, k=5))
         serial = ''.join(random.choices(string.ascii_letters, k=10))
 
         # Make a chronic insertions
-        date = str(datetime.date(2019, rng.integers(1, 12), rng.integers(1, 28)))
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(one, subject='ZM_1150')
         cls.eid = str(eid)
         # Currently the task protocol of a session must contain 'ephys' in order to create an insertion!
         one.alyx.rest('sessions', 'partial_update', id=cls.eid, data={'task_protocol': 'ephys'})
@@ -142,7 +137,6 @@ class TestAlignmentQcExisting(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls) -> None:
-        rng = np.random.default_rng()
         data = np.load(Path(Path(__file__).parent.parent.
                             joinpath('fixtures', 'qc', 'data_alignmentqc_existing.npz')),
                        allow_pickle=True)
@@ -155,8 +149,7 @@ def setUpClass(cls) -> None:
         insertion = data['insertion'].tolist()
         insertion['name'] = ''.join(random.choices(string.ascii_letters, k=5))
         insertion['json'] = {'xyz_picks': cls.xyz_picks}
-        date = str(datetime.date(2019, rng.integers(1, 12), rng.integers(1, 28)))
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(one, subject='ZM_1150')
         cls.eid = str(eid)
         # Currently the task protocol of a session must contain 'ephys' in order to create an insertion!
         one.alyx.rest('sessions', 'partial_update', id=cls.eid, data={'task_protocol': 'ephys'})
@@ -265,7 +258,6 @@ class TestAlignmentQcManual(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls) -> None:
-        rng = np.random.default_rng()
         fixture_path = Path(__file__).parent.parent.joinpath('fixtures', 'qc')
         data = np.load(fixture_path / 'data_alignmentqc_manual.npz', allow_pickle=True)
         cls.xyz_picks = (data['xyz_picks'] * 1e6).tolist()
@@ -277,8 +269,7 @@ def setUpClass(cls) -> None:
         insertion['name'] = ''.join(random.choices(string.ascii_letters, k=5))
         insertion['json'] = {'xyz_picks': cls.xyz_picks}
 
-        date = str(datetime.date(2018, rng.integers(1, 12), rng.integers(1, 28)))
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(one, subject='ZM_1150')
         cls.eid = str(eid)
         insertion['session'] = cls.eid
         probe_insertion = one.alyx.rest('insertions', 'create', data=insertion)

ibllib/tests/qc/test_base_qc.py

@@ -1,13 +1,12 @@
 import unittest
 from unittest import mock
-import random
 
 import numpy as np
+from one.api import ONE
 
 from ibllib.tests import TEST_DB
 from ibllib.qc.base import QC
-from one.api import ONE
-from one.registration import RegistrationClient
+from ibllib.tests.fixtures.utils import register_new_session
 
 one = ONE(**TEST_DB)
 
@@ -20,8 +19,7 @@ class TestQC(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        date = f'20{random.randint(0, 30):02}-{random.randint(1, 12):02}-{random.randint(1, 28):02}'
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(one, subject='ZM_1150')
         cls.eid = str(eid)
 
     def setUp(self) -> None:

ibllib/tests/qc/test_critical_reasons.py

@@ -3,14 +3,12 @@
 import json
 import random
 import string
-import datetime
 
-import numpy as np
 import requests
 from one.api import ONE
-from one.registration import RegistrationClient
 
 from ibllib.tests import TEST_DB
+from ibllib.tests.fixtures.utils import register_new_session
 import ibllib.qc.critical_reasons as usrpmt
 
 one = ONE(**TEST_DB)
@@ -28,12 +26,10 @@ def mock_input(prompt):
 class TestUserPmtSess(unittest.TestCase):
 
     def setUp(self) -> None:
-        rng = np.random.default_rng()
         # Make sure tests use correct session ID
         one.alyx.clear_rest_cache()
         # Create new session on database with a random date to avoid race conditions
-        date = str(datetime.date(2022, rng.integers(1, 12), rng.integers(1, 28)))
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(one, subject='ZM_1150')
         eid = str(eid)
         # Currently the task protocol of a session must contain 'ephys' in order to create an insertion!
         one.alyx.rest('sessions', 'partial_update', id=eid, data={'task_protocol': 'ephys'})
@@ -159,7 +155,7 @@ def tearDown(self) -> None:
 
 class TestSignOffNote(unittest.TestCase):
     def setUp(self) -> None:
-        path, eid = RegistrationClient(one).create_new_session('ZM_1743')
+        path, eid = register_new_session(one, subject='ZM_1743')
         self.eid = str(eid)
         self.sign_off_keys = ['biasedChoiceWorld_00', 'passiveChoiceWorld_01']
         data = {'sign_off_checklist': dict.fromkeys(map(lambda x: f'{x}', self.sign_off_keys)),

ibllib/tests/qc/test_task_qc_viewer.py

@@ -77,7 +77,7 @@ def test_show_session_task_qc(self, trials_tasks_mock, run_app_mock):
         qc_mock.extractor.data = {'intervals': np.array([[0, 1]])}
         qc_mock.extractor.frame_ttls = qc_mock.extractor.audio_ttls = qc_mock.extractor.bpod_ttls = mock.MagicMock()
 
-        active_task = mock.Mock(spec=ChoiceWorldTrialsNidq)
+        active_task = mock.Mock(spec=ChoiceWorldTrialsNidq, unsafe=True)
         active_task.run_qc.return_value = qc_mock
         active_task.name = 'Trials_activeChoiceWorld_01'
         trials_tasks_mock.return_value = [passive_task, active_task]

ibllib/tests/test_pipes.py

@@ -8,8 +8,6 @@
 from pathlib import Path
 from unittest import mock
 from functools import partial
-import numpy as np
-import datetime
 import random
 import string
 from uuid import uuid4
@@ -289,9 +287,7 @@ def test_create_alyx_probe_insertions(self):
         # Connect to test DB
         one = ONE(**TEST_DB)
         # Create new session on database with a random date to avoid race conditions
-        date = str(datetime.date(2022, np.random.randint(1, 12), np.random.randint(1, 28)))
-        from one.registration import RegistrationClient
-        _, eid = RegistrationClient(one).create_new_session('ZM_1150', date=date)
+        _, eid = fu.register_new_session(one, subject='ZM_1150')
         eid = str(eid)
         # Currently the task protocol of a session must contain 'ephys' in order to create an insertion!
         one.alyx.rest('sessions', 'partial_update', id=eid, data={'task_protocol': 'ephys'})

ibllib/tests/test_plots.py

@@ -5,13 +5,12 @@
 from pathlib import Path
 from PIL import Image
 from urllib.parse import urlparse
-import datetime
-import numpy as np
 
 from one.api import ONE
 from one.webclient import http_download_file
 
 from ibllib.tests import TEST_DB
+from ibllib.tests.fixtures.utils import register_new_session
 from ibllib.plots.snapshot import Snapshot
 from ibllib.plots.figures import dlc_qc_plot
 
@@ -34,9 +33,7 @@ def setUpClass(cls):
         cls.notes = []
 
         # make a new test session
-        date = str(datetime.date(2018, np.random.randint(1, 12), np.random.randint(1, 28)))
-        from one.registration import RegistrationClient
-        _, eid = RegistrationClient(cls.one).create_new_session('ZM_1150', date=date)
+        _, eid = register_new_session(cls.one, subject='ZM_1150')
         cls.eid = str(eid)
 
     def _get_image(self, url):