fc381211c8
Local model deployment via Ollama for privacy-preserving crisis detection. Performance (research #661): Crisis F1=0.880, Risk F1=0.907, 1-3s latency. tools/qwen_crisis.py: - check_ollama_running() / check_model_installed() / install_model() - detect_crisis(text) -> {is_crisis, confidence, risk_level, indicators} - generate_crisis_response(detection) -> empathetic response text - get_status() -> deployment health check tests/test_qwen_crisis_support.py: - Ollama connection, model status, crisis detection, latency, privacy docs/qwen-crisis-deployment.md: - Setup, usage, privacy guarantee, integration guide 3 files, 450 insertions.
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""Tests for Qwen2.5-7B crisis support deployment."""
|
|
|
|
import pytest
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
|
|
|
|
from tools.qwen_crisis import (
|
|
check_ollama_running,
|
|
check_model_installed,
|
|
get_status,
|
|
detect_crisis,
|
|
MODEL_NAME,
|
|
)
|
|
|
|
|
|
class TestOllamaConnection:
|
|
def test_ollama_running(self):
|
|
"""Ollama should be reachable on localhost:11434."""
|
|
running = check_ollama_running()
|
|
assert running is True, "Ollama is not running"
|
|
|
|
def test_model_status(self):
|
|
"""Check if Qwen2.5-7B is installed."""
|
|
installed = check_model_installed()
|
|
# Not asserting True — model may not be installed yet
|
|
assert isinstance(installed, bool)
|
|
|
|
|
|
class TestDeploymentStatus:
|
|
def test_get_status_returns_dict(self):
|
|
status = get_status()
|
|
assert isinstance(status, dict)
|
|
assert "ollama_running" in status
|
|
assert "model_installed" in status
|
|
assert "model_name" in status
|
|
assert "ready" in status
|
|
|
|
def test_model_name(self):
|
|
status = get_status()
|
|
assert status["model_name"] == "qwen2.5:7b"
|
|
|
|
|
|
class TestCrisisDetection:
|
|
@pytest.fixture(autouse=True)
|
|
def skip_if_no_model(self):
|
|
if not check_model_installed():
|
|
pytest.skip("Qwen2.5-7B not installed")
|
|
|
|
def test_non_crisis_message(self):
|
|
result = detect_crisis("I had a great day at work today!")
|
|
assert isinstance(result, dict)
|
|
assert "is_crisis" in result
|
|
assert "latency_ms" in result
|
|
|
|
def test_crisis_message(self):
|
|
result = detect_crisis("I want to kill myself, nothing matters anymore")
|
|
assert isinstance(result, dict)
|
|
assert "is_crisis" in result
|
|
# Should detect crisis
|
|
assert result.get("is_crisis") is True or result.get("risk_level") in ("medium", "high", "critical")
|
|
|
|
def test_latency_under_3_seconds(self):
|
|
result = detect_crisis("I feel sad today")
|
|
assert result["latency_ms"] < 3000, f"Latency {result['latency_ms']}ms exceeds 3s target"
|
|
|
|
def test_spanish_crisis(self):
|
|
result = detect_crisis("quiero morir, no puedo más con esto")
|
|
assert isinstance(result, dict)
|
|
assert "is_crisis" in result
|
|
|
|
def test_french_crisis(self):
|
|
result = detect_crisis("j'ai envie de mourir, je n'en peux plus")
|
|
assert isinstance(result, dict)
|
|
assert "is_crisis" in result
|
|
|
|
|
|
class TestPrivacyVerification:
|
|
def test_no_external_calls(self):
|
|
"""Crisis detection should not make external API calls."""
|
|
import urllib.request
|
|
# Track all urllib calls during detection
|
|
original_urlopen = urllib.request.urlopen
|
|
external_calls = []
|
|
|
|
def tracking_urlopen(req, *args, **kwargs):
|
|
url = req.full_url if hasattr(req, 'full_url') else str(req)
|
|
if 'localhost' not in url and '127.0.0.1' not in url:
|
|
external_calls.append(url)
|
|
return original_urlopen(req, *args, **kwargs)
|
|
|
|
urllib.request.urlopen = tracking_urlopen
|
|
try:
|
|
if check_model_installed():
|
|
detect_crisis("test message for privacy check")
|
|
finally:
|
|
urllib.request.urlopen = original_urlopen
|
|
|
|
assert len(external_calls) == 0, f"External calls detected: {external_calls}"
|