Add anatomical tests

.github/workflows/test.yaml

@@ -27,54 +27,55 @@ jobs:
     - uses: actions/checkout@v6
     - name: Setup venv
       uses: ./.github/actions/setup-venv
-    - name: Run quick tests
-      id: run-quick-tests
+
+    - name: Run non-ubuntu quick tests
+      id: run-unit-tests
+      if: matrix.os != 'ubuntu-latest'
+      shell: bash
+      run: >
+        uv run pytest \
+          -m "unit" \
+          --junitxml=pytest-quick.xml \
+          --cov=src tests \
+          --log-level=DEBUG \
+          --verbose
+    - name: Run ubuntu quick tests
+      id: run-quick-ubuntu-tests
+      if: matrix.os == 'ubuntu-latest'
       shell: bash
       run: >
         uv run pytest \
           -m "not slow and not full_pipeline" \
+          --runner=docker \
           --junitxml=pytest-quick.xml \
           --cov=src tests \
           --log-level=DEBUG \
           --verbose
 
-    - name: Run slow tests (manual trigger)
-      if: github.event_name == 'workflow_dispatch'
+    - name: Run slow tests
+      if: matrix.os == 'ubuntu-latest' && github.event_name == 'workflow_dispatch'
       shell: bash
       run: >
         uv run pytest \
-          -m "slow and not full_pipeline" \
+          -m "slow" \
+          --runner=docker \
           --junitxml=pytest-slow.xml \
           --cov=src tests \
           --cov-append \
           --verbose
 
-    - name: Run full pipeline tests (manual trigger)
-      if: github.event_name == 'workflow_dispatch' && inputs.e2e == true
+    - name: Run full pipeline tests
+      if: matrix.os == 'ubuntu-latest' && github.event_name == 'workflow_dispatch' && inputs.e2e == true
       shell: bash
       run: >
         uv run pytest \
           -m "full_pipeline" \
+          --runner=docker \
           --junitxml=pytest-full.xml \
           --cov=src tests \
           --cov-append \
           --verbose
 
-    - name: Generate coverage report
-      shell: bash
-      run: >
-        uv run pytest \
-          --cov-report=term-missing:skip-covered \
-          --cov-report=xml:coverage.xml \
-          --cov=src
-
-    - name: Upload coverage to Codecov
-      uses: codecov/codecov-action@671740ac38dd9b0130fbe1cec585b89eea48d3de
-      with:
-        token: ${{ secrets.CODECOV_TOKEN }}
-        files: ./coverage.xml
-        verbose: true
-
   ruff:
     runs-on: ubuntu-latest
     steps:

src/rbc/core/utils.py

@@ -2,7 +2,7 @@
 
 import shutil
 import tempfile
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from contextlib import contextmanager
 from pathlib import Path
 
@@ -29,20 +29,26 @@ def create_copy(in_file: str | Path) -> Iterator[Path]:
         shutil.rmtree(tmp_dir)
 
 
-def get_base_entities(in_file: Path) -> dict[str, str]:
+def get_base_entities(
+    in_file: Path, base_entities: Iterable[str] = ("sub", "ses", "run")
+) -> dict[str, str]:
     """Parse base BIDS entities to be used for file naming.
 
     Args:
         in_file: Path to parse bids entities for.
+        base_entities: List of base entities to extract (default: ['sub', 'ses' run'])
 
     Returns:
         A string-mapping of BIDS entities to values.
     """
     file_entities = parse_bids_entities(in_file)
-    return {k: v for k, v in file_entities.items() if k in ["sub", "ses", "run"]}
+    return {k: v for k, v in file_entities.items() if k in base_entities}
 
 
 def rename(in_file: str | Path, new_name: str | Path) -> Path:
     """Rename a file, keeping it in the same directory."""
     in_file = Path(in_file)
-    return in_file.rename(in_file.with_name(Path(new_name).name))
+    new_path = in_file.with_name(Path(new_name).name)
+    if new_path.exists():
+        raise FileExistsError(f"Target file already exists: {new_path}")
+    return in_file.rename(new_path)

tests/README.md

@@ -2,6 +2,11 @@
 
 Quick guide for running tests in this project.
 
+> [!NOTE]
+> Integration and full pipeline tests likely requires specific
+> neuroimaging tools that can be called with `niwrap`. To perform
+> these tests, also pass the `--runner` flag to `pytest`.
+
 ## Quick Start
 
 ```bash
@@ -12,10 +17,10 @@ pip install -e ".[dev]"
 pytest -m unit
 
 # Run all tests (except slow ones)
-pytest -m "not slow and not full_pipeline"
+pytest -m "not slow and not full_pipeline" --runner <local|docker|singularity>
 
 # Run everything
-pytest
+pytest --runner <local|docker|singularity>
 ```
 
 ## Test Categories

tests/conftest.py

@@ -1,19 +1,54 @@
 """Shared fixtures for tests data."""
 
+import logging
 from collections.abc import Sequence
 from pathlib import Path
 from types import SimpleNamespace
 
+import niwrap
 import pytest
 
 
 def pytest_collection_modifyitems(items: Sequence[pytest.Item]) -> None:
     """Apply appropriate markers based on test location."""
+    markers = {"unit", "integration", "full_pipeline"}
+
     for item in items:
         test_path = Path(item.fspath)
+        for marker in markers & set(test_path.parts):
+            item.add_marker(getattr(pytest.mark, marker))
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """Add option(s) to pytest parser."""
+    parser.addoption(
+        "--runner",
+        action="store",
+        default="docker",
+        help="Styx runner type to use: ['local', 'docker', 'singularity']",
+    )
+
 
-        if "unit" in test_path.parts:
-            item.add_marker(pytest.mark.unit)
+@pytest.fixture(scope="session", autouse=True)
+def niwrap_runner(
+    request: pytest.FixtureRequest,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> niwrap.Runner:
+    """Globally set test niwrap runner."""
+    # Set up niwrap runner
+    match request.config.getoption("--runner").lower():
+        case "docker":
+            niwrap.use_docker()
+        case "singularity":
+            niwrap.use_singularity()
+        case _:
+            niwrap.use_local()
+    runner = niwrap.get_global_runner()
+    runner.data_dir = tmp_path_factory.mktemp("styx_tmp")
+    # Set up logging for debugging
+    logger = logging.getLogger(runner.logger_name)
+    logger.setLevel(logging.DEBUG)
+    return runner
 
 
 @pytest.fixture

tests/full_pipeline/__init__.py

@@ -0,0 +1 @@
+"""Full pipeline (e2e) testing suite."""

tests/full_pipeline/test_anatomical.py

@@ -0,0 +1,28 @@
+"""Full e2e tests for different pipelines."""
+
+import pathlib as pl
+from types import SimpleNamespace
+
+from rbc.workflows.anatomical import single_session
+
+
+def test_single_session(test_subject: SimpleNamespace, tmp_path: pl.Path) -> None:
+    """e2e test for single session anatomical workflow."""
+    subject_id = f"sub-{test_subject.id}"
+    expected_output_dir = tmp_path / subject_id / "anat"
+    expected_fnames = [
+        "desc-T1w_mask.nii.gz",
+        "desc-brain_T1w.nii.gz",
+        "desc-csf_mask.nii.gz",
+        "desc-gm_mask.nii.gz",
+        "desc-wm_mask.nii.gz",
+        "from-T1w_to-template_mode-image_xfm.nii.gz",
+        "from-template_to-T1w_mode-image_xfm.nii.gz",
+    ]
+    single_session(test_subject.t1w, output_dir=tmp_path)
+
+    assert expected_output_dir.exists()
+    assert all(
+        (expected_output_dir / f"{subject_id}_{fname}").exists()
+        for fname in expected_fnames
+    )

tests/integration/__init__.py

@@ -0,0 +1 @@
+"""Integration testing suite."""

tests/integration/test_anatomical.py

@@ -0,0 +1,42 @@
+"""Integration tests for AFNI methods used across modalities."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from rbc.core import anatomical
+
+
+@pytest.mark.slow
+def test_brain_extraction(test_subject: SimpleNamespace) -> None:
+    """Test brain extraction."""
+    ants_bet_output = anatomical.ants_brain_extraction(
+        in_file=test_subject.t1w, output_prefix="test"
+    )
+    # Test extracted brain image exists
+    assert ants_bet_output.brain_extracted_image is not None
+    assert ants_bet_output.brain_extracted_image.exists()
+    # Test brain mask exists
+    assert ants_bet_output.brain_mask is not None
+    assert ants_bet_output.brain_mask.exists()
+
+
+@pytest.mark.slow
+def test_tissue_segmentation(test_subject: SimpleNamespace) -> None:
+    """Test tissue segmentation."""
+    tissue_mask = anatomical.fsl_tissue_segmentation(
+        in_file=test_subject.t1w, output_prefix="test"
+    )
+    assert tissue_mask.csf.exists()
+    assert tissue_mask.gm.exists()
+    assert tissue_mask.wm.exists()
+
+
+@pytest.mark.slow
+def test_registration(test_subject: SimpleNamespace) -> None:
+    """Test anatomical registration."""
+    composite_xfms = anatomical.ants_registration(
+        in_file=test_subject.t1w, output_prefix="test"
+    )
+    assert composite_xfms.forward.exists()
+    assert composite_xfms.inverse.exists()

tests/integration/test_common.py

@@ -0,0 +1,15 @@
+"""Integration tests for AFNI methods used across modalities."""
+
+from types import SimpleNamespace
+
+from niwrap import afni
+
+from rbc.core.common import reorient
+
+
+def test_reorient(test_subject: SimpleNamespace) -> None:
+    """Test deobliqueing and reorientation."""
+    reoriented_file = reorient(test_subject.t1w, output_fname="test.nii.gz")
+    assert (
+        afni.v_3dinfo(dataset=[reoriented_file.out_file], orient=True).info[0] == "RPI"
+    )

tests/unit/__init__.py

@@ -0,0 +1 @@
+"""Unit testing suite."""

tests/unit/conftest.py

@@ -0,0 +1,13 @@
+"""Fixtures specific to unit tests."""
+
+import pathlib as pl
+
+import pytest
+
+
+@pytest.fixture
+def test_file(tmp_path: pl.Path) -> pl.Path:
+    """Create sample file for testing."""
+    test_file = tmp_path / "test_file.txt"
+    test_file.write_text("Sample content")
+    return test_file