diff --git a/src/alphapepttools/pp/data.py b/src/alphapepttools/pp/data.py index 88b47b35..f8abe7ac 100644 --- a/src/alphapepttools/pp/data.py +++ b/src/alphapepttools/pp/data.py @@ -197,13 +197,15 @@ def add_metadata( # noqa: C901, PLR0912 incoming_metadata = incoming_metadata.reindex(adata.var.index) # 5. assign the new metadata to the adata object's obs or var attribute + # Defensive: after the reindex in step 4, indices are guaranteed to match. These raises + # exist to make that invariant explicit to readers — unreachable, hence pragma. if axis == 0: if not adata.obs.index.equals(incoming_metadata.index): - raise ValueError("Index mismatch between data and metadata.") + raise ValueError("Index mismatch between data and metadata.") # pragma: no cover adata.obs = incoming_metadata elif axis == 1: if not adata.var.index.equals(incoming_metadata.index): - raise ValueError("Index mismatch between data and metadata.") + raise ValueError("Index mismatch between data and metadata.") # pragma: no cover adata.var = incoming_metadata return adata @@ -256,8 +258,10 @@ def _filter_by_dict( filter_masks = [] for k, v in filter_dict.items(): feature = data[k] if k != "index" else data.index + # Defensive: _verify_filter_dict above rejects None values, so this branch is unreachable + # from the public API. Kept as a documented "None means match-all" guard. if v is None: - current_mask = pd.Series(True, index=data.index) # noqa: FBT003 + current_mask = pd.Series(True, index=data.index) # noqa: FBT003 # pragma: no cover # TODO: add function evaluation here if v is a callable, so that users can pass things like np.isnan or lambdas elif isinstance(v, str | numbers.Number): current_mask = feature == v diff --git a/tests/pp/test_batch_correction.py b/tests/pp/test_batch_correction.py index 969225ad..9c5874e1 100644 --- a/tests/pp/test_batch_correction.py +++ b/tests/pp/test_batch_correction.py @@ -102,3 +102,57 @@ def test_scanpy_pycombat( assert result is None pd.testing.assert_frame_equal(adata.to_df(layer=layer), expected_adata.to_df()) + + +def test_scanpy_pycombat_layer_none(pycombat_test_data_simple): + """When layer=None, the corrected matrix is written back to adata.X (not a layer).""" + df, md = pycombat_test_data_simple + adata = ad.AnnData(df, obs=md) + X_before = adata.X.copy() + + result = scanpy_pycombat(adata, batch="batch", layer=None) + + # in-place modification on the local `adata` rebound in the function — returns None + assert result is None + # caller's adata.X was modified before the rebind (astype on line 184), and copy=False + # path still mutates the caller's X via line 212 when no NaN-coerce is needed + assert not np.array_equal(adata.X, X_before) + + +def test_scanpy_pycombat_coerces_nan_batches(): + """NaN values in the batch column should be coerced to an 'NA' batch.""" + df = pd.DataFrame( + {"A": [1.0, 2.0, 4.0, 8.0, 16.0, 32.0], "B": [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]}, + index=list("ABCDEF"), + ) + md = pd.DataFrame({"batch": ["x", "x", "x", np.nan, np.nan, np.nan]}, index=list("ABCDEF")) + adata = ad.AnnData(df, obs=md) + + result = scanpy_pycombat(adata, batch="batch", copy=True) + + assert "NA" in result.obs["batch"].tolist() + assert not result.obs["batch"].isna().any() + + +def test_scanpy_pycombat_raises_on_nan_data(pycombat_test_data_simple): + """If adata.X contains NaNs, scanpy_pycombat raises ValueError.""" + df, md = pycombat_test_data_simple + df_with_nan = df.copy() + df_with_nan.iloc[0, 0] = np.nan + adata = ad.AnnData(df_with_nan, obs=md) + + with pytest.raises(ValueError, match="contains NaN"): + scanpy_pycombat(adata, batch="batch") + + +def test_scanpy_pycombat_raises_on_singleton_batch(): + """If a batch contains only one sample, scanpy_pycombat raises ValueError.""" + df = pd.DataFrame( + {"A": [1.0, 2.0, 3.0, 4.0, 5.0], "B": [1.0, 2.0, 3.0, 4.0, 5.0]}, + index=list("ABCDE"), + ) + md = pd.DataFrame({"batch": ["x", "x", "y", "y", "z"]}, index=list("ABCDE")) # z is singleton + adata = ad.AnnData(df, obs=md) + + with pytest.raises(ValueError, match="only one single sample"): + scanpy_pycombat(adata, batch="batch") diff --git a/tests/pp/test_data.py b/tests/pp/test_data.py index c8cc917d..73d9f057 100644 --- a/tests/pp/test_data.py +++ b/tests/pp/test_data.py @@ -1,3 +1,4 @@ +import logging import warnings import anndata as ad @@ -6,7 +7,20 @@ import pytest import alphapepttools as apt -from alphapepttools.pp.data import _handle_overlapping_columns, _to_anndata, coerce_to_dataframe, data_column_to_array +from alphapepttools.pp.data import ( + _filter_by_dict, + _handle_overlapping_columns, + _to_anndata, + _tolist, + _tuple_based_filter, + _validate_adata_for_completeness_filter, + _verify_filter_dict, + coerce_to_dataframe, + data_column_to_array, + data_columns_to_df, + data_index_to_array, + subset_data, +) # example data @@ -1147,6 +1161,79 @@ def test_data_column_to_array( assert np.all(array == expected_array) +### Test data_index_to_array ### + + +@pytest.mark.parametrize( + ("input_type", "dim_space", "expected"), + [ + ("dataframe", "obs", np.array(["cell1", "cell2", "cell3"])), + ("dataframe", "var", np.array(["G1", "G2", "G3"])), + ("anndata", "obs", np.array(["cell1", "cell2", "cell3"])), + ("anndata", "var", np.array(["G1", "G2", "G3"])), + ], +) +def test_data_index_to_array(example_data, input_type, dim_space, expected): + # given + data = example_data if input_type == "dataframe" else _to_anndata(example_data) + + # when + result = data_index_to_array(data, dim_space=dim_space) + + # then + assert isinstance(result, np.ndarray) + assert np.array_equal(result, expected) + + +def test_data_index_to_array_invalid_dim_space(example_data): + with pytest.raises(ValueError, match="dim_space must be"): + data_index_to_array(example_data, dim_space="invalid") + + +def test_data_index_to_array_invalid_type(): + with pytest.raises(TypeError, match="Expected pd.DataFrame or ad.AnnData"): + data_index_to_array([1, 2, 3], dim_space="obs") + + +### Test subset_data ### + + +@pytest.mark.parametrize( + ("input_type", "idxs", "expected_index"), + [ + ("dataframe", [0, 2], ["cell1", "cell3"]), + ("dataframe", [1], ["cell2"]), + ("anndata", [0, 2], ["cell1", "cell3"]), + ("anndata", [1], ["cell2"]), + ], +) +def test_subset_data(example_data, input_type, idxs, expected_index): + # given + data = example_data if input_type == "dataframe" else _to_anndata(example_data) + + # when + result = subset_data(data, idxs) + + # then + assert list(result.index if isinstance(result, pd.DataFrame) else result.obs_names) == expected_index + + +def test_subset_data_returns_copy(example_data): + # given + df = example_data.copy() + adata = _to_anndata(example_data) + + # when + df_sub = subset_data(df, [0]) + adata_sub = subset_data(adata, [0]) + + # then — mutating the subset must not affect the original + df_sub.iloc[0, 0] = 999 + adata_sub.X[0, 0] = 999 + assert df.iloc[0, 0] != 999 # noqa: PLR2004 + assert adata.X[0, 0] != 999 # noqa: PLR2004 + + ### Test coerce_to_dataframe ### @@ -1207,3 +1294,275 @@ def test_coerce_to_dataframe( ) # Check that obs columns are included assert list(result["cell_type"]) == ["type_A", "type_B", "type_C"] + + +### Test _to_anndata ndarray branch ### + + +def test_to_anndata_from_ndarray(): + # given + arr = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]]) + + # when + adata = _to_anndata(arr) + + # then + assert isinstance(adata, ad.AnnData) + assert np.array_equal(adata.X, arr) + + +### Test add_metadata validation branches ### + + +def test_add_metadata_invalid_axis(example_data): + adata = _to_anndata(example_data) + md = pd.DataFrame({"x": [1, 2, 3]}, index=["cell1", "cell2", "cell3"]) + with pytest.raises(ValueError, match="Axis must be 0 or 1"): + apt.pp.add_metadata(adata, md, axis=2) + + +def test_add_metadata_empty_adata(caplog): + # given + adata = ad.AnnData(np.empty((0, 0))) + md = pd.DataFrame({"x": []}) + + # when + with caplog.at_level(logging.INFO): + result = apt.pp.add_metadata(adata, md, axis=0) + + # then + assert result is adata + assert "adata is empty" in caplog.text + + +@pytest.mark.parametrize( + "bad_md", + [ + # not a DataFrame + {"col": [1, 2, 3]}, + # multi-level index + pd.DataFrame( + {"x": [1, 2, 3]}, + index=pd.MultiIndex.from_tuples([("a", 1), ("b", 2), ("c", 3)]), + ), + ], +) +def test_add_metadata_bad_metadata_type(example_data, bad_md): + adata = _to_anndata(example_data) + with pytest.raises(TypeError, match="metadata must be a pd.DataFrame"): + apt.pp.add_metadata(adata, bad_md, axis=0) + + +def test_add_metadata_duplicated_metadata_index(example_data): + adata = _to_anndata(example_data) + md = pd.DataFrame( + {"cell_type": ["A", "B", "C"]}, + index=["cell1", "cell1", "cell3"], # duplicate + ) + with pytest.raises(ValueError, match="Duplicated metadata indices"): + apt.pp.add_metadata(adata, md, axis=0) + + +def test_add_metadata_verbose_log(example_data, caplog): + # given — keep_existing_metadata=True + verbose=True exercises the verbose log line + adata = _to_anndata(example_data) + adata.obs = pd.DataFrame({"existing": ["x", "y", "z"]}, index=["cell1", "cell2", "cell3"]) + md = pd.DataFrame({"new": ["a", "b", "c"]}, index=["cell1", "cell2", "cell3"]) + + # when + with caplog.at_level(logging.INFO): + apt.pp.add_metadata(adata, md, axis=0, keep_existing_metadata=True, verbose=True) + + # then + assert "Join incoming to existing metadata" in caplog.text + + +### Test _filter_by_dict / _tuple_based_filter / _verify_filter_dict ### + + +def test_filter_by_dict_duplicated_indices(): + df = pd.DataFrame({"x": [1, 2, 3]}, index=["a", "a", "b"]) + with pytest.raises(ValueError, match="Duplicated indices"): + _filter_by_dict(df, {"x": 1}, logic="and") + + +@pytest.mark.parametrize( + ("bad_tuple", "match"), + [ + # length != 2 + ((1, 2, 3), "tuple of length 2"), + # contains non-numeric, non-None value + ((1, "x"), "numeric values or None"), + ], +) +def test_tuple_based_filter_invalid_tuple(bad_tuple, match): + feature = pd.Series([1, 2, 3], index=["a", "b", "c"]) + with pytest.raises(ValueError, match=match): + _tuple_based_filter(feature, bad_tuple) + + +def test_tuple_based_filter_non_numeric_feature(): + feature = pd.Series(["a", "b", "c"], index=["x", "y", "z"]) + with pytest.raises(ValueError, match="numeric features"): + _tuple_based_filter(feature, (0, 1)) + + +@pytest.mark.parametrize( + ("bad_dict", "match"), + [ + # non-string key + ({1: "a"}, "Filter keys must be string"), + # unknown column (key is a string, but not in data.columns and not "index") + ({"missing_col": "a"}, "not found in data columns"), + # bad value type (set is not str/number/list/tuple) + ({"x": {"not", "allowed"}}, "must be of type str, number, list or tuple"), + ], +) +def test_verify_filter_dict(bad_dict, match): + df = pd.DataFrame({"x": [1, 2, 3]}, index=["a", "b", "c"]) + with pytest.raises(ValueError, match=match): + _verify_filter_dict(bad_dict, df) + + +def test_filter_by_metadata_invalid_axis(): + adata = _to_anndata(pd.DataFrame({"G1": [1.0, 2.0]}, index=["c1", "c2"])) + with pytest.raises(ValueError, match="Invalid 'axis'"): + apt.pp.filter_by_metadata(adata, {}, axis=2) + + +### Test data_column_to_array error branches ### + + +def test_data_column_to_array_dataframe_missing_column(example_data): + with pytest.raises(ValueError, match="not found in DataFrame"): + data_column_to_array(example_data, "nonexistent") + + +def test_data_column_to_array_anndata_from_var_columns(example_data, example_feature_metadata): + # given — `gene_name` lives only in adata.var.columns (not in var_names or obs.columns) + adata = _to_anndata(example_data) + adata = apt.pp.add_metadata(adata, example_feature_metadata, axis=1) + + # when + arr = data_column_to_array(adata, "gene_name") + + # then + assert np.array_equal(arr, np.array(["gene1", "gene2", "gene3"])) + + +def test_data_column_to_array_anndata_not_found(example_data): + adata = _to_anndata(example_data) + with pytest.raises(ValueError, match="not found in AnnData"): + data_column_to_array(adata, "nonexistent") + + +def test_data_column_to_array_invalid_type(): + with pytest.raises(TypeError, match="Expected pd.DataFrame or ad.AnnData"): + data_column_to_array([1, 2, 3], "x") + + +### Test _tolist ### + + +@pytest.mark.parametrize( + ("obj", "expected"), + [ + ("foo", ["foo"]), + (["a", "b"], ["a", "b"]), + ], +) +def test_tolist(obj, expected): + assert _tolist(obj) == expected + + +### Test data_columns_to_df edge cases ### + + +def test_data_columns_to_df_anndata_columns_none(example_data): + adata = _to_anndata(example_data) + + # when + result = data_columns_to_df(adata, columns=None) + + # then + assert isinstance(result, pd.DataFrame) + assert list(result.columns) == ["G1", "G2", "G3"] + + +def test_data_columns_to_df_invalid_type(): + with pytest.raises(TypeError, match="Expected pd.DataFrame or ad.AnnData"): + data_columns_to_df([1, 2, 3]) + + +### Test scale_and_center unknown scaler ### + + +def test_scale_and_center_unknown_scaler(example_data): + adata = _to_anndata(example_data) + with pytest.raises(NotImplementedError, match="Scaler .* not implemented"): + apt.pp.scale_and_center(adata, scaler="unknown") + + +### Test _validate_adata_for_completeness_filter ### + + +def test_validate_completeness_not_anndata(): + with pytest.raises(TypeError, match="adata must be an AnnData object"): + _validate_adata_for_completeness_filter(pd.DataFrame({"x": [1, 2]}), action="drop", var_colname="x") + + +def test_validate_completeness_no_features(): + adata = ad.AnnData(np.empty((3, 0))) + with pytest.raises(ValueError, match="no features"): + _validate_adata_for_completeness_filter(adata, action="drop", var_colname="x") + + +def test_validate_completeness_non_numeric_x(): + adata = ad.AnnData(np.array([["a", "b"], ["c", "d"]], dtype=object)) + with pytest.raises(ValueError, match="must be numeric"): + _validate_adata_for_completeness_filter(adata, action="drop", var_colname="x") + + +def test_validate_completeness_duplicated_obs(): + df = pd.DataFrame({"G1": [1.0, 2.0]}, index=["dup", "dup"]) + adata = _to_anndata(df) + with pytest.raises(ValueError, match="Duplicated indices"): + _validate_adata_for_completeness_filter(adata, action="drop", var_colname="x") + + +def test_validate_completeness_invalid_action(): + adata = _to_anndata(pd.DataFrame({"G1": [1.0, 2.0]}, index=["c1", "c2"])) + with pytest.raises(ValueError, match="must be 'flag' or 'drop'"): + _validate_adata_for_completeness_filter(adata, action="bogus", var_colname="x") + + +def test_validate_completeness_existing_var_colname(example_data, caplog): + # given — pre-fill var with the colname we'll later flag + adata = _to_anndata(example_data) + adata.var = pd.DataFrame({"my_flag": [True, True, True]}, index=["G1", "G2", "G3"]) + + # when + with caplog.at_level(logging.INFO): + _validate_adata_for_completeness_filter(adata, action="flag", var_colname="my_flag") + + # then + assert "already exists, will overwrite" in caplog.text + + +### Test filter_data_completeness error branches ### + + +@pytest.mark.parametrize("bad_threshold", [-0.1, 1.5]) +def test_filter_data_completeness_invalid_threshold(data_test_completeness_filter, bad_threshold): + with pytest.raises(ValueError, match="Threshold must be between 0 and 1"): + apt.pp.filter_data_completeness(data_test_completeness_filter, max_missing=bad_threshold) + + +def test_filter_data_completeness_unknown_group(data_test_completeness_filter): + with pytest.raises(ValueError, match="not found in"): + apt.pp.filter_data_completeness( + data_test_completeness_filter, + max_missing=0.5, + group_column="batch", + groups=["nonexistent_batch"], + ) diff --git a/tests/pp/test_impute.py b/tests/pp/test_impute.py index 5abf2649..aba9bf85 100644 --- a/tests/pp/test_impute.py +++ b/tests/pp/test_impute.py @@ -1,3 +1,4 @@ +import logging from typing import Any import anndata as ad @@ -12,6 +13,7 @@ _impute_knn, _impute_nanmedian, _raise_on_nan_values, + _validate_knn_grouping, ) @@ -129,6 +131,53 @@ def test___raise_on_nan_values(dummy_data_all_nan) -> None: _raise_on_nan_values(dummy_data_all_nan, mode="all") +def test_raise_on_nan_values_all_pandas(): + """`mode='all'` on a pandas DataFrame raises on all-NaN columns.""" + df = pd.DataFrame({"good": [1.0, 2.0], "all_nan": [np.nan, np.nan]}) + with pytest.raises(ValueError, match="contain all nan values"): + _raise_on_nan_values(df, mode="all") + + +def test_raise_on_nan_values_invalid_mode(): + with pytest.raises(ValueError, match="Mode must be either"): + _raise_on_nan_values(np.array([1.0, 2.0, np.nan]), mode="invalid") + + +def test_impute_nanmedian_no_nans(caplog): + """Complete data hits the short-circuit branch and is returned unchanged.""" + data = np.array([[1.0, 2.0], [3.0, 4.0]]) + + with caplog.at_level(logging.INFO): + result = _impute_nanmedian(data) + + assert "no missing values" in caplog.text + assert np.array_equal(result, data) + + +def test_impute_knn_no_nans(caplog): + """Complete data hits the short-circuit branch in _impute_knn.""" + data = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]]) + + with caplog.at_level(logging.INFO): + result = _impute_knn(data) + + assert "no missing values" in caplog.text + assert np.array_equal(result, data) + + +def test_validate_knn_grouping_nan_key(): + groups = {np.nan: np.array([0, 1])} + with pytest.raises(ValueError, match="contains nans"): + _validate_knn_grouping(groups, n_neighbors=2) + + +def test_validate_knn_grouping_group_too_small(): + """Group with fewer members than n_neighbors must raise.""" + groups = {"A": np.array([0]), "B": np.array([1, 2])} + with pytest.raises(ValueError, match="greater equal number of"): + _validate_knn_grouping(groups, n_neighbors=2) + + def test__impute_nanmedian(median_imputation_dummy_data) -> None: """Test median imputation for data with nan values""" X, X_ref = median_imputation_dummy_data @@ -565,6 +614,19 @@ def test_impute_bpca(self, bpca_imputation_dummy_anndata, layer: str, *, copy: b assert np.allclose(X_imputed, X_ref, equal_nan=False) + def test_impute_bpca_group_column(self, bpca_imputation_dummy_anndata) -> None: + """BPCA imputation with `group_column` runs the per-group write-back path.""" + adata, _, _, kwargs = bpca_imputation_dummy_anndata + X_original = adata.X.copy() + nan_mask = np.isnan(X_original) + + result = impute_bpca(adata, group_column="sample_group", **kwargs, copy=True) + + # NaNs were imputed + assert not np.any(np.isnan(result.X)) + # Non-NaN values were preserved + assert np.allclose(result.X[~nan_mask], X_original[~nan_mask]) + @pytest.mark.parametrize("group_column", [None, "sample_group"]) def test_impute_bpca__feature_all_nan(self, bpca_imputation_dummy_anndata_all_nan, group_column: str) -> None: """Test BPCA imputation raises if a feature contains all nan""" diff --git a/tests/pp/test_transform.py b/tests/pp/test_transform.py index 0acf44ee..a3b8b3c9 100644 --- a/tests/pp/test_transform.py +++ b/tests/pp/test_transform.py @@ -146,3 +146,9 @@ def test_check_data_integrity(input_data, expected_mask, verbosity, expect_warni assert "2 negative values in the data." in caplog.text assert "1 inf values in the data." in caplog.text assert "1 negative_inf values in the data." in caplog.text + + +def test_detect_special_values_invalid_type(): + """Non-ndarray input should raise TypeError.""" + with pytest.raises(TypeError, match="numpy.ndarray"): + detect_special_values([1.0, 2.0, 3.0])