diff --git a/ibllib/pipes/local_server.py b/ibllib/pipes/local_server.py
index 92f1cf39a..b692fa42b 100644
--- a/ibllib/pipes/local_server.py
+++ b/ibllib/pipes/local_server.py
@@ -17,8 +17,7 @@
 from one.api import ONE
 from one.webclient import AlyxClient
 from one.remote.globus import get_lab_from_endpoint_id, get_local_endpoint_id
-from one.alf.spec import is_session_path
-from one.alf.path import session_path_parts
+from one.alf.path import ALFPath
 
 from ibllib import __version__ as ibllib_version
 from ibllib.pipes import tasks
@@ -82,12 +81,11 @@ def job_creator(root_path, one=None, dry=False, rerun=False):
      1) create the session on Alyx
      2) create the tasks to be run on Alyx
 
-    For legacy sessions the raw data are registered separately, instead of within a pipeline task.
-
     Parameters
     ----------
     root_path : str, pathlib.Path
-        Main path containing sessions or a session path.
+        Main path containing sessions or a session path. NB: If a session path is passed,
+        a raw_session.flag file needn't be present.
     one : one.api.OneAlyx
         An ONE instance for registering the session(s).
     dry : bool
@@ -106,13 +104,16 @@ def job_creator(root_path, one=None, dry=False, rerun=False):
     if not one:
         one = ONE(cache_rest=None)
     rc = IBLRegistrationClient(one=one)
-    flag_files = Path(root_path).glob('*/????-??-??/*/raw_session.flag')
-    flag_files = filter(lambda x: is_session_path(x.parent), flag_files)
+    if (root_path := ALFPath(root_path)).is_session_path():
+        flag_files = [root_path.joinpath('raw_session.flag')]
+    else:
+        flag_files = root_path.glob('*/????-??-??/*/raw_session.flag')
+        flag_files = filter(lambda f: f.parent.is_session_path(), flag_files)
     pipes = []
     all_datasets = []
     for flag_file in flag_files:
         session_path = flag_file.parent
-        if session_path_parts(session_path)[1] in ('test', 'test_subject'):
+        if session_path.subject in ('test', 'test_subject'):
             _logger.debug('skipping test session %s', session_path)
             continue
         _logger.info(f'creating session for {session_path}')
diff --git a/ibllib/tests/test_pipes.py b/ibllib/tests/test_pipes.py
index 6f35d164d..de42fa9be 100644
--- a/ibllib/tests/test_pipes.py
+++ b/ibllib/tests/test_pipes.py
@@ -60,6 +60,24 @@ def test_task_queue(self, lab_repo_mock):
         queue = local_server.task_queue(mode='small', lab='foolab', alyx=alyx)
         self.assertEqual([tasks[2]], queue)
 
+    def test_job_creator(self):
+        """Test ibllib.pipes.local_server.job_creator function.
+
+        This test simply checks that a specific session path can be passed and that a raw_session.flag
+        file is not necessary. For a full test of the job creator, see ci.tests.iblscripts.test_report_create_jobs
+        in iblscripts.
+        """
+        session_path = self.tmpdir / 'foo' / '2020-01-01' / '001'
+        assert not session_path.joinpath('raw_session.flag').exists()
+        with self.assertLogs('ibllib.pipes.local_server', level='INFO') as log:
+            local_server.job_creator(session_path, dry=True)
+        self.assertIn('creating session for', log.output[-1])
+        # Check skip when test subject
+        session_path = self.tmpdir / 'test' / '2020-01-01' / '001'
+        with self.assertLogs('ibllib.pipes.local_server', level='DEBUG') as log:
+            local_server.job_creator(session_path, dry=True)
+        self.assertIn('skipping test session', log.output[-1])
+
 
 class TestPipesMisc(unittest.TestCase):
     """"""