File size: 5,795 Bytes
10795db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """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")
|