add test that monkeypatches funloc to emulate session-specific MRIs

mne_bids_pipeline/tests/test_run.py

@@ -9,7 +9,9 @@
 from typing import Any, TypedDict
 
 import pytest
+from mne_bids import BIDSPath, get_bids_path_from_fname
 
+from mne_bids_pipeline._config_import import _import_config
 from mne_bids_pipeline._download import main as download_main
 from mne_bids_pipeline._main import main
 
@@ -135,7 +137,10 @@ class _TestOptionsT(TypedDict, total=False):
     "MNE-phantom-KIT-data": {
         "config": "config_MNE_phantom_KIT_data.py",
     },
-    "MNE-funloc-data": {"config": "config_funloc.py"},
+    "MNE-funloc-data": {
+        "config": "config_funloc.py",
+        "steps": ["init", "preprocessing", "sensor", "source"],
+    },
 }
 
 
@@ -275,3 +280,98 @@ def test_missing_sessions(
         print()
         with context:
             main()
+
+
+@pytest.mark.dataset_test
+@pytest.mark.parametrize("dataset", ["MNE-funloc-data"])
+def test_session_specific_mri(
+    dataset: str,
+    monkeypatch: pytest.MonkeyPatch,
+    dataset_test: Any,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Test of (faked) session-specific MRIs."""
+    test_options = TEST_SUITE[dataset]
+    config = test_options.get("config", f"config_{dataset}.py")
+    config_path = BIDS_PIPELINE_DIR / "tests" / "configs" / config
+    # copy the dataset to a tmpdir
+    config_obj = _import_config(config_path=config_path)
+    # make it seem like there's only one subj with different MRIs for different sessions
+    new_bids_path = BIDSPath(root=tmp_path / dataset, subject="01", session="a")
+    # sub-01/* → sub-01/ses-a/*
+    src_dir = config_obj.bids_root / "sub-01"
+    dst_dir = new_bids_path.root / "sub-01" / "ses-a"
+    for root, dirs, files in src_dir.walk():
+        offset = root.relative_to(src_dir)
+        for _dir in dirs:
+            (dst_dir / offset / _dir).mkdir(parents=True)
+        for _file in files:
+            bp = get_bids_path_from_fname(root / _file)
+            bp.update(root=new_bids_path.root, session="a")
+            shutil.copyfile(src=root / _file, dst=dst_dir / offset / bp.basename)
+    # sub-02/* → sub-01/ses-b/*
+    src_dir = config_obj.bids_root / "sub-02"
+    dst_dir = new_bids_path.root / "sub-01" / "ses-b"
+    for root, dirs, files in src_dir.walk():
+        offset = root.relative_to(src_dir)
+        for _dir in dirs:
+            (dst_dir / offset / _dir).mkdir(parents=True)
+        for _file in files:
+            bp = get_bids_path_from_fname(root / _file)
+            bp.update(root=new_bids_path.root, subject="01", session="b")
+            shutil.copyfile(src=root / _file, dst=dst_dir / offset / bp.basename)
+    # emptyroom
+    src_dir = config_obj.bids_root / "sub-emptyroom"
+    dst_dir = new_bids_path.root / "sub-emptyroom"
+    shutil.copytree(src=src_dir, dst=dst_dir)
+    # root-level files (dataset description, etc)
+    src_dir = config_obj.bids_root
+    dst_dir = new_bids_path.root
+    files = [f for f in src_dir.iterdir() if f.is_file()]
+    for _file in files:
+        shutil.copyfile(src=_file, dst=dst_dir / _file.name)
+    # now move derivatives (freesurfer files)
+    src_dir = config_obj.bids_root / "derivatives" / "freesurfer" / "subjects"
+    dst_dir = new_bids_path.root / "derivatives" / "freesurfer" / "subjects"
+    dst_dir.mkdir(parents=True)
+    for root, dirs, files in src_dir.walk():
+        new_root = root
+        if "sub01" in root.parts:
+            new_root = Path(
+                *["sub-01_ses-a" if p == "sub01" else p for p in new_root.parts]
+            )
+        elif "sub02" in root.parts:
+            new_root = Path(
+                *["sub-01_ses-b" if p == "sub02" else p for p in new_root.parts]
+            )
+        offset = new_root.relative_to(src_dir)
+        for _dir in dirs:
+            if _dir == "sub01":
+                _dir = "sub-01_ses-a"
+            elif _dir == "sub02":
+                _dir = "sub-01_ses-b"
+            (dst_dir / offset / _dir).mkdir()
+        for _file in files:
+            dst_file = _file
+            if _file.startswith("sub01"):
+                dst_file = dst_file.replace("sub01", "sub-01_ses-a")
+            elif _file.startswith("sub02"):
+                dst_file = dst_file.replace("sub02", "sub-01_ses-b")
+            shutil.copyfile(src=root / _file, dst=dst_dir / offset / dst_file)
+    # print_dir_tree(new_bids_path.root)  # for debugging
+    # hack in the new bids_root
+    extra_config = dict(bids_root=str(new_bids_path.root))
+    extra_path = tmp_path / "extra_config.py"
+    extra_path.write_text(str(extra_config))
+    monkeypatch.setenv("_MNE_BIDS_STUDY_TESTING_EXTRA_CONFIG", str(extra_path))
+    # Run the tests.
+    steps = test_options.get("steps", ())
+    command = ["mne_bids_pipeline", str(config_path), f"--steps={','.join(steps)}"]
+    if "--pdb" in sys.argv:
+        command.append("--n_jobs=1")
+    monkeypatch.setenv("_MNE_BIDS_STUDY_TESTING", "true")
+    monkeypatch.setattr(sys, "argv", command)
+    with capsys.disabled():
+        print()
+        main()