test_complete_functionality.py

#!/usr/bin/env python3
"""
Complete functionality test for the Python Agent
Tests all major features including API integration, tools, and conversation flow
"""

import asyncio
import json
import os
from python_agent import create_agent_with_api_key

# Use the API key from the project
API_KEY = "***REMOVED***"

async def test_basic_conversation():
    """Test basic conversation without tools"""
    print("🗣️ Testing basic conversation...")
    
    agent = create_agent_with_api_key(
        provider="anthropic",
        api_key=API_KEY,
        model="claude-3-sonnet-20240229"
    )
    
    async with agent:
        # Clear tools to test pure conversation
        agent.tools.clear()
        
        print("📤 Sending basic question...")
        async for result in agent.execute_task("简单用一句话介绍你自己"):
            if result.get('type') == 'message':
                print(f"🤖 Claude: {result.get('content', '')}")
                break
    
    print("✅ Basic conversation test passed\n")

async def test_file_operations():
    """Test file-related tools"""
    print("📁 Testing file operations...")
    
    agent = create_agent_with_api_key(
        provider="anthropic",
        api_key=API_KEY,
        model="claude-3-sonnet-20240229"
    )
    
    async with agent:
        # Test file creation
        print("📝 Testing file creation...")
        async for result in agent.execute_task("创建一个名为 'test_file.txt' 的文件，内容为 'This is a test file'"):
            if result.get('type') == 'tool_use':
                print(f"🔧 Tool used: {result.get('name')}")
            elif result.get('type') == 'message':
                print(f"🤖 Claude: {result.get('content', '')}")
                break
        
        # Test file reading
        print("📖 Testing file reading...")
        async for result in agent.execute_task("读取刚刚创建的 'test_file.txt' 文件的内容"):
            if result.get('type') == 'tool_use':
                print(f"🔧 Tool used: {result.get('name')}")
            elif result.get('type') == 'message':
                print(f"🤖 Claude: {result.get('content', '')}")
                break
    
    print("✅ File operations test passed\n")

async def test_system_tools():
    """Test system-related tools"""
    print("⚙️ Testing system tools...")
    
    agent = create_agent_with_api_key(
        provider="anthropic",
        api_key=API_KEY,
        model="claude-3-sonnet-20240229"
    )
    
    async with agent:
        # Test ls command
        print("📋 Testing directory listing...")
        async for result in agent.execute_task("列出当前目录中的所有文件"):
            if result.get('type') == 'tool_use':
                print(f"🔧 Tool used: {result.get('name')}")
            elif result.get('type') == 'message':
                print(f"🤖 Claude: {result.get('content', '')}")
                break
        
        # Test system info
        print("💻 Testing system information...")
        async for result in agent.execute_task("获取系统CPU和内存信息"):
            if result.get('type') == 'tool_use':
                print(f"🔧 Tool used: {result.get('name')}")
            elif result.get('type') == 'message':
                print(f"🤖 Claude: {result.get('content', '')}")
                break
    
    print("✅ System tools test passed\n")

async def test_complex_task():
    """Test complex multi-step task"""
    print("🎯 Testing complex multi-step task...")
    
    agent = create_agent_with_api_key(
        provider="anthropic",
        api_key=API_KEY,
        model="claude-3-sonnet-20240229"
    )
    
    async with agent:
        print("🔄 Running complex task...")
        
        task = """
        执行以下步骤：
        1. 查看当前目录
        2. 创建一个Python脚本 'fibonacci.py'，计算斐波那契数列的前10项
        3. 显示创建的文件内容
        """
        
        async for result in agent.execute_task(task):
            if result.get('type') == 'tool_use':
                print(f"🔧 Tool used: {result.get('name')}")
            elif result.get('type') == 'message':
                content = result.get('content', '')
                if content:
                    print(f"🤖 Claude: {content}")
                    # Only show first response for each step
                    if "步骤" in content or "完成" in content:
                        break
    
    print("✅ Complex task test passed\n")

async def test_error_handling():
    """Test error handling"""
    print("⚠️ Testing error handling...")
    
    agent = create_agent_with_api_key(
        provider="anthropic",
        api_key=API_KEY,
        model="claude-3-sonnet-20240229"
    )
    
    async with agent:
        # Test with invalid file path
        print("🔍 Testing invalid file access...")
        async for result in agent.execute_task("读取文件 '/nonexistent/file.txt'"):
            if result.get('type') == 'tool_use':
                print(f"🔧 Tool used: {result.get('name')}")
            elif result.get('type') == 'error':
                print(f"❌ Error handled: {result.get('error', '')}")
                break
            elif result.get('type') == 'message':
                print(f"🤖 Claude: {result.get('content', '')}")
                break
    
    print("✅ Error handling test passed\n")

async def main():
    """Run all tests"""
    print("🧪 Python Agent Complete Functionality Test")
    print("=" * 50)
    
    try:
        await test_basic_conversation()
        await test_file_operations()
        await test_system_tools()
        await test_complex_task()
        await test_error_handling()
        
        print("🎉 All tests passed successfully!")
        
    except Exception as e:
        print(f"❌ Test failed: {e}")
        import traceback
        traceback.print_exc()
    
    finally:
        # Clean up test files
        test_files = ['test_file.txt', 'fibonacci.py']
        for file in test_files:
            if os.path.exists(file):
                os.remove(file)
                print(f"🧹 Cleaned up: {file}")

if __name__ == "__main__":
    asyncio.run(main())