Skip to content
Merged
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
143 changes: 143 additions & 0 deletions tests/test_data_loaders.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
"""Tests for data/loaders.py — sequence normalization, validation, and CSV loading."""
from __future__ import annotations

import csv
import tempfile
from pathlib import Path

from openamp_foundry.data.loaders import (
CANONICAL_AA,
is_valid_sequence,
load_candidates_csv,
normalize_sequence,
)


class TestNormalizeSequence:
def test_uppercase_conversion(self):
assert normalize_sequence("kwklf") == "KWKLF"

def test_strips_whitespace(self):
assert normalize_sequence(" KWKLF ") == "KWKLF"

def test_removes_internal_whitespace(self):
assert normalize_sequence("KWK LF") == "KWKLF"

def test_empty_string(self):
assert normalize_sequence("") == ""

def test_already_normalized(self):
assert normalize_sequence("KWKLFKKIGAVLKVL") == "KWKLFKKIGAVLKVL"

def test_mixed_case_and_spaces(self):
assert normalize_sequence(" kWk Lf ") == "KWKLF"


class TestIsValidSequence:
def test_canonical_sequence_valid(self):
assert is_valid_sequence("KWKLFKKIGAVLKVL") is True

def test_all_canonical_aa_valid(self):
assert is_valid_sequence("ACDEFGHIKLMNPQRSTVWY") is True

def test_empty_string_invalid(self):
assert is_valid_sequence("") is False

def test_lowercase_normalizes_and_is_valid(self):
# is_valid_sequence calls normalize_sequence internally, so lowercase is accepted
assert is_valid_sequence("kwklf") is True

def test_non_canonical_character_invalid(self):
assert is_valid_sequence("KWKLFX") is False

def test_number_in_sequence_invalid(self):
assert is_valid_sequence("KWKLF1") is False

def test_dash_in_sequence_invalid(self):
assert is_valid_sequence("KWKLF-KVLF") is False

def test_custom_allowed_set(self):
assert is_valid_sequence("AAA", allowed={"A"}) is True
assert is_valid_sequence("AAB", allowed={"A"}) is False

def test_single_residue_valid(self):
assert is_valid_sequence("K") is True

def test_all_20_canonical_aa_are_allowed(self):
for aa in CANONICAL_AA:
assert is_valid_sequence(aa) is True, f"{aa!r} should be valid"


class TestLoadCandidatesCsv:
def _write_csv(self, rows: list[dict], path: Path) -> None:
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
writer.writeheader()
writer.writerows(rows)

def test_loads_correct_count(self):
rows = [
{"id": "C1", "sequence": "KWKLFKKIGAVLKVL", "source": "test"},
{"id": "C2", "sequence": "DEDEDEDE", "source": "test"},
]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert len(candidates) == 2

def test_sequence_normalized_to_upper(self):
rows = [{"id": "C1", "sequence": "kwklf", "source": "test"}]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert candidates[0].sequence == "KWKLF"

def test_id_preserved(self):
rows = [{"id": "MY-SPECIAL-ID", "sequence": "KWKLF", "source": "test"}]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert candidates[0].candidate_id == "MY-SPECIAL-ID"

def test_missing_id_gets_default(self):
rows = [{"sequence": "KWKLF", "source": "test"}]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert candidates[0].candidate_id.startswith("candidate-")

def test_source_preserved(self):
rows = [{"id": "C1", "sequence": "KWKLF", "source": "my_dataset"}]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert candidates[0].source == "my_dataset"

def test_missing_source_defaults_to_csv(self):
rows = [{"id": "C1", "sequence": "KWKLF"}]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert candidates[0].source == "csv"

def test_returns_peptide_candidate_objects(self):
from openamp_foundry.types import PeptideCandidate
rows = [{"id": "C1", "sequence": "KWKLF", "source": "test"}]
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "candidates.csv"
self._write_csv(rows, path)
candidates = load_candidates_csv(path)
assert isinstance(candidates[0], PeptideCandidate)

def test_uses_demo_file_successfully(self):
demo = Path(__file__).parents[1] / "examples" / "sequences" / "demo_candidates.csv"
candidates = load_candidates_csv(demo)
assert len(candidates) > 0
for c in candidates:
assert len(c.sequence) > 0
188 changes: 180 additions & 8 deletions tests/test_physchem.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,184 @@
from openamp_foundry.features.physchem import compute_features, net_charge_proxy
"""Tests for features/physchem.py — all feature computation functions."""
from __future__ import annotations

from openamp_foundry.features.physchem import (
AROMATIC,
HYDROPHOBIC,
NEGATIVE,
POSITIVE,
compute_features,
fraction,
longest_repeat_run,
net_charge_proxy,
)

def test_net_charge_proxy():
assert net_charge_proxy("KKD") == 1
assert net_charge_proxy("DEK") == -1

class TestNetChargeProxy:
def test_positive_residues(self):
assert net_charge_proxy("KKK") == 3

def test_compute_features_length():
features = compute_features("KWKLFKKIGAVLKVL")
assert features["length"] == 15
assert features["net_charge_proxy"] > 0
def test_negative_residues(self):
assert net_charge_proxy("DDE") == -3 # D(-1) + D(-1) + E(-1)

def test_mixed_charge(self):
assert net_charge_proxy("KKD") == 1
assert net_charge_proxy("DEK") == -1

def test_neutral_sequence(self):
assert net_charge_proxy("AAAA") == 0

def test_empty_sequence(self):
assert net_charge_proxy("") == 0

def test_r_counts_positive(self):
assert net_charge_proxy("RRR") == 3

def test_h_counts_positive(self):
assert net_charge_proxy("HHH") == 3

def test_all_positive_aa(self):
assert net_charge_proxy("KRH") == 3

def test_d_and_e_both_negative(self):
assert net_charge_proxy("DE") == -2

def test_balanced_long_sequence(self):
assert net_charge_proxy("KKKKDDDD") == 0


class TestFraction:
def test_all_hydrophobic(self):
# AILMFWVY are HYDROPHOBIC
assert fraction("AAAA", HYDROPHOBIC) == 1.0

def test_none_hydrophobic(self):
assert fraction("KKKK", HYDROPHOBIC) == 0.0

def test_half_hydrophobic(self):
assert abs(fraction("KAKAKAKA", HYDROPHOBIC) - 0.5) < 1e-9

def test_empty_returns_zero(self):
assert fraction("", HYDROPHOBIC) == 0.0

def test_aromatic_fraction(self):
# FWY are AROMATIC
seq = "FWYK"
expected = 3 / 4
assert abs(fraction(seq, AROMATIC) - expected) < 1e-9

def test_single_residue_present(self):
assert fraction("K", POSITIVE) == 1.0

def test_single_residue_absent(self):
assert fraction("K", NEGATIVE) == 0.0


class TestLongestRepeatRun:
def test_no_repeats(self):
assert longest_repeat_run("KWKLF") == 1

def test_single_pair(self):
assert longest_repeat_run("KWKKF") == 2

def test_triple_run(self):
assert longest_repeat_run("KWKKKF") == 3

def test_whole_sequence_same(self):
assert longest_repeat_run("AAAAAAA") == 7

def test_empty_sequence(self):
assert longest_repeat_run("") == 0

def test_single_residue(self):
assert longest_repeat_run("K") == 1

def test_multiple_runs_returns_longest(self):
# "AAA" and "BB" — longest is 3
assert longest_repeat_run("KWKAAABBL") == 3

def test_run_at_end(self):
assert longest_repeat_run("KWLLLLL") == 5

def test_run_at_start(self):
assert longest_repeat_run("KKKKLF") == 4


class TestComputeFeatures:
def test_all_required_keys_present(self):
feat = compute_features("KWKLFKKIGAVLKVL")
required = {
"length", "net_charge_proxy", "charge_density",
"hydrophobic_fraction", "aromatic_fraction", "cysteine_fraction",
"glycine_fraction", "proline_fraction", "longest_repeat_run",
"hydrophobic_moment", "boman_index", "gravy", "residue_counts",
}
assert required <= set(feat.keys())

def test_length_correct(self):
assert compute_features("KWKLFKKIGAVLKVL")["length"] == 15
assert compute_features("AAAAA")["length"] == 5

def test_empty_sequence(self):
feat = compute_features("")
assert feat["length"] == 0
assert feat["net_charge_proxy"] == 0
assert feat["charge_density"] == 0.0
assert feat["hydrophobic_moment"] == 0.0

def test_charge_density_proportional_to_charge(self):
# KKK: charge=3, length=3 → density=1.0
feat = compute_features("KKK")
assert abs(feat["charge_density"] - 1.0) < 1e-9

def test_hydrophobic_fraction_correct(self):
# AAAK: A is hydrophobic (in AILMFWVY set? Let me check)
# HYDROPHOBIC = set("AILMFWVY") — A is IN this set
feat = compute_features("AAAK")
assert abs(feat["hydrophobic_fraction"] - 0.75) < 1e-9

def test_aromatic_fraction_correct(self):
# FWY are aromatic
feat = compute_features("FKWKY")
assert abs(feat["aromatic_fraction"] - 0.60) < 1e-9

def test_cysteine_fraction_correct(self):
feat = compute_features("KCCK")
assert abs(feat["cysteine_fraction"] - 0.5) < 1e-9

def test_glycine_fraction_correct(self):
feat = compute_features("KGGG")
assert abs(feat["glycine_fraction"] - 0.75) < 1e-9

def test_proline_fraction_correct(self):
feat = compute_features("KPPP")
assert abs(feat["proline_fraction"] - 0.75) < 1e-9

def test_longest_repeat_run_correct(self):
feat = compute_features("KWKAAAF")
assert feat["longest_repeat_run"] == 3

def test_residue_counts_match(self):
feat = compute_features("KKWKK")
counts = feat["residue_counts"]
assert counts["K"] == 4
assert counts["W"] == 1

def test_hydrophobic_moment_nonnegative(self):
feat = compute_features("KWKLFKKIGAVLKVL")
assert feat["hydrophobic_moment"] >= 0.0

def test_all_features_numeric(self):
feat = compute_features("KWKLFKKIGAVLKVL")
for key, val in feat.items():
if key != "residue_counts":
assert isinstance(val, (int, float)), f"{key}={val!r} is not numeric"

def test_positive_amp_has_positive_charge_density(self):
# Typical AMPs are cationic
feat = compute_features("KWKLFKKIGAVLKVL")
assert feat["charge_density"] > 0

def test_all_canonical_aa_no_crash(self):
seq = "ACDEFGHIKLMNPQRSTVWY"
feat = compute_features(seq)
assert feat["length"] == 20
Loading
Loading