test_utils.py

"""
Test file để kiểm tra toàn bộ utils system
"""

import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from utils import (
    process_question_with_tools,
    get_agent_state,
    reset_agent_state,
    analyze_question_type,
    reverse_text_if_needed,
    get_system_prompt
)

def test_text_processing():
    """Test text processing tool"""
    print("=== Testing Text Processing ===")
    
    # Test reverse text
    reversed_question = ".rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw ,ecnetnes siht dnatsrednu uoy fI"
    processed = reverse_text_if_needed(reversed_question)
    print(f"Original: {reversed_question}")
    print(f"Processed: {processed}")
    print()

def test_question_analysis():
    """Test question type analysis"""
    print("=== Testing Question Analysis ===")
    
    test_questions = [
        "Who was Marie Curie?",
        "What is in this YouTube video? https://www.youtube.com/watch?v=dQw4w9WgXcQ",
        "Can you read this image for me?",
        "What does this audio file say?",
        "Calculate 2+2*3",
        ".rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw"
    ]
    
    for question in test_questions:
        q_type = analyze_question_type(question)
        print(f"Question: {question[:50]}...")
        print(f"Type: {q_type}")
        print()

def test_state_management():
    """Test state management"""
    print("=== Testing State Management ===")
    
    # Reset state
    reset_agent_state()
    state = get_agent_state()
    
    # Start task
    task = state.start_new_task("test_001", "Who is Einstein?")
    print(f"Task created: {task.task_id}")
    print(f"Question: {task.question}")
    
    # Update context
    state.update_task_context(question_type="wiki", processed_question="Who is Einstein?")
    print(f"Updated type: {task.question_type}")
    
    # Generate summary
    summary = state.generate_context_summary()
    print(f"Context summary:\n{summary}")
    print()

def test_prompts():
    """Test prompt system"""
    print("=== Testing Prompts ===")
    
    # System prompt
    system_prompt = get_system_prompt("main_agent")
    print(f"System prompt length: {len(system_prompt)} characters")
    print(f"First 100 chars: {system_prompt[:100]}...")
    print()

def test_orchestration():
    """Test tool orchestration"""
    print("=== Testing Tool Orchestration ===")
    
    # Test with Wikipedia question (no external dependencies)
    question = "Who was Albert Einstein?"
    print(f"Processing question: {question}")
    
    try:
        result = process_question_with_tools(question)
        print(f"Question type detected: {result['question_type']}")
        print(f"Tools used: {result['tools_used']}")
        print(f"Successful tools: {result['successful_tools']}")
        print(f"Failed tools: {result['failed_tools']}")
        
        # Print tool results summary
        if result['tool_results']:
            print("\nTool Results Summary:")
            for tool_result in result['tool_results']:
                status = "✅" if tool_result.success else "❌"
                print(f"  {status} {tool_result.tool_name}: {tool_result.execution_time:.2f}s")
                if not tool_result.success and tool_result.error_message:
                    print(f"    Error: {tool_result.error_message}")
        
        print("\nCached data keys:", list(result['cached_data'].keys()))
        
    except Exception as e:
        print(f"Error in orchestration: {e}")
        import traceback
        traceback.print_exc()
    
    print()

def test_youtube_detection():
    """Test YouTube URL detection"""
    print("=== Testing YouTube Detection ===")
    
    youtube_question = "What is this video about? https://www.youtube.com/watch?v=dQw4w9WgXcQ"
    print(f"Processing YouTube question: {youtube_question}")
    
    try:
        result = process_question_with_tools(youtube_question)
        print(f"Question type: {result['question_type']}")
        print(f"Tools to run: {result['tools_used']}")
        
        # Check if YouTube tool was attempted
        youtube_results = [r for r in result['tool_results'] if r.tool_name == "youtube_tool"]
        if youtube_results:
            youtube_result = youtube_results[0]
            print(f"YouTube tool status: {'Success' if youtube_result.success else 'Failed'}")
            if youtube_result.success:
                print(f"YouTube content available: {bool(youtube_result.result)}")
            else:
                print(f"YouTube error: {youtube_result.error_message}")
        
    except Exception as e:
        print(f"Error testing YouTube: {e}")
    
    print()

def run_all_tests():
    """Run all tests"""
    print("🧪 Starting Utils System Tests\n")
    
    test_text_processing()
    test_question_analysis() 
    test_state_management()
    test_prompts()
    test_orchestration()
    test_youtube_detection()
    
    print("✅ All tests completed!")

if __name__ == "__main__":
    # Set environment variables if needed
    os.environ.setdefault("GROQ_API_KEY", "your_groq_key_here")
    os.environ.setdefault("HF_TOKEN", "your_hf_token_here")
    
    run_all_tests()