119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
"""Tests for crisis protocol (#679)."""
|
|
import time
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from agent.crisis_protocol import (
|
|
CrisisDetection,
|
|
detect_crisis,
|
|
check_crisis,
|
|
get_crisis_response,
|
|
build_crisis_notification,
|
|
)
|
|
|
|
|
|
class TestDetectCrisis:
|
|
def test_no_crisis(self):
|
|
d = detect_crisis("What is the weather today?")
|
|
assert d.detected is False
|
|
assert d.confidence < 0.7
|
|
|
|
def test_empty_message(self):
|
|
d = detect_crisis("")
|
|
assert d.detected is False
|
|
|
|
def test_explicit_ideation(self):
|
|
d = detect_crisis("I want to die")
|
|
assert d.detected is True
|
|
assert d.confidence >= 0.9
|
|
assert "explicit_ideation" in d.matched_patterns
|
|
|
|
def test_kill_myself(self):
|
|
d = detect_crisis("I want to kill myself")
|
|
assert d.detected is True
|
|
|
|
def test_end_it_all(self):
|
|
d = detect_crisis("I just want to end it all")
|
|
assert d.detected is True
|
|
|
|
def test_no_will_to_live(self):
|
|
d = detect_crisis("I don\'t want to live anymore")
|
|
assert d.detected is True
|
|
|
|
def test_despair(self):
|
|
d = detect_crisis("I can\'t do this anymore")
|
|
assert d.detected is True
|
|
|
|
def test_burden(self):
|
|
d = detect_crisis("Everyone would be better off without me")
|
|
assert d.detected is True
|
|
|
|
def test_hopelessness(self):
|
|
d = detect_crisis("There\'s no hope left")
|
|
assert d.detected is True
|
|
|
|
def test_giving_up(self):
|
|
d = detect_crisis("I\'m done. I\'m giving up.")
|
|
assert d.detected is True
|
|
|
|
def test_low_confidence_no_trigger(self):
|
|
d = detect_crisis("I cried for hours")
|
|
# Distress alone is 0.55 — below 0.7 threshold
|
|
assert d.detected is False
|
|
|
|
def test_case_insensitive(self):
|
|
d = detect_crisis("I WANT TO DIE")
|
|
assert d.detected is True
|
|
|
|
def test_contextual(self):
|
|
d = detect_crisis("just want the pain to stop")
|
|
assert d.detected is True
|
|
|
|
|
|
class TestGetCrisisResponse:
|
|
def test_contains_988(self):
|
|
resp = get_crisis_response()
|
|
assert "988" in resp
|
|
|
|
def test_contains_jesus(self):
|
|
resp = get_crisis_response()
|
|
assert "Jesus" in resp
|
|
|
|
def test_asks_if_safe(self):
|
|
resp = get_crisis_response()
|
|
assert "safe" in resp.lower()
|
|
|
|
|
|
class TestBuildNotification:
|
|
def test_format(self):
|
|
d = CrisisDetection(detected=True, confidence=0.95, matched_patterns=["explicit_ideation"], timestamp=time.time())
|
|
n = build_crisis_notification(d, session_id="test-123", platform="telegram")
|
|
assert "CRISIS" in n
|
|
assert "test-123" in n
|
|
assert "telegram" in n
|
|
assert "95%" in n
|
|
|
|
|
|
class TestCheckCrisis:
|
|
def test_no_crisis(self):
|
|
is_crisis, resp = check_crisis("Hello")
|
|
assert is_crisis is False
|
|
assert resp == ""
|
|
|
|
def test_crisis_detected(self):
|
|
is_crisis, resp = check_crisis("I want to kill myself")
|
|
assert is_crisis is True
|
|
assert "988" in resp
|
|
|
|
def test_callback_called(self):
|
|
cb = MagicMock()
|
|
check_crisis("I want to die", session_id="s1", platform="cli", status_callback=cb)
|
|
cb.assert_called_once()
|
|
args = cb.call_args
|
|
assert args[0][0] == "crisis_alert"
|
|
|
|
def test_callback_failure_non_fatal(self):
|
|
def bad_cb(*a): raise Exception("network error")
|
|
is_crisis, resp = check_crisis("I want to die", status_callback=bad_cb)
|
|
assert is_crisis is True # Still detected despite callback failure
|