Tighten Stage 4 tests (PR review fixups)

- Flag the xfm filename quirk (space-longitudinal + from-/to- is
  redundant) with a comment pointing at handover note #18, so the
  hard-coded expected names are not mistaken for a spec.
- Replace the Mock-based longitudinal anat outputs in the orchestration
  tests with a real AnatomicalLongOutputs NamedTuple so attribute typos
  fail loudly instead of silently returning a new Mock.
- Add resolve_longitudinal_anat unit tests covering the happy path,
  brain_mask optionality, and bids_safe_label round-trip for session
  labels with non-alphanumerics.

Author: nx10

tests/unit/bids/test_longitudinal_anatomical.py

@@ -4,7 +4,12 @@
 
 from typing import TYPE_CHECKING
 
-from rbc.bids.longitudinal.anatomical import export_longitudinal_anat
+import polars as pl
+
+from rbc.bids.longitudinal.anatomical import (
+    export_longitudinal_anat,
+    resolve_longitudinal_anat,
+)
 from rbc.context import RunContext
 from rbc.workflows.longitudinal.anatomical import AnatomicalLongOutputs
 
@@ -43,6 +48,11 @@ def test_writes_expected_files(self, tmp_path: Path) -> None:
         export_longitudinal_anat(aex, _make_long_outputs(workdir))
 
         saved = sorted(p.name for p in out_dir.rglob("*.*"))
+        # Xfm filenames currently carry both ``space-longitudinal`` and the
+        # ``from-…_to-…`` pair, which is redundant (Jason handover #18 flags
+        # the ``from-`` workaround more broadly). We assert the current shape
+        # so regressions are caught, but if that cleanup lands the expected
+        # xfm names here should drop ``space-longitudinal_``.
         assert saved == [
             "sub-01_ses-baseline_space-longitudinal_desc-T1w_mask.nii.gz",
             "sub-01_ses-baseline_space-longitudinal_desc-brain_T1w.nii.gz",
@@ -68,3 +78,150 @@ def test_tissue_masks_not_produced(self, tmp_path: Path) -> None:
             assert not any(f"desc-{tissue}_mask" in n for n in names), (
                 f"Expected no space-longitudinal {tissue}_mask; got {names}"
             )
+
+
+def _anat_row(
+    *,
+    sub: str,
+    ses: str,
+    suffix: str,
+    desc: str | None,
+    ext: str = ".nii.gz",
+    extra: list[dict[str, str]] | None = None,
+) -> dict[str, object]:
+    """Build a single BIDS-like row, including an ``extra_entities`` list."""
+    desc_part = f"_desc-{desc}" if desc else ""
+    path = f"sub-{sub}/ses-{ses}/anat/sub-{sub}_ses-{ses}{desc_part}_{suffix}{ext}"
+    return {
+        "datatype": "anat",
+        "suffix": suffix,
+        "ext": ext,
+        "sub": sub,
+        "ses": ses,
+        "desc": desc,
+        "root": "/data",
+        "path": path,
+        "extra_entities": extra or [],
+    }
+
+
+def _df(*rows: dict[str, object]) -> pl.DataFrame:
+    return pl.DataFrame(list(rows))
+
+
+class TestResolveLongitudinalAnat:
+    """Tests for :func:`resolve_longitudinal_anat`."""
+
+    def test_resolves_all_inputs(self, tmp_path: Path) -> None:
+        """Happy path: all four keys resolve to the expected paths."""
+        anat_df = _df(
+            _anat_row(sub="01", ses="baseline", suffix="T1w", desc="brain"),
+            _anat_row(sub="01", ses="baseline", suffix="mask", desc="T1w"),
+        )
+        tpl_df = _df(
+            _anat_row(sub="01", ses="longitudinal", suffix="T1w", desc=None),
+            _anat_row(
+                sub="01",
+                ses="longitudinal",
+                suffix="xfm",
+                desc=None,
+                ext=".txt",
+                extra=[
+                    {"key": "from", "value": "baseline"},
+                    {"key": "to", "value": "longitudinal"},
+                ],
+            ),
+        )
+
+        ctx = RunContext(sub="01", ses="baseline", output_dir=tmp_path)
+        anat_q = ctx.bids(datatype="anat")
+        tpl_q = anat_q.derive(ses="longitudinal")
+
+        resolved = resolve_longitudinal_anat(
+            anat_q, tpl_q, anat_df, tpl_df, ses="baseline"
+        )
+
+        assert set(resolved) == {
+            "template",
+            "subj_to_template_xfm",
+            "brain",
+            "brain_mask",
+        }
+        assert str(resolved["template"]).endswith("sub-01_ses-longitudinal_T1w.nii.gz")
+        assert str(resolved["subj_to_template_xfm"]).endswith(
+            "sub-01_ses-longitudinal_xfm.txt"
+        )
+        assert str(resolved["brain"]).endswith(
+            "sub-01_ses-baseline_desc-brain_T1w.nii.gz"
+        )
+        assert str(resolved["brain_mask"]).endswith(
+            "sub-01_ses-baseline_desc-T1w_mask.nii.gz"
+        )
+
+    def test_missing_brain_mask_is_none(self, tmp_path: Path) -> None:
+        """brain_mask is optional (``.find``) — returns None when absent."""
+        anat_df = _df(
+            _anat_row(sub="01", ses="baseline", suffix="T1w", desc="brain"),
+        )
+        tpl_df = _df(
+            _anat_row(sub="01", ses="longitudinal", suffix="T1w", desc=None),
+            _anat_row(
+                sub="01",
+                ses="longitudinal",
+                suffix="xfm",
+                desc=None,
+                ext=".txt",
+                extra=[
+                    {"key": "from", "value": "baseline"},
+                    {"key": "to", "value": "longitudinal"},
+                ],
+            ),
+        )
+
+        ctx = RunContext(sub="01", ses="baseline", output_dir=tmp_path)
+        anat_q = ctx.bids(datatype="anat")
+        tpl_q = anat_q.derive(ses="longitudinal")
+
+        resolved = resolve_longitudinal_anat(
+            anat_q, tpl_q, anat_df, tpl_df, ses="baseline"
+        )
+
+        assert resolved["brain_mask"] is None
+        assert resolved["brain"] is not None
+
+    def test_xfm_query_uses_bids_safe_label(self, tmp_path: Path) -> None:
+        """Session labels with non-alphanumerics resolve through ``bids_safe_label``.
+
+        ``pre-op`` is stored as ``from-preop`` in the xfm entity; the resolver
+        must sanitize the ses arg before querying or the xfm won't be found.
+        """
+        anat_df = _df(
+            _anat_row(sub="01", ses="pre-op", suffix="T1w", desc="brain"),
+        )
+        tpl_df = _df(
+            _anat_row(sub="01", ses="longitudinal", suffix="T1w", desc=None),
+            _anat_row(
+                sub="01",
+                ses="longitudinal",
+                suffix="xfm",
+                desc=None,
+                ext=".txt",
+                extra=[
+                    {"key": "from", "value": "preop"},
+                    {"key": "to", "value": "longitudinal"},
+                ],
+            ),
+        )
+
+        ctx = RunContext(sub="01", ses="pre-op", output_dir=tmp_path)
+        anat_q = ctx.bids(datatype="anat")
+        tpl_q = anat_q.derive(ses="longitudinal")
+
+        resolved = resolve_longitudinal_anat(
+            anat_q, tpl_q, anat_df, tpl_df, ses="pre-op"
+        )
+
+        assert resolved["subj_to_template_xfm"] is not None
+        assert str(resolved["subj_to_template_xfm"]).endswith(
+            "sub-01_ses-longitudinal_xfm.txt"
+        )

tests/unit/orchestration/test_longitudinal.py

@@ -17,6 +17,7 @@
 from rbc.context import RunContext
 from rbc.orchestration import Filters
 from rbc.orchestration.longitudinal import process_anat, process_func
+from rbc.workflows.longitudinal.anatomical import AnatomicalLongOutputs
 
 _SCHEMA = [
     "datatype",
@@ -48,14 +49,15 @@ def _func_row(sub: str, ses: str, task: str = "rest") -> tuple:
     return ("func", "bold", ".nii.gz", sub, ses, None, task, None, None, "/data", path)
 
 
-def _mock_anat_outputs() -> Mock:
+def _mock_anat_outputs() -> AnatomicalLongOutputs:
+    """Build a real AnatomicalLongOutputs so typos on missing attrs fail loudly."""
     fake = Path("fake_workdir")
-    m = Mock()
-    m.brain = fake / "brain.nii.gz"
-    m.brain_mask = fake / "brain_mask.nii.gz"
-    m.long_to_template_xfm = fake / "long_to_template_xfm.nii.gz"
-    m.template_to_long_xfm = fake / "template_to_long_xfm.nii.gz"
-    return m
+    return AnatomicalLongOutputs(
+        brain=fake / "brain.nii.gz",
+        brain_mask=fake / "brain_mask.nii.gz",
+        long_to_template_xfm=fake / "long_to_template_xfm.nii.gz",
+        template_to_long_xfm=fake / "template_to_long_xfm.nii.gz",
+    )
 
 
 def _mock_func_outputs(*, with_bold_mask: bool = True) -> Mock:
@@ -299,7 +301,7 @@ def test_missing_required_output_raises(
         pipe_ctx = RunContext(sub="01", ses="baseline", output_dir=tmp_path)
         outputs = _mock_anat_outputs()
         if side_effect is None:
-            setattr(outputs, null_field, None)
+            outputs = outputs._replace(**{null_field: None})  # type: ignore[arg-type]
             get_patch = patch(
                 "rbc.bids.query.find_file",
                 return_value=Path("fake_workdir/file.nii.gz"),