BioPatch / backend /tests /test_ai_engine.py
Jerry
BioPatch backend (clean state, history rewritten)
10795db
Raw History Blame Contribute Delete
5.8 kB
"""Smoke tests for the BioPatch AI engine.
Run with `python -m pytest backend/tests` once pytest is installed, or
just `python backend/tests/test_ai_engine.py` for a manual sanity check.
"""
from __future__ import annotations
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from ai.features import PatientFactors # noqa: E402
from core.ai_engine import Reading, SepsisRiskEngine, assess_reading # noqa: E402
def test_stable_healing_reading_is_low_risk() -> None:
result = assess_reading(temperature_c=33.5, ph=6.0, humidity=75.0)
assert result["score"] < 25, result
assert result["severity"] == "stable"
assert result["bacterial_signature"] is None
def test_classic_ssi_pattern_is_critical() -> None:
result = assess_reading(temperature_c=38.6, ph=8.1, humidity=82.0)
assert result["score"] >= 75, result
assert result["severity"] == "critical"
assert result["bacterial_signature"] is not None
def test_pseudomonas_signature_detection() -> None:
result = assess_reading(temperature_c=37.8, ph=8.3, humidity=88.0)
sig = result["bacterial_signature"] or ""
assert "Pseudomonas" in sig, result
def test_velocity_term_flags_rapid_co_deterioration() -> None:
engine = SepsisRiskEngine()
base = datetime.now(timezone.utc)
# Baseline reading (6h earlier, in the velocity window)
engine.assess(Reading(36.8, 7.0, 78.0, timestamp=base - timedelta(hours=5)))
# Now: temp + pH have risen significantly
later = engine.assess(Reading(38.4, 7.9, 80.0, timestamp=base))
drivers_text = " | ".join(later.drivers)
assert "rising" in drivers_text.lower() or "shift" in drivers_text.lower(), later
def test_scores_are_bounded() -> None:
extreme = assess_reading(temperature_c=42.0, ph=10.0, humidity=99.0)
assert 0.0 <= extreme["score"] <= 100.0
def test_granulation_classification() -> None:
result = assess_reading(temperature_c=34.5, ph=7.2, humidity=80.0)
assert result["tissue_state"] == "granulation"
assert result["tissue_confidence"] >= 0.65
assert "Granulação" in result["tissue_state_label_pt"]
def test_devitalized_classification_ischemic_acidic() -> None:
"""Spec: low temp + acidic pH (<6.5) -> devitalized."""
result = assess_reading(temperature_c=30.5, ph=6.0, humidity=60.0)
assert result["tissue_state"] == "devitalized", result
assert "Desvitalização" in result["tissue_state_label_pt"]
def test_inflammatory_tissue_state() -> None:
result = assess_reading(temperature_c=37.8, ph=7.6, humidity=78.0)
assert result["tissue_state"] == "inflammatory"
def test_fusion_quality_penalises_isolated_axis_spike() -> None:
isolated = assess_reading(temperature_c=33.5, ph=8.4, humidity=78.0)
full = assess_reading(temperature_c=38.4, ph=8.4, humidity=88.0)
# Sensor contributions should be more concentrated in the isolated case
assert isolated["fusion_quality"] < full["fusion_quality"], (isolated, full)
assert full["score"] > isolated["score"]
def test_xai_sensor_contributions_sum_to_one() -> None:
result = assess_reading(temperature_c=38.2, ph=7.6, humidity=84.0)
contribs = result["sensor_contributions"]
assert set(contribs) == {"temperature", "ph", "humidity"}
total = sum(contribs.values())
assert 0.97 <= total <= 1.03, total
def test_context_factor_raises_with_risk_factors() -> None:
base_result = assess_reading(
temperature_c=37.2, ph=7.3, humidity=80.0,
factors=PatientFactors(age=45, diabetes=False, smoker=False),
)
frail_result = assess_reading(
temperature_c=37.2, ph=7.3, humidity=80.0,
factors=PatientFactors(age=78, diabetes=True, smoker=True),
)
assert frail_result["context_factor"] > base_result["context_factor"]
# Same sensor numbers, more fragile patient -> at least same severity bucket
sev_rank = {"stable": 0, "watch": 1, "warning": 2, "critical": 3}
assert sev_rank[frail_result["severity"]] >= sev_rank[base_result["severity"]]
def test_model_loaded_and_returns_class_probabilities() -> None:
result = assess_reading(temperature_c=36.0, ph=7.0, humidity=78.0)
probs = result["class_probabilities"]
assert set(probs) == {"stable", "watch", "warning", "critical"}
assert 0.97 <= sum(probs.values()) <= 1.03
assert result["model_version"].startswith("rf-")
def test_feature_importance_reasonable_after_training() -> None:
result = assess_reading(temperature_c=36.0, ph=7.0, humidity=78.0)
fi = result["feature_importance"]
expected = {
"temperature_c", "ph", "humidity",
"temp_delta_6h", "ph_delta_6h", "humidity_volatility_6h",
"age", "diabetes", "smoker",
}
assert set(fi) == expected
# The three sensors should carry most of the model's signal
sensor_sum = fi["temperature_c"] + fi["ph"] + fi["humidity"]
velocity_sum = fi["temp_delta_6h"] + fi["ph_delta_6h"] + fi["humidity_volatility_6h"]
assert sensor_sum + velocity_sum > 0.7
if __name__ == "__main__":
test_stable_healing_reading_is_low_risk()
test_classic_ssi_pattern_is_critical()
test_pseudomonas_signature_detection()
test_velocity_term_flags_rapid_co_deterioration()
test_scores_are_bounded()
test_granulation_classification()
test_devitalized_classification_ischemic_acidic()
test_inflammatory_tissue_state()
test_fusion_quality_penalises_isolated_axis_spike()
test_xai_sensor_contributions_sum_to_one()
test_context_factor_raises_with_risk_factors()
test_model_loaded_and_returns_class_probabilities()
test_feature_importance_reasonable_after_training()
print("OK - all ai_engine smoke tests passed")