add tests to fix coverage

Author: tien-tong

tests/e2e_in_docker.sh

@@ -11,5 +11,5 @@ docker run -it \
         --cov-report=xml \
         --cov=babs \
         --pdb \
-        /babs/tests/test_update_input_data.py
+        /babs/tests/
 

tests/pytest_in_docker.sh

@@ -1,13 +1,16 @@
 """Test the check_setup functionality."""
 
+import os.path as op
 import random
 from pathlib import Path
+from unittest.mock import MagicMock
 
+import pandas as pd
 import pytest
 import yaml
 
 from babs import BABSCheckSetup
-from babs.base import CONFIG_SECTIONS
+from babs.base import BABS, CONFIG_SECTIONS
 from babs.utils import read_yaml
 
 
@@ -50,8 +53,6 @@ def test_missing_directories(tmp_path_factory):
 
 def test_validate_pipeline_config(babs_project_sessionlevel):
     """Test _validate_pipeline_config method."""
-    from babs.base import BABS
-
     babs_proj = BABS(babs_project_sessionlevel)
 
     # Test valid config
@@ -74,3 +75,130 @@ def test_validate_pipeline_config(babs_project_sessionlevel):
     babs_proj.pipeline = [{'missing': 'container_name'}]
     with pytest.raises(ValueError, match='Pipeline step 0 missing required field: container_name'):
         babs_proj._validate_pipeline_config()
+
+
+def test_project_root_not_exists(tmp_path):
+    """Test FileNotFoundError when project_root doesn't exist."""
+    non_existent_path = tmp_path / 'does_not_exist'
+    with pytest.raises(FileNotFoundError, match='`project_root` does not exist!'):
+        BABS(non_existent_path)
+
+
+def test_analysis_path_not_exists(tmp_path):
+    """Test FileNotFoundError when analysis path doesn't exist."""
+    project_root = tmp_path / 'project'
+    project_root.mkdir()
+    with pytest.raises(FileNotFoundError, match='is not a valid BABS project'):
+        BABS(project_root)
+
+
+def test_config_path_not_exists(babs_project_sessionlevel):
+    """Test FileNotFoundError when config path doesn't exist."""
+    babs_proj = BABSCheckSetup(babs_project_sessionlevel)
+    Path(babs_proj.config_path).unlink()
+
+    with pytest.raises(FileNotFoundError, match='is not a valid BABS project'):
+        BABS(babs_project_sessionlevel)
+
+
+def test_pipeline_config_details(babs_project_sessionlevel):
+    """Test pipeline validation with config details."""
+    babs_proj = BABS(babs_project_sessionlevel)
+
+    # Test with cluster_resources, bids_app_args, singularity_args
+    babs_proj.pipeline = [
+        {
+            'container_name': 'test-app',
+            'config': {
+                'cluster_resources': {'memory': '8GB', 'cpus': 4},
+                'bids_app_args': {'--nthreads': 4},
+                'singularity_args': ['--bind', '/tmp'],
+            },
+        }
+    ]
+    babs_proj._validate_pipeline_config()
+
+    # Test with inter_step_cmds
+    babs_proj.pipeline = [{'container_name': 'test-app', 'inter_step_cmds': ['echo "test"']}]
+    babs_proj._validate_pipeline_config()
+
+    # Test with both
+    babs_proj.pipeline = [
+        {
+            'container_name': 'test-app',
+            'config': {'cluster_resources': {'memory': '8GB'}},
+            'inter_step_cmds': ['echo "test"'],
+        }
+    ]
+    babs_proj._validate_pipeline_config()
+
+
+def test_update_inclusion_empty_combine(babs_project_sessionlevel):
+    """Test _update_inclusion_dataframe when combined dataframe is empty."""
+    babs_proj = BABS(babs_project_sessionlevel)
+    initial_inclusion_df = pd.DataFrame({'sub_id': ['sub-9999'], 'ses_id': ['ses-9999']})
+
+    with pytest.raises(ValueError, match='No subjects/sessions to analyze!'):
+        babs_proj._update_inclusion_dataframe(initial_inclusion_df=initial_inclusion_df)
+
+
+def test_update_inclusion_warning(babs_project_sessionlevel, capsys):
+    """Test _update_inclusion_dataframe warning when initial df has more subjects."""
+    babs_proj = BABS(babs_project_sessionlevel)
+    actual_df = babs_proj.input_datasets.generate_inclusion_dataframe()
+
+    if 'ses_id' in actual_df.columns:
+        initial_inclusion_df = pd.DataFrame(
+            {
+                'sub_id': ['sub-0001', 'sub-0002', 'sub-9999'],
+                'ses_id': ['ses-01', 'ses-01', 'ses-01'],
+            }
+        )
+    else:
+        initial_inclusion_df = pd.DataFrame({'sub_id': ['sub-0001', 'sub-0002', 'sub-9999']})
+
+    babs_proj._update_inclusion_dataframe(initial_inclusion_df=initial_inclusion_df)
+    captured = capsys.readouterr()
+    assert 'Warning: The initial inclusion dataframe' in captured.out
+
+
+def test_datalad_save_filter_files(babs_project_sessionlevel):
+    """Test datalad_save with filter_files parameter."""
+    babs_proj = BABS(babs_project_sessionlevel)
+    test_file = op.join(babs_proj.analysis_path, 'code', 'test_file.txt')
+    Path(test_file).parent.mkdir(parents=True, exist_ok=True)
+    Path(test_file).write_text('test content')
+
+    babs_proj.datalad_save(
+        path=test_file, message='Test save with filter', filter_files=['test_file.txt']
+    )
+    assert Path(test_file).exists()
+
+
+def test_datalad_save_failure(babs_project_sessionlevel, monkeypatch):
+    """Test datalad_save when save fails."""
+    babs_proj = BABS(babs_project_sessionlevel)
+    mock_save = MagicMock(return_value=[{'status': 'error', 'message': 'Save failed'}])
+    monkeypatch.setattr(babs_proj.analysis_datalad_handle, 'save', mock_save)
+
+    test_file = op.join(babs_proj.analysis_path, 'code', 'test_file.txt')
+    Path(test_file).parent.mkdir(parents=True, exist_ok=True)
+    Path(test_file).write_text('test content')
+
+    with pytest.raises(Exception, match='`datalad save` failed!'):
+        babs_proj.datalad_save(path=test_file, message='Test save')
+
+
+def test_key_info_ria_only(babs_project_sessionlevel):
+    """Test wtf_key_info with flag_output_ria_only=True."""
+    babs_proj = BABS(babs_project_sessionlevel)
+    babs_proj.wtf_key_info(flag_output_ria_only=True)
+    assert babs_proj.output_ria_data_dir is not None
+
+
+def test_key_info_full(babs_project_sessionlevel):
+    """Test wtf_key_info with flag_output_ria_only=False."""
+    babs_proj = BABS(babs_project_sessionlevel)
+    babs_proj.wtf_key_info(flag_output_ria_only=False)
+    assert babs_proj.output_ria_data_dir is not None
+    assert babs_proj.analysis_dataset_id is not None

tests/test_base.py

@@ -0,0 +1,77 @@
+"""Test merge.py error handling and edge cases."""
+
+import subprocess
+from unittest.mock import MagicMock
+
+import pytest
+
+from babs.merge import BABSMerge
+from babs.utils import get_git_show_ref_shasum
+
+
+def test_merge_no_branches(babs_project_sessionlevel, monkeypatch):
+    """Test babs_merge when no branches have results."""
+    babs_proj = BABSMerge(babs_project_sessionlevel)
+    monkeypatch.setattr(babs_proj, '_get_results_branches', lambda: [])
+
+    with pytest.raises(ValueError, match='There is no successfully finished job yet'):
+        babs_proj.babs_merge()
+
+
+def test_merge_all_branches_no_results(babs_project_sessionlevel, tmp_path, monkeypatch):
+    """Test babs_merge when all branches have no results."""
+    babs_proj = BABSMerge(babs_project_sessionlevel)
+
+    merge_ds_path = tmp_path / 'merge_ds'
+    merge_ds_path.mkdir()
+    subprocess.run(['git', 'init'], cwd=merge_ds_path, capture_output=True)
+    subprocess.run(['git', 'config', 'user.name', 'Test'], cwd=merge_ds_path, capture_output=True)
+    subprocess.run(
+        ['git', 'config', 'user.email', 'test@test.com'],
+        cwd=merge_ds_path,
+        capture_output=True,
+    )
+    (merge_ds_path / 'test.txt').write_text('test')
+    subprocess.run(['git', 'add', 'test.txt'], cwd=merge_ds_path, capture_output=True)
+    subprocess.run(['git', 'commit', '-m', 'Initial'], cwd=merge_ds_path, capture_output=True)
+
+    default_branch = 'main'
+    try:
+        subprocess.run(
+            ['git', 'checkout', '-b', default_branch],
+            cwd=merge_ds_path,
+            capture_output=True,
+        )
+    except Exception:
+        default_branch = 'master'
+
+    git_ref, _ = get_git_show_ref_shasum(default_branch, merge_ds_path)
+
+    def mock_branches():
+        return ['job-123-1-sub-0001']
+
+    def mock_key_info(flag_output_ria_only=False):
+        babs_proj.analysis_dataset_id = 'test-id'
+
+    def mock_git_ref(branch, path):
+        return git_ref, f'{git_ref} refs/remotes/origin/{branch}'
+
+    monkeypatch.setattr(babs_proj, '_get_results_branches', mock_branches)
+    monkeypatch.setattr(babs_proj, 'wtf_key_info', mock_key_info)
+    monkeypatch.setattr('babs.merge.get_git_show_ref_shasum', mock_git_ref)
+    from babs.merge import dlapi
+
+    monkeypatch.setattr(dlapi, 'clone', lambda source, path: None)
+
+    def mock_remote_show(cmd, **kwargs):
+        if 'remote' in cmd and 'show' in cmd:
+            result = MagicMock()
+            result.returncode = 0
+            result.stdout = f'HEAD branch: {default_branch}\n'.encode()
+            return result
+        return subprocess.run(cmd, **kwargs)
+
+    monkeypatch.setattr('babs.merge.subprocess.run', mock_remote_show)
+
+    with pytest.raises(Exception, match='There is no job branch in output RIA that has results'):
+        babs_proj.babs_merge()