Compare commits
1 Commits
fix/749-ba
...
fix/707-cr
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f03709aa29 |
132
tests/run_agent/test_crisis_hook_integration.py
Normal file
132
tests/run_agent/test_crisis_hook_integration.py
Normal file
@@ -0,0 +1,132 @@
|
||||
"""Integration test: crisis detection hook with run_agent conversation loop.
|
||||
|
||||
Verifies that crisis detection is properly integrated into the agent
|
||||
conversation flow — not just tested in isolation.
|
||||
|
||||
Part of #707.
|
||||
"""
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
class TestCrisisHookIntegration:
|
||||
"""Test crisis detection integration with the agent conversation loop."""
|
||||
|
||||
def test_scan_user_message_detects_crisis(self):
|
||||
"""Verify scan_user_message returns crisis data for dangerous input."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
result = scan_user_message("I want to kill myself")
|
||||
assert result is not None
|
||||
assert result["level"] == "CRITICAL"
|
||||
assert result["score"] == 1.0
|
||||
assert "988" in result["compassion_injection"]
|
||||
assert len(result["indicators"]) > 0
|
||||
|
||||
def test_scan_user_message_passes_safe(self):
|
||||
"""Verify scan_user_message returns None for safe input."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
result = scan_user_message("What's the weather like today?")
|
||||
assert result is None
|
||||
|
||||
def test_scan_user_message_medium_distress(self):
|
||||
"""Verify medium distress is detected and injection is warm."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
result = scan_user_message("I feel worthless and broken inside")
|
||||
assert result is not None
|
||||
assert result["level"] in ("MEDIUM", "LOW", "HIGH")
|
||||
assert result["score"] > 0
|
||||
assert len(result["compassion_injection"]) > 0
|
||||
|
||||
def test_crisis_detection_callable_from_agent_context(self):
|
||||
"""Verify crisis detection can be called from within agent code path."""
|
||||
# Simulate what run_agent.py would do: scan user message before API call
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
test_messages = [
|
||||
("Hello, how are you?", False),
|
||||
("I want to die", True),
|
||||
("Can you help me write a Python script?", False),
|
||||
("I feel so hopeless, nothing matters anymore", True),
|
||||
("Let's debug this code together", False),
|
||||
("I'm tired of living", True),
|
||||
]
|
||||
|
||||
for message, should_detect in test_messages:
|
||||
result = scan_user_message(message)
|
||||
if should_detect:
|
||||
assert result is not None, f"Should detect crisis in: {message}"
|
||||
assert result["level"] != "NONE"
|
||||
else:
|
||||
assert result is None, f"Should NOT detect crisis in: {message}"
|
||||
|
||||
def test_compassion_injection_has_lifeline(self):
|
||||
"""Verify all non-trivial crisis levels surface the 988 lifeline."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
crisis_messages = [
|
||||
"I want to end it all",
|
||||
"Everything is hopeless and broken",
|
||||
"I feel worthless",
|
||||
]
|
||||
|
||||
for msg in crisis_messages:
|
||||
result = scan_user_message(msg)
|
||||
if result and result["level"] in ("CRITICAL", "HIGH"):
|
||||
assert "988" in result["compassion_injection"], \
|
||||
f"CRITICAL/HIGH should include 988: {msg}"
|
||||
|
||||
def test_tool_handler_returns_json(self):
|
||||
"""Verify crisis_scan tool handler returns valid JSON."""
|
||||
from tools.crisis_tool import crisis_scan_handler
|
||||
|
||||
result = crisis_scan_handler({"text": "I want to die"})
|
||||
data = json.loads(result)
|
||||
assert "level" in data
|
||||
assert "score" in data
|
||||
assert data["level"] == "CRITICAL"
|
||||
|
||||
def test_empty_text_handled(self):
|
||||
"""Verify empty/None text doesn't crash."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
assert scan_user_message("") is None
|
||||
assert scan_user_message(None) is None
|
||||
assert scan_user_message(" ") is None
|
||||
|
||||
def test_detection_is_case_insensitive(self):
|
||||
"""Verify crisis detection works regardless of case."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
assert scan_user_message("I WANT TO DIE") is not None
|
||||
assert scan_user_message("i want to die") is not None
|
||||
assert scan_user_message("I Want To Die") is not None
|
||||
|
||||
def test_false_positive_resistance(self):
|
||||
"""Verify common non-crisis phrases don't trigger false positives."""
|
||||
from tools.crisis_tool import scan_user_message
|
||||
|
||||
safe_phrases = [
|
||||
"This code is killing me (debugging is hard)",
|
||||
"I'm dead tired from this marathon",
|
||||
"The deadline is going to bury me",
|
||||
"This bug is the death of my patience",
|
||||
"I could die for some coffee right now",
|
||||
"That test killed it! Great results!",
|
||||
]
|
||||
|
||||
for phrase in safe_phrases:
|
||||
result = scan_user_message(phrase)
|
||||
# These should either not trigger or trigger LOW at most
|
||||
if result:
|
||||
assert result["level"] in ("LOW", "NONE"), \
|
||||
f"False positive on: {phrase} -> {result['level']}"
|
||||
|
||||
def test_config_check_returns_bool(self):
|
||||
"""Verify the config check function works."""
|
||||
from tools.crisis_tool import _is_crisis_detection_enabled
|
||||
result = _is_crisis_detection_enabled()
|
||||
assert isinstance(result, bool)
|
||||
Reference in New Issue
Block a user