Restructure tests to match architecture layers (#268)

Extract Filters.apply() to consolidate duplicated filtering logic across
all 6 orchestration modules. Move orchestration tests out of cli/ into
their own test directory, rewrite them to call orchestration directly,
and delete redundant tests (runner setup, mock-heavy filtering counts,
dead export assertions).

Author: nx10

src/rbc/orchestration/__init__.py

@@ -11,6 +11,8 @@
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
+import polars as pl
+
 from rbc.core import CPAC_ANTS_SEED
 from rbc.core.niwrap import setup_runner
 
@@ -39,6 +41,28 @@ class Filters:
     session_label: Sequence[str] = field(default_factory=tuple)
     task: str | None = None
 
+    def apply(self, df: pl.DataFrame, *base_exprs: pl.Expr) -> pl.DataFrame:
+        """Apply user-level and workflow-specific filters to a BIDS table.
+
+        Args:
+            df: BIDS table to filter.
+            *base_exprs: Workflow-specific filter expressions
+                (e.g. space, datatype constraints).
+
+        Returns:
+            Filtered DataFrame.
+        """
+        exprs = list(base_exprs)
+        if len(self.participant_label) > 0:
+            exprs.append(pl.col("sub").is_in(self.participant_label))
+        if len(self.session_label) > 0:
+            exprs.append(pl.col("ses").is_in(self.session_label))
+        if self.task is not None:
+            exprs.append(pl.col("task") == self.task)
+        if not exprs:
+            return df
+        return df.filter(pl.all_horizontal(exprs))
+
 
 @dataclass(frozen=True)
 class RunnerConfig:

src/rbc/orchestration/all.py

@@ -69,18 +69,12 @@ def run(
         dataset_dir=input_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
 
-    filter_exprs = [
+    df = filters.apply(
+        df,
         pl.col("ses") != "longitudinal",
         pl.col("space").is_null(),
         pl.col("desc").is_null(),
-    ]
-    if len(filters.participant_label) > 0:
-        filter_exprs.append(pl.col("sub").is_in(filters.participant_label))
-    if len(filters.session_label) > 0:
-        filter_exprs.append(pl.col("ses").is_in(filters.session_label))
-    if filters.task is not None:
-        filter_exprs.append(pl.col("task") == filters.task)
-    df = df.filter(pl.all_horizontal(filter_exprs))
+    )
 
     for _, sub_ses_group in tqdm(
         df.group_by(SUB_SES_QUERY, maintain_order=True), disable=not verbose

src/rbc/orchestration/anatomical.py

@@ -73,16 +73,12 @@ def run(
         dataset_dir=input_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
 
-    filter_exprs = [
+    df = filters.apply(
+        df,
         pl.col("ses") != "longitudinal",
         pl.col("space").is_null(),
         pl.col("desc").is_null(),
-    ]
-    if len(filters.participant_label) > 0:
-        filter_exprs.append(pl.col("sub").is_in(filters.participant_label))
-    if len(filters.session_label) > 0:
-        filter_exprs.append(pl.col("ses").is_in(filters.session_label))
-    df = df.filter(pl.all_horizontal(filter_exprs))
+    )
 
     for _, sub_ses_group in tqdm(
         df.group_by(SUB_SES_QUERY, maintain_order=True), disable=not verbose

src/rbc/orchestration/functional.py

@@ -106,17 +106,11 @@ def run(
         dataset_dir=input_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
 
-    filter_exprs = [
+    df = filters.apply(
+        df,
         pl.col("ses") != "longitudinal",
         pl.col("space").is_null(),
-    ]
-    if len(filters.participant_label) > 0:
-        filter_exprs.append(pl.col("sub").is_in(filters.participant_label))
-    if len(filters.session_label) > 0:
-        filter_exprs.append(pl.col("ses").is_in(filters.session_label))
-    if filters.task is not None:
-        filter_exprs.append(pl.col("task") == filters.task)
-    df = df.filter(pl.all_horizontal(filter_exprs))
+    )
 
     for _, sub_ses_group in tqdm(
         df.group_by(SUB_SES_QUERY, maintain_order=True), disable=not verbose

src/rbc/orchestration/longitudinal.py

@@ -126,12 +126,7 @@ def run(
         dataset_dir=input_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
 
-    filter_exprs = [pl.col("ses") != "longitudinal"]
-    if len(filters.participant_label) > 0:
-        filter_exprs.append(pl.col("sub").is_in(filters.participant_label))
-    if len(filters.session_label) > 0:
-        filter_exprs.append(pl.col("ses").is_in(filters.session_label))
-    group_df = df.filter(pl.all_horizontal(filter_exprs))
+    group_df = filters.apply(df, pl.col("ses") != "longitudinal")
 
     for _, sub_ses_group in tqdm(
         group_df.group_by(SUB_SES_QUERY, maintain_order=True),

src/rbc/orchestration/metrics.py

@@ -108,19 +108,13 @@ def run(
         dataset_dir=output_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
 
-    filter_exprs = [
+    df = filters.apply(
+        df,
         pl.col("datatype") == "func",
         pl.col("suffix") == "bold",
         pl.col("desc") == "preproc",
         pl.col("space") == TemplateSpace.MNI152NLIN6ASYM,
-    ]
-    if len(filters.participant_label) > 0:
-        filter_exprs.append(pl.col("sub").is_in(filters.participant_label))
-    if len(filters.session_label) > 0:
-        filter_exprs.append(pl.col("ses").is_in(filters.session_label))
-    if filters.task is not None:
-        filter_exprs.append(pl.col("task") == filters.task)
-    df = df.filter(pl.all_horizontal(filter_exprs))
+    )
 
     for _, group in tqdm(df.group_by(SUB_SES_QUERY), disable=not verbose):
         sub: str = group["sub"][0]

src/rbc/orchestration/qc.py

@@ -48,19 +48,13 @@ def run(
         dataset_dir=output_dir, index_fpath=None, max_workers=0, verbose=verbose
     )
 
-    filter_exprs = [
+    df = filters.apply(
+        df,
         pl.col("datatype") == "func",
         pl.col("suffix") == "bold",
         pl.col("desc") == "preproc",
         pl.col("space") == TemplateSpace.MNI152NLIN6ASYM,
-    ]
-    if len(filters.participant_label) > 0:
-        filter_exprs.append(pl.col("sub").is_in(filters.participant_label))
-    if len(filters.session_label) > 0:
-        filter_exprs.append(pl.col("ses").is_in(filters.session_label))
-    if filters.task is not None:
-        filter_exprs.append(pl.col("task") == filters.task)
-    df = df.filter(pl.all_horizontal(filter_exprs))
+    )
 
     for _, group in tqdm(df.group_by(SUB_SES_QUERY), disable=not verbose):
         sub: str = group["sub"][0]