Spaces:
Sleeping
Sleeping
File size: 9,354 Bytes
0304d75 | 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 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 | """
tests/test_engine.py
====================
Unit tests for intervention/engine.py — RecommendationEngine.
"""
import pytest
from intervention.engine import (
Intervention,
RecommendationEngine,
RecommendationPayload,
)
@pytest.fixture
def engine() -> RecommendationEngine:
return RecommendationEngine()
# ---------------------------------------------------------------------------
# Layer 1: Circuit Breaker
# ---------------------------------------------------------------------------
class TestCircuitBreaker:
def test_suicide_triggers_crisis(self, engine):
payload = engine.recommend("I want to commit suicide")
assert payload.is_crisis is True
assert payload.crisis_message is not None
assert "988" in payload.crisis_message
def test_kill_myself_triggers_crisis(self, engine):
payload = engine.recommend("I want to kill myself")
assert payload.is_crisis is True
def test_end_it_all_triggers_crisis(self, engine):
payload = engine.recommend("I just want to end it all")
assert payload.is_crisis is True
def test_self_harm_triggers_crisis(self, engine):
payload = engine.recommend("I've been self-harming lately")
assert payload.is_crisis is True
def test_want_to_die_triggers_crisis(self, engine):
payload = engine.recommend("I want to die")
assert payload.is_crisis is True
def test_crisis_halts_further_processing(self, engine):
"""When crisis is detected, only emergency interventions should be returned."""
payload = engine.recommend(
"I can't sleep and I want to kill myself", is_volatile=True
)
assert payload.is_crisis is True
# Should NOT have matched sleep triggers (halted)
assert len(payload.matched_triggers) == 0
def test_normal_text_no_crisis(self, engine):
payload = engine.recommend("I had a normal day at work")
assert payload.is_crisis is False
assert payload.crisis_message is None
# ---------------------------------------------------------------------------
# Layer 2: Context Matcher
# ---------------------------------------------------------------------------
class TestContextMatcher:
def test_sleep_trigger(self, engine):
payload = engine.recommend("I can't sleep at all")
assert "sleep" in payload.matched_triggers
assert any("sleep" in iv.title.lower() or "breath" in iv.title.lower()
for iv in payload.interventions)
def test_money_trigger(self, engine):
payload = engine.recommend("I'm worried about my debt and bills")
assert "money" in payload.matched_triggers
def test_exam_trigger(self, engine):
payload = engine.recommend("My finals are next week and I'm panicking")
assert "exam" in payload.matched_triggers
def test_work_trigger(self, engine):
payload = engine.recommend("My boss is terrible and I have a deadline")
assert "work" in payload.matched_triggers
def test_multiple_triggers(self, engine):
payload = engine.recommend("I can't sleep because of work stress and debt")
assert "sleep" in payload.matched_triggers
assert "work" in payload.matched_triggers
assert "money" in payload.matched_triggers
def test_no_triggers_for_generic_text(self, engine):
payload = engine.recommend("Today was a beautiful sunny day")
assert len(payload.matched_triggers) == 0
assert len(payload.interventions) == 0
# ---------------------------------------------------------------------------
# Layer 3: Preventive Nudges
# ---------------------------------------------------------------------------
class TestPreventiveNudges:
def test_volatile_user_gets_nudges(self, engine):
payload = engine.recommend(
"Today was a beautiful day", is_volatile=True
)
assert len(payload.interventions) > 0
assert any(iv.category == "grounding" for iv in payload.interventions)
def test_non_volatile_no_nudges(self, engine):
payload = engine.recommend("Today was a beautiful day", is_volatile=False)
assert len(payload.interventions) == 0
# ---------------------------------------------------------------------------
# Integration
# ---------------------------------------------------------------------------
class TestRecommendationIntegration:
def test_interventions_sorted_by_priority(self, engine):
payload = engine.recommend("I can't sleep and work is stressful")
priorities = [iv.priority for iv in payload.interventions]
assert priorities == sorted(priorities, reverse=True)
def test_payload_dataclass_fields(self, engine):
payload = engine.recommend("test text")
assert isinstance(payload, RecommendationPayload)
assert isinstance(payload.interventions, list)
assert isinstance(payload.matched_triggers, list)
assert isinstance(payload.is_crisis, bool)
# ---------------------------------------------------------------------------
# Layer 2: New trigger categories (relationship, health, grief, loneliness)
# ---------------------------------------------------------------------------
class TestNewTriggerCategories:
def test_relationship_trigger(self, engine):
payload = engine.recommend("My girlfriend and I had a terrible fight")
assert "relationship" in payload.matched_triggers
assert len(payload.interventions) > 0
def test_relationship_breakup_trigger(self, engine):
payload = engine.recommend("We just broke up and I am devastated")
assert "relationship" in payload.matched_triggers
def test_health_trigger(self, engine):
payload = engine.recommend("I have been sick and in a lot of pain")
assert "health" in payload.matched_triggers
assert len(payload.interventions) > 0
def test_grief_trigger(self, engine):
payload = engine.recommend("I am grieving the loss of my father")
assert "grief" in payload.matched_triggers
assert len(payload.interventions) > 0
def test_grief_passed_away(self, engine):
payload = engine.recommend("My grandmother passed away last week")
assert "grief" in payload.matched_triggers
def test_loneliness_trigger(self, engine):
payload = engine.recommend("I feel so lonely and isolated all the time")
assert "loneliness" in payload.matched_triggers
assert len(payload.interventions) > 0
def test_loneliness_alone_trigger(self, engine):
payload = engine.recommend("I have no friends and nobody cares about me")
assert "loneliness" in payload.matched_triggers
def test_new_categories_have_two_interventions_each(self, engine):
for category, text in [
("relationship", "My partner and I keep arguing"),
("health", "I have been dealing with a chronic illness"),
("grief", "I am still grieving the loss"),
("loneliness", "I feel lonely and excluded"),
]:
payload = engine.recommend(text)
matched_ivs = [
iv for iv in payload.interventions
if category in payload.matched_triggers
]
assert len(matched_ivs) >= 2, (
f"Expected ≥2 interventions for '{category}', got {len(matched_ivs)}"
)
# ---------------------------------------------------------------------------
# Layer 4: Escalation Tracker
# ---------------------------------------------------------------------------
class TestEscalationTracker:
def test_requires_escalation_flag_set(self, engine):
payload = engine.recommend(
"I am still so stressed", requires_escalation=True
)
assert payload.requires_escalation is True
def test_escalation_adds_high_priority_intervention(self, engine):
payload = engine.recommend(
"I am so stressed", requires_escalation=True
)
assert any(iv.priority >= 5 for iv in payload.interventions)
def test_no_escalation_by_default(self, engine):
payload = engine.recommend("I feel stressed at work")
assert payload.requires_escalation is False
def test_escalation_intervention_category_is_resource(self, engine):
payload = engine.recommend(
"Everything is overwhelming", requires_escalation=True
)
escalation_ivs = [iv for iv in payload.interventions if iv.priority >= 5]
assert any(iv.category == "resource" for iv in escalation_ivs)
def test_escalation_mentions_counsellor(self, engine):
payload = engine.recommend(
"I feel hopeless about everything", requires_escalation=True
)
text = " ".join(iv.description for iv in payload.interventions)
assert any(
kw in text.lower()
for kw in ("counsellor", "counselor", "therapist", "professional", "samhsa")
)
def test_crisis_takes_priority_over_escalation(self, engine):
"""Crisis (Layer 1) must still halt processing even with escalation flag."""
payload = engine.recommend(
"I want to end my life", requires_escalation=True
)
assert payload.is_crisis is True
|