Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions src/alphapepttools/pp/data.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why no cover by tests? (for the above raise) statements it might be fine to not cover them, this is a functional piece of logic)

# 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
Expand Down
54 changes: 54 additions & 0 deletions tests/pp/test_batch_correction.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment on lines +133 to +134

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could we test for the full result here?



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")
Loading
Loading