Merge pull request usra-riacs#66 from bernalde/fix/issue-65-coverage-workflows

Improve coverage for workflow modules

Author: bernalde

src/cross_validation.py

@@ -452,8 +452,8 @@ def seq_search_evaluate(eval_test, parameter_names, response_col):
             "TotalBudget": "resource",
             response_col: "response",
             # base :'response',
-            "ConfInt=upper_" + response_col: "response_lower",
-            "ConfInt=lower_" + response_col: "response_upper",
+            "ConfInt=lower_" + response_col: "response_lower",
+            "ConfInt=upper_" + response_col: "response_upper",
         },
         inplace=True,
     )

src/utils_ws.py

@@ -286,7 +286,7 @@ def interpolate_df(
                 if all_datapoints:
                     # Create a dataframe with the interesting reads as index and all the columns
                     dummy_df = pd.DataFrame(
-                        np.NaN, index=interpolate_resource, columns=df_index.columns
+                        np.nan, index=interpolate_resource, columns=df_index.columns
                     )
                     dummy_df.drop(columns=resource_column, inplace=True)
                     # Fill out the values that we have certain

tests/test_bootstrap.py

@@ -37,6 +37,19 @@ def dummy_update_rule(self, df):
     pass
 
 
+class InlinePool:
+    """Small Pool stand-in that executes map calls in process for unit tests."""
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def map(self, func, iterable):
+        return [func(item) for item in iterable]
+
+
 class TestBootstrapParameters:
     """Test class for BootstrapParameters dataclass."""
 
@@ -818,6 +831,47 @@ def test_bootstrap_with_list_input(self):
                 result = Bootstrap([df1, df2], ['group'], [params])
 
         assert isinstance(result, pd.DataFrame)
+
+    def test_bootstrap_with_list_of_pickle_paths(self):
+        """Test Bootstrap function with a list of serialized DataFrames."""
+        with tempfile.NamedTemporaryFile(suffix='.pkl', delete=False) as tmp_file1:
+            df1 = pd.DataFrame({
+                'energy': [100, 80],
+                'time': [10, 15],
+                'group': ['A', 'A']
+            })
+            df1.to_pickle(tmp_file1.name)
+
+        with tempfile.NamedTemporaryFile(suffix='.pkl', delete=False) as tmp_file2:
+            df2 = pd.DataFrame({
+                'energy': [120, 90],
+                'time': [8, 12],
+                'group': ['B', 'B']
+            })
+            df2.to_pickle(tmp_file2.name)
+
+        shared_args = {'response_col': 'energy', 'resource_col': 'time'}
+        params = BootstrapParameters(shared_args=shared_args, update_rule=dummy_update_rule, downsample=1)
+
+        try:
+            with patch('bootstrap.Pool') as mock_pool:
+                mock_result = pd.DataFrame({'result': [1], 'group': ['A'], 'boots': [1]})
+                mock_pool.return_value.__enter__.return_value.map.return_value = [mock_result]
+
+                result = Bootstrap([tmp_file1.name, tmp_file2.name], ['group'], [params])
+        finally:
+            os.unlink(tmp_file1.name)
+            os.unlink(tmp_file2.name)
+
+        assert isinstance(result, pd.DataFrame)
+
+    def test_bootstrap_rejects_unsupported_input_type(self):
+        """Test Bootstrap fails clearly for unsupported input objects."""
+        shared_args = {'response_col': 'energy', 'resource_col': 'time'}
+        params = BootstrapParameters(shared_args=shared_args, update_rule=dummy_update_rule)
+
+        with pytest.raises(TypeError, match="Expected DataFrame"):
+            Bootstrap(42, ['group'], [params])
 
     def test_bootstrap_with_progress_dir(self):
         """Test Bootstrap function with progress directory."""
@@ -843,6 +897,69 @@ def test_bootstrap_with_progress_dir(self):
 
             assert isinstance(result, pd.DataFrame)
 
+    def test_bootstrap_executes_grouped_work_in_process(self, monkeypatch):
+        """Test Bootstrap's grouped apply path without multiprocessing."""
+        from pandas.core.groupby.generic import DataFrameGroupBy
+
+        monkeypatch.setattr(
+            DataFrameGroupBy,
+            'progress_apply',
+            DataFrameGroupBy.apply,
+        )
+
+        df = pd.DataFrame({
+            'energy': [100, 80, 120, 90],
+            'time': [10, 15, 8, 12],
+            'group': ['A', 'A', 'B', 'B']
+        })
+
+        shared_args = {'response_col': 'energy', 'resource_col': 'time'}
+        params = BootstrapParameters(
+            shared_args=shared_args,
+            update_rule=dummy_update_rule,
+            downsample=3,
+        )
+
+        def grouped_result(group_df, bs_params):
+            return pd.DataFrame({'rows_seen': [len(group_df)]})
+
+        with patch('bootstrap.Pool', InlinePool), patch(
+            'bootstrap.BootstrapSingle', side_effect=grouped_result
+        ):
+            result = Bootstrap(df, ['group'], [params])
+
+        assert set(result['group']) == {'A', 'B'}
+        assert result['rows_seen'].tolist() == [2, 2]
+        assert result['boots'].eq(3).all()
+
+    def test_bootstrap_uses_progress_file_without_recomputing(self):
+        """Test Bootstrap loads cached progress files before grouped work."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            df = pd.DataFrame({
+                'energy': [100, 80],
+                'time': [10, 15],
+                'group': ['A', 'A']
+            })
+            shared_args = {'response_col': 'energy', 'resource_col': 'time'}
+            params = BootstrapParameters(
+                shared_args=shared_args,
+                update_rule=dummy_update_rule,
+                downsample=4,
+            )
+            cached = pd.DataFrame({'loaded': [True], 'boots': [4]})
+            cached.to_pickle(
+                os.path.join(temp_dir, 'bootstrapped_results_boots=4.pkl')
+            )
+
+            with patch('bootstrap.Pool', InlinePool), patch(
+                'bootstrap.BootstrapSingle'
+            ) as mock_bootstrap_single:
+                result = Bootstrap(df, ['group'], [params], progress_dir=temp_dir)
+
+        mock_bootstrap_single.assert_not_called()
+        assert result['loaded'].tolist() == [True]
+        assert result['boots'].tolist() == [4]
+
 
 class TestBootstrapReduceMem:
     """Test class for Bootstrap_reduce_mem function."""
@@ -977,4 +1094,4 @@ def test_init_bootstrap_single_row(self):
 
 
 if __name__ == "__main__":
-    pytest.main([__file__])
+    pytest.main([__file__])

tests/test_cross_validation.py

@@ -7,6 +7,7 @@
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'src')))
 
 import stats
+import cross_validation as cv
 
 from cross_validation import (
     baseline_evaluate,
@@ -17,6 +18,19 @@
 )
 
 
+@pytest.fixture(autouse=True)
+def clear_cross_validation_globals():
+    cv.parameters_dict.clear()
+    cv.performance_dict.clear()
+    cv.parameters_summarized_dict.clear()
+    cv.performance_summarized_dict.clear()
+    yield
+    cv.parameters_dict.clear()
+    cv.performance_dict.clear()
+    cv.parameters_summarized_dict.clear()
+    cv.performance_summarized_dict.clear()
+
+
 class TestBaselineEvaluate:
     def test_basic(self):
         df = pd.DataFrame({
@@ -102,5 +116,104 @@ def test_seq_search_evaluate(self):
         params_df, eval_df = seq_search_evaluate(df, ['param1'], 'Resp')
         assert 'resource' in params_df.columns
         assert len(eval_df) == 2
+        assert eval_df.loc[0, "response_lower"] == 10
+        assert eval_df.loc[0, "response_upper"] == 12
+
+
+class TestCrossValidationFileWorkflows:
+    def test_load_parameters_concatenates_splits_and_warns_for_missing_files(self, tmp_path):
+        folders = [tmp_path / "split0", tmp_path / "split1"]
+        for folder in folders:
+            (folder / "params_plotting").mkdir(parents=True)
+
+        pd.DataFrame({
+            "Unnamed: 0": [0],
+            "resource": [1],
+            "param1": [0.1],
+        }).to_csv(folders[0] / "params_plotting" / "baseline.csv", index=False)
+        pd.DataFrame({
+            "resource": [1],
+            "param1": [0.2],
+        }).to_csv(folders[1] / "params_plotting" / "baseline.csv", index=False)
+        pd.DataFrame({
+            "resource": [1],
+            "param1": [0.4],
+        }).to_csv(folders[0] / "params_plotting" / "experiment.csv", index=False)
+
+        with pytest.warns(UserWarning, match="experiment.csv not found"):
+            cv.load_parameters([str(folder) for folder in folders], ["experiment"])
+
+        baseline = cv.parameters_dict["baseline"]
+        experiment = cv.parameters_dict["experiment"]
+        assert "Unnamed: 0" not in baseline.columns
+        assert baseline["split_ind"].tolist() == [0, 1]
+        assert experiment["split_ind"].tolist() == [0]
+
+    def test_process_params_across_splits_summarizes_each_experiment(self):
+        cv.parameters_dict["baseline"] = pd.DataFrame({
+            "resource": [1, 1, 2, 2],
+            "param1": [0.0, 2.0, 4.0, 6.0],
+        })
+
+        cv.process_params_across_splits(["param1"], confidence_level=68)
+
+        summary = cv.parameters_summarized_dict["baseline"]["param1"]
+        assert summary["resource"].tolist() == [1, 2]
+        assert summary.loc[summary["resource"] == 1, "mean"].iloc[0] == pytest.approx(1.0)
+        assert {"CI_l", "CI_u"} <= set(summary.columns)
+
+    def test_load_and_process_performance_across_splits(self, tmp_path):
+        folders = [tmp_path / "split0", tmp_path / "split1"]
+        for idx, folder in enumerate(folders):
+            (folder / "performance_plotting").mkdir(parents=True)
+            pd.DataFrame({
+                "Unnamed: 0": [0],
+                "resource": [1],
+                "response": [0.5 + idx],
+            }).to_csv(folder / "performance_plotting" / "baseline.csv", index=False)
+            pd.DataFrame({
+                "resource": [1],
+                "response": [0.7 + idx],
+                "response_lower": [0.6 + idx],
+                "response_upper": [0.8 + idx],
+            }).to_csv(folder / "performance_plotting" / "experiment.csv", index=False)
+
+        cv.load_performance(
+            [str(folder) for folder in folders],
+            ["experiment"],
+            interpolate_flag=False,
+        )
+        assert "Unnamed: 0" not in cv.performance_dict["baseline"].columns
+
+        cv.process_performance_across_splits(stats_measure="mean")
+
+        baseline = cv.performance_summarized_dict["baseline"]
+        experiment = cv.performance_summarized_dict["experiment"]
+        assert baseline.loc[0, "mean"] == pytest.approx(1.0)
+        assert {"mean", "CI_l", "CI_u"} <= set(experiment.columns)
+
+    def test_create_eval_params_dfs_dispatches_projection_evaluator(self, tmp_path):
+        response_col = "Resp"
+        folders = [tmp_path / "split0", tmp_path / "split1"]
+        for idx, folder in enumerate(folders):
+            folder.mkdir()
+            pd.DataFrame({
+                "resource": [1],
+                "param1": [idx + 0.5],
+                response_col: [idx + 1.0],
+                "ConfInt=lower_" + response_col: [idx + 0.8],
+                "ConfInt=upper_" + response_col: [idx + 1.2],
+            }).to_pickle(folder / "projection.pkl")
+
+        raw, params, perf = cv.create_eval_params_dfs(
+            "projection.pkl",
+            [str(folder) for folder in folders],
+            ["param1"],
+            "proj_expt_evaluate",
+            response_col,
+        )
 
+        assert raw["split"].tolist() == [1, 2]
+        assert params["split"].tolist() == [1, 2]
+        assert perf["response_lower"].tolist() == [0.8, 1.8]