adding tests

Author: birajstha

CPAC/utils/tests/test_datasource.py

@@ -20,17 +20,23 @@
 import json
 from pathlib import Path
 from typing import Any, Literal, TypeAlias
+from unittest.mock import mock_open, patch
 
 from networkx.classes.digraph import DiGraph
 import pytest
 
 from CPAC.pipeline import nipype_pipeline_engine as pe
-from CPAC.utils.test_resources import setup_test_wf
-from CPAC.utils.utils import PE_DIRECTION, get_fmap_type
 from CPAC.utils.datasource import (
     match_epi_fmaps,
     match_epi_fmaps_function_node,
 )
+from CPAC.utils.test_resources import setup_test_wf
+from CPAC.utils.utils import (
+    get_fmap_build_info,
+    get_fmap_metadata_at_build_time,
+    get_fmap_type,
+    PE_DIRECTION,
+)
 
 
 @dataclass
@@ -484,3 +490,192 @@ def test_get_fmap_type_real_world_examples() -> None:
     # Real-world phase example (only required fields)
     phase_metadata = {"EchoTime": 0.00746}
     assert get_fmap_type(phase_metadata) == "phase"
+
+
+class TestGetFmapMetadataAtBuildTime:
+    """Test get_fmap_metadata_at_build_time function."""
+
+    def test_missing_fmap_key(self):
+        """Test when fieldmap key doesn't exist in sub_dict."""
+        sub_dict = {"fmap": {"other_key": {}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "missing_key", "", "")
+        assert result is None
+
+    def test_missing_scan_parameters(self):
+        """Test when scan_parameters field is missing."""
+        sub_dict = {"fmap": {"test_key": {"scan": "path/to/scan.nii.gz"}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result is None
+
+    def test_direct_dict_metadata(self):
+        """Test when metadata is provided as a direct dictionary."""
+        metadata = {"EchoTime1": 0.006, "EchoTime2": 0.007}
+        sub_dict = {"fmap": {"test_key": {"scan_parameters": metadata}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result == metadata
+
+    @patch("builtins.open", new_callable=mock_open, read_data='{"EchoTime": 0.006}')
+    @patch("os.path.exists", return_value=True)
+    def test_json_file_metadata(self, mock_exists, mock_file):
+        """Test loading metadata from JSON file."""
+        sub_dict = {"fmap": {"test_key": {"scan_parameters": "/path/to/metadata.json"}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result == {"EchoTime": 0.006}
+        mock_file.assert_called_once_with(
+            "/path/to/metadata.json", "r", encoding="utf-8"
+        )
+
+    @patch("os.path.exists", return_value=False)
+    def test_nonexistent_file(self, mock_exists):
+        """Test when JSON file doesn't exist."""
+        sub_dict = {"fmap": {"test_key": {"scan_parameters": "/nonexistent/file.json"}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result is None
+
+    @patch("builtins.open", side_effect=json.JSONDecodeError("Invalid JSON", "", 0))
+    @patch("os.path.exists", return_value=True)
+    def test_invalid_json(self, mock_exists, mock_file):
+        """Test when JSON file contains invalid JSON."""
+        sub_dict = {"fmap": {"test_key": {"scan_parameters": "/path/to/invalid.json"}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result is None
+
+    def test_non_json_file(self):
+        """Test when file path doesn't end with .json."""
+        sub_dict = {"fmap": {"test_key": {"scan_parameters": "/path/to/file.txt"}}}
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result is None
+
+    def test_exception_handling(self):
+        """Test general exception handling."""
+        sub_dict = {"fmap": {"test_key": {"scan_parameters": 123}}}  # Invalid type
+        result = get_fmap_metadata_at_build_time(sub_dict, "test_key", "", "")
+        assert result is None
+
+
+class TestGetFmapBuildInfo:
+    """Test get_fmap_build_info function."""
+
+    def test_none_metadata_raises_error(self):
+        """Test that None metadata raises ValueError."""
+        with pytest.raises(
+            ValueError, match="Fieldmap metadata dictionary is required"
+        ):
+            get_fmap_build_info(None)
+
+    def test_empty_metadata_raises_error(self):
+        """Test that empty metadata raises ValueError."""
+        with pytest.raises(
+            ValueError, match="Fieldmap metadata dictionary is required"
+        ):
+            get_fmap_build_info({})
+
+    def test_unknown_fmap_type_raises_error(self):
+        """Test that unknown fieldmap type raises ValueError."""
+        metadata = {"SomeUnknownField": "value"}
+        with pytest.raises(ValueError, match="Could not determine fieldmap type"):
+            get_fmap_build_info(metadata)
+
+    def test_phase_fieldmap_info(self):
+        """Test phase fieldmap build info."""
+        metadata = {"EchoTime": 0.006}
+        result = get_fmap_build_info(metadata)
+        expected = {
+            "fmap_type": "phase",
+            "needs_echo_times": True,
+            "needs_phasediff_processing": True,
+            "is_epi": False,
+        }
+        assert result == expected
+
+    def test_phasediff_fieldmap_info(self):
+        """Test phasediff fieldmap build info."""
+        metadata = {"EchoTime1": 0.006, "EchoTime2": 0.007}
+        result = get_fmap_build_info(metadata)
+        expected = {
+            "fmap_type": "phasediff",
+            "needs_echo_times": True,
+            "needs_phasediff_processing": True,
+            "is_epi": False,
+        }
+        assert result == expected
+
+    def test_epi_fieldmap_info(self):
+        """Test EPI fieldmap build info."""
+        metadata = {"PhaseEncodingDirection": "j-"}
+        result = get_fmap_build_info(metadata)
+        expected = {
+            "fmap_type": "epi",
+            "needs_echo_times": True,
+            "needs_phasediff_processing": False,
+            "is_epi": True,
+        }
+        assert result == expected
+
+    @pytest.mark.parametrize(
+        "metadata,expected_fmap_type",
+        [
+            ({"EchoTime": 0.006}, "phase"),
+            ({"EchoTime1": 0.006, "EchoTime2": 0.007}, "phasediff"),
+            ({"PhaseEncodingDirection": "j-"}, "epi"),
+        ],
+    )
+    def test_various_fieldmap_types(self, metadata, expected_fmap_type):
+        """Test that various fieldmap types are correctly identified."""
+        result = get_fmap_build_info(metadata)
+        assert result["fmap_type"] == expected_fmap_type
+
+    def test_real_world_metadata_examples(self):
+        """Test with realistic metadata examples from the existing tests."""
+        # Use some of the test data from the existing test_get_fmap_type tests
+
+        # Phasediff example
+        phasediff_metadata = {
+            "EchoTime1": 0.00600,
+            "EchoTime2": 0.00746,
+            "IntendedFor": ["bids::sub-01/func/sub-01_task-motor_bold.nii.gz"],
+        }
+        result = get_fmap_build_info(phasediff_metadata)
+        assert result["fmap_type"] == "phasediff"
+        assert result["needs_echo_times"] is True
+        assert result["needs_phasediff_processing"] is True
+        assert result["is_epi"] is False
+
+        # EPI example
+        epi_metadata = {
+            "PhaseEncodingDirection": "j-",
+            "TotalReadoutTime": 0.095,
+            "IntendedFor": "bids::sub-01/func/sub-01_task-motor_bold.nii.gz",
+        }
+        result = get_fmap_build_info(epi_metadata)
+        assert result["fmap_type"] == "epi"
+        assert result["needs_echo_times"] is True
+        assert result["needs_phasediff_processing"] is False
+        assert result["is_epi"] is True
+
+    def test_phase_fieldmap_with_extra_fields(self):
+        """Test phase fieldmap with additional optional fields."""
+        metadata = {
+            "EchoTime": 0.006,
+            "IntendedFor": "bids::sub-01/func/sub-01_task-motor_bold.nii.gz",
+            "B0FieldIdentifier": "my_fieldmap",
+        }
+        result = get_fmap_build_info(metadata)
+        assert result["fmap_type"] == "phase"
+        assert result["needs_echo_times"] is True
+        assert result["needs_phasediff_processing"] is True
+        assert result["is_epi"] is False
+
+    def test_phasediff_fieldmap_with_extra_fields(self):
+        """Test phasediff fieldmap with additional optional fields."""
+        metadata = {
+            "EchoTime1": 0.006,
+            "EchoTime2": 0.007,
+            "IntendedFor": ["bids::sub-01/func/sub-01_task-motor_bold.nii.gz"],
+            "B0FieldIdentifier": "my_phasediff",
+        }
+        result = get_fmap_build_info(metadata)
+        assert result["fmap_type"] == "phasediff"
+        assert result["needs_echo_times"] is True
+        assert result["needs_phasediff_processing"] is True
+        assert result["is_epi"] is False

CPAC/utils/utils.py

@@ -2796,25 +2796,35 @@ def get_fmap_build_info(metadata_dict):
         - 'needs_echo_times': bool
         - 'needs_phasediff_processing': bool
         - 'is_epi': bool
+
+    Raises
+    ------
+    ValueError
+        If metadata_dict is None or if fieldmap type cannot be determined
     """
     from CPAC.utils.utils import get_fmap_type
 
+    if not metadata_dict:
+        raise ValueError(
+            "Fieldmap metadata dictionary is required but was None. "
+            "Cannot determine fieldmap processing requirements without metadata."
+        )
+
+    fmap_type = get_fmap_type(metadata_dict)
+
+    if fmap_type is None:
+        raise ValueError(
+            f"Could not determine fieldmap type from metadata: {metadata_dict}. "
+            "Metadata must contain required BIDS fields for fieldmap type detection."
+        )
+
     build_info = {
-        "fmap_type": None,
+        "fmap_type": fmap_type,
         "needs_echo_times": False,
         "needs_phasediff_processing": False,
         "is_epi": False,
     }
 
-    if not metadata_dict:
-        # Conservative fallback - assume we might need processing
-        build_info["needs_echo_times"] = True
-        build_info["needs_phasediff_processing"] = True
-        return build_info
-
-    fmap_type = get_fmap_type(metadata_dict)
-    build_info["fmap_type"] = fmap_type
-
     match fmap_type:
         case "phase":
             build_info["needs_echo_times"] = True
@@ -2831,9 +2841,10 @@ def get_fmap_build_info(metadata_dict):
         case "fieldmap":
             build_info["needs_phasediff_processing"] = True
 
-        case None:
-            # Conservative fallback
-            build_info["needs_echo_times"] = True
-            build_info["needs_phasediff_processing"] = True
+        case _:
+            raise ValueError(
+                f"Unsupported fieldmap type '{fmap_type}'. "
+                "Supported types are: 'phase', 'phasediff', 'epi', 'fieldmap'."
+            )
 
     return build_info