test_system.py

"""
Test script for GemmaSOS Crisis Intervention System
Tests various crisis scenarios and validates system responses
"""

import sys
import os
import logging
from typing import Dict, List, Any
import json

# Add the current directory to the path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from crisis_detector import CrisisDetector
from response_generator import CrisisResponseGenerator
from safety_system import SafetySystem

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CrisisSystemTester:
    """
    Test suite for the crisis intervention system
    """
    
    def __init__(self):
        """Initialize the test suite"""
        self.test_results = []
        self.crisis_detector = None
        self.response_generator = None
        self.safety_system = None
        
    def setup_components(self):
        """Initialize all system components"""
        try:
            logger.info("Setting up test components...")
            self.crisis_detector = CrisisDetector()
            self.response_generator = CrisisResponseGenerator()
            self.safety_system = SafetySystem()
            logger.info("Components initialized successfully")
            return True
        except Exception as e:
            logger.error(f"Error setting up components: {e}")
            return False
    
    def test_crisis_detection(self):
        """Test crisis detection with various scenarios"""
        logger.info("Testing crisis detection...")
        
        test_cases = [
            {
                "text": "I'm feeling really down today",
                "expected_crisis": False,
                "description": "General sadness - no crisis"
            },
            {
                "text": "I want to hurt myself tonight",
                "expected_crisis": True,
                "expected_type": "self_harm",
                "description": "Self-harm crisis"
            },
            {
                "text": "I don't want to live anymore",
                "expected_crisis": True,
                "expected_type": "suicide",
                "description": "Suicidal ideation"
            },
            {
                "text": "I'm going to hurt someone",
                "expected_crisis": True,
                "expected_type": "violence",
                "description": "Violence threat"
            },
            {
                "text": "He hits me every day",
                "expected_crisis": True,
                "expected_type": "abuse",
                "description": "Abuse situation"
            },
            {
                "text": "I took too many pills",
                "expected_crisis": True,
                "expected_type": "overdose",
                "description": "Overdose situation"
            }
        ]
        
        results = []
        for i, test_case in enumerate(test_cases):
            try:
                result = self.crisis_detector.detect_crisis_from_text(test_case["text"])
                
                # Check if crisis detection matches expectation
                crisis_detected = result["crisis_detected"]
                expected_crisis = test_case["expected_crisis"]
                
                # Check crisis type if expected
                crisis_type_match = True
                if "expected_type" in test_case and result["categories"]:
                    detected_type = result["categories"][0]["category"]
                    expected_type = test_case["expected_type"]
                    crisis_type_match = detected_type == expected_type
                
                test_result = {
                    "test_case": i + 1,
                    "description": test_case["description"],
                    "text": test_case["text"],
                    "crisis_detected": crisis_detected,
                    "expected_crisis": expected_crisis,
                    "crisis_type_match": crisis_type_match,
                    "confidence": result["confidence"],
                    "passed": crisis_detected == expected_crisis and crisis_type_match,
                    "result": result
                }
                
                results.append(test_result)
                
                status = "PASS" if test_result["passed"] else "FAIL"
                logger.info(f"Test {i+1}: {status} - {test_case['description']}")
                
            except Exception as e:
                logger.error(f"Error in test case {i+1}: {e}")
                results.append({
                    "test_case": i + 1,
                    "description": test_case["description"],
                    "error": str(e),
                    "passed": False
                })
        
        return results
    
    def test_response_generation(self):
        """Test response generation for different crisis types"""
        logger.info("Testing response generation...")
        
        test_cases = [
            {
                "crisis_type": "self_harm",
                "user_message": "I want to cut myself",
                "confidence": 0.8,
                "immediate_risk": True,
                "description": "High-risk self-harm"
            },
            {
                "crisis_type": "suicide",
                "user_message": "I don't want to live anymore",
                "confidence": 0.7,
                "immediate_risk": False,
                "description": "Suicidal ideation"
            },
            {
                "crisis_type": "violence",
                "user_message": "I'm going to hurt someone",
                "confidence": 0.6,
                "immediate_risk": True,
                "description": "Violence threat"
            },
            {
                "crisis_type": "abuse",
                "user_message": "He hits me every day",
                "confidence": 0.9,
                "immediate_risk": True,
                "description": "Abuse situation"
            },
            {
                "crisis_type": "overdose",
                "user_message": "I took too many pills",
                "confidence": 0.8,
                "immediate_risk": True,
                "description": "Overdose situation"
            }
        ]
        
        results = []
        for i, test_case in enumerate(test_cases):
            try:
                response = self.response_generator.generate_response(
                    crisis_type=test_case["crisis_type"],
                    user_message=test_case["user_message"],
                    confidence=test_case["confidence"],
                    immediate_risk=test_case["immediate_risk"]
                )
                
                # Validate response structure
                required_fields = ["response", "resources", "immediate_risk", "crisis_type"]
                has_required_fields = all(field in response for field in required_fields)
                
                # Check if response is empathetic
                response_text = response["response"].lower()
                empathetic_indicators = ["i'm", "i care", "i understand", "i'm here", "support"]
                is_empathetic = any(indicator in response_text for indicator in empathetic_indicators)
                
                # Check if resources are provided
                has_resources = len(response["resources"]) > 0
                
                test_result = {
                    "test_case": i + 1,
                    "description": test_case["description"],
                    "has_required_fields": has_required_fields,
                    "is_empathetic": is_empathetic,
                    "has_resources": has_resources,
                    "response_length": len(response["response"]),
                    "resource_count": len(response["resources"]),
                    "passed": has_required_fields and is_empathetic and has_resources,
                    "response": response
                }
                
                results.append(test_result)
                
                status = "PASS" if test_result["passed"] else "FAIL"
                logger.info(f"Response Test {i+1}: {status} - {test_case['description']}")
                
            except Exception as e:
                logger.error(f"Error in response test case {i+1}: {e}")
                results.append({
                    "test_case": i + 1,
                    "description": test_case["description"],
                    "error": str(e),
                    "passed": False
                })
        
        return results
    
    def test_safety_system(self):
        """Test safety system and content filtering"""
        logger.info("Testing safety system...")
        
        test_cases = [
            {
                "text": "I'm feeling sad today",
                "expected_safe": True,
                "description": "Safe content"
            },
            {
                "text": "How to cut yourself step by step",
                "expected_safe": False,
                "description": "Harmful instructional content"
            },
            {
                "text": "I want to hurt myself",
                "expected_safe": True,
                "description": "Crisis content (should be safe to process)"
            },
            {
                "text": "Don't tell anyone about this",
                "expected_safe": False,
                "description": "Dangerous advice"
            }
        ]
        
        results = []
        for i, test_case in enumerate(test_cases):
            try:
                validation = self.safety_system.validate_input(text=test_case["text"])
                
                is_safe = validation["is_safe"]
                expected_safe = test_case["expected_safe"]
                
                test_result = {
                    "test_case": i + 1,
                    "description": test_case["description"],
                    "text": test_case["text"],
                    "is_safe": is_safe,
                    "expected_safe": expected_safe,
                    "warnings": validation["warnings"],
                    "blocked_content": validation["blocked_content"],
                    "passed": is_safe == expected_safe,
                    "validation": validation
                }
                
                results.append(test_result)
                
                status = "PASS" if test_result["passed"] else "FAIL"
                logger.info(f"Safety Test {i+1}: {status} - {test_case['description']}")
                
            except Exception as e:
                logger.error(f"Error in safety test case {i+1}: {e}")
                results.append({
                    "test_case": i + 1,
                    "description": test_case["description"],
                    "error": str(e),
                    "passed": False
                })
        
        return results
    
    def test_privacy_protection(self):
        """Test privacy protection measures"""
        logger.info("Testing privacy protection...")
        
        test_data = "This is sensitive test data"
        
        try:
            # Test data processing
            data_id = self.safety_system.ensure_privacy(test_data, "test_session")
            
            # Test privacy summary
            privacy_info = self.safety_system.get_privacy_summary()
            
            # Validate privacy measures
            privacy_checks = {
                "data_processing": privacy_info["data_processing"] == "on_device_only",
                "data_storage": privacy_info["data_storage"] == "temporary_metadata_only",
                "data_transmission": privacy_info["data_transmission"] == "none",
                "has_privacy_measures": len(privacy_info["privacy_measures"]) > 0
            }
            
            all_checks_passed = all(privacy_checks.values())
            
            test_result = {
                "description": "Privacy protection validation",
                "data_id_generated": data_id is not None,
                "privacy_checks": privacy_checks,
                "all_checks_passed": all_checks_passed,
                "passed": all_checks_passed,
                "privacy_info": privacy_info
            }
            
            status = "PASS" if test_result["passed"] else "FAIL"
            logger.info(f"Privacy Test: {status} - Privacy protection validation")
            
            return [test_result]
            
        except Exception as e:
            logger.error(f"Error in privacy test: {e}")
            return [{
                "description": "Privacy protection validation",
                "error": str(e),
                "passed": False
            }]
    
    def run_all_tests(self):
        """Run all test suites"""
        logger.info("Starting comprehensive system testing...")
        
        if not self.setup_components():
            logger.error("Failed to setup components. Aborting tests.")
            return False
        
        all_results = {
            "crisis_detection": self.test_crisis_detection(),
            "response_generation": self.test_response_generation(),
            "safety_system": self.test_safety_system(),
            "privacy_protection": self.test_privacy_protection()
        }
        
        # Calculate overall results
        total_tests = sum(len(results) for results in all_results.values())
        passed_tests = sum(
            sum(1 for result in results if result.get("passed", False))
            for results in all_results.values()
        )
        
        # Print summary
        logger.info("\n" + "="*50)
        logger.info("TEST SUMMARY")
        logger.info("="*50)
        
        for test_suite, results in all_results.items():
            suite_passed = sum(1 for result in results if result.get("passed", False))
            suite_total = len(results)
            logger.info(f"{test_suite}: {suite_passed}/{suite_total} tests passed")
        
        logger.info(f"Overall: {passed_tests}/{total_tests} tests passed")
        
        if passed_tests == total_tests:
            logger.info("🎉 ALL TESTS PASSED! System is ready for use.")
        else:
            logger.warning(f"⚠️ {total_tests - passed_tests} tests failed. Please review the results.")
        
        # Save detailed results
        self.save_test_results(all_results)
        
        return passed_tests == total_tests
    
    def save_test_results(self, results: Dict[str, List[Dict]]):
        """Save test results to file"""
        try:
            with open("test_results.json", "w") as f:
                json.dump(results, f, indent=2)
            logger.info("Test results saved to test_results.json")
        except Exception as e:
            logger.error(f"Error saving test results: {e}")

def main():
    """Main test function"""
    tester = CrisisSystemTester()
    success = tester.run_all_tests()
    
    if success:
        print("\n✅ All tests passed! The system is ready for use.")
        sys.exit(0)
    else:
        print("\n❌ Some tests failed. Please review the results.")
        sys.exit(1)

if __name__ == "__main__":
    main()