File size: 3,432 Bytes
66a0e97 | 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 | import asyncio
import pytest
from unittest.mock import MagicMock, AsyncMock, patch
from app.utils.guvi_handler import GUVIHandler
from app.api.schemas import GUVIInputRequest, GUVIOutputResponseInternal, GUVIEngagementMetrics, GUVIIntelligence
# Mock objects
mock_orchestrator = MagicMock()
mock_orchestrator.initialized = True
mock_orchestrator.conversation_manager.get_or_create = AsyncMock(return_value={"history": []})
mock_orchestrator.process_message = AsyncMock(return_value={
"conversation": {"message_count": 2},
"is_scam": True,
"threat_level": "HIGH",
"should_finalize": True,
"aggregated_intelligence": {"sys_callback_sent": False}
})
mock_orchestrator.conversation_manager.update_intelligence = AsyncMock()
@pytest.mark.asyncio
async def test_handshake_compliance():
"""Test that handshake (message=None) returns strictly minimal reply and NO data."""
handler = GUVIHandler()
# Mock orchestrator inside handler
with patch("app.utils.guvi_handler.orchestrator", mock_orchestrator):
request = GUVIInputRequest(sessionId="test-session", message=None)
response = await handler.process_guvi_message(request)
assert response.reply == "Hello?"
assert response.data is None
assert response.status == "success"
@pytest.mark.asyncio
async def test_callback_trigger_logic():
"""Test that callback fires when total_messages >= 2 and scam detected."""
handler = GUVIHandler()
mock_bg_tasks = MagicMock()
# Mock orchestrator result for scam
mock_orchestrator.process_message.return_value = {
"conversation": {"message_count": 1}, # 1 turn = 2 messages
"is_scam": True,
"threat_level": "HIGH",
"should_finalize": True,
"aggregated_intelligence": {"sys_callback_sent": False},
"honeypot_response": "I am interested"
}
with patch("app.utils.guvi_handler.orchestrator", mock_orchestrator):
# Mock callback client imports (target the SOURCE, since it's locally imported)
with patch("app.utils.callback_client.guvi_callback") as mock_cb:
with patch("app.utils.guvi_handler.get_session_state", return_value="active"):
with patch("app.utils.guvi_handler.set_session_state"):
request = GUVIInputRequest(sessionId="test-session", message={"text": "scam"})
await handler.process_guvi_message(request, background_tasks=mock_bg_tasks)
# Verify Background Task Added
mock_bg_tasks.add_task.assert_called_once()
print("✅ Background task scheduled for callback")
# Verify DB Update Mock
mock_orchestrator.conversation_manager.update_intelligence.assert_called_with(
"test-session", {"sys_callback_sent": True}
)
print("✅ Intelligence updated (sys_callback_sent=True)")
if __name__ == "__main__":
# Manual run wrapper for environment without pytest
loop = asyncio.get_event_loop()
print("Running Handshake Test...")
loop.run_until_complete(test_handshake_compliance())
print("Running Callback Trigger Test...")
loop.run_until_complete(test_callback_trigger_logic())
print("🎉 ALL COMPLIANCE TESTS PASSED")
|