verify_enhanced_features.py

"""
Manual verification script for Phase 6 testing.
Tests session management, file operations, and multi-language execution.
"""

import requests
import time
import sys
from pathlib import Path

BASE_URL = "http://localhost:7860"

def print_test(name):
    print(f"\n{'='*60}")
    print(f"TEST: {name}")
    print(f"{'='*60}")

def print_success(msg):
    print(f"✅ {msg}")

def print_error(msg):
    print(f"❌ {msg}")
    
def print_info(msg):
    print(f"ℹ️  {msg}")


def test_1_session_lifecycle():
    """Test creating, getting, and destroying a session"""
    print_test("Session Lifecycle")
    
    # Create session
    print_info("Creating session...")
    resp = requests.post(f"{BASE_URL}/sessions", json={
        "metadata": {"test": "lifecycle"},
        "timeout_minutes": 30
    })
    
    if resp.status_code != 201:
        print_error(f"Failed to create session: {resp.status_code}")
        return None
    
    session = resp.json()
    session_id = session["session_id"]
    print_success(f"Session created: {session_id}")
    
    # Wait for container to be ready
    time.sleep(3)
    
    # Get session details
    print_info("Getting session details...")
    resp = requests.get(f"{BASE_URL}/sessions/{session_id}")
    if resp.status_code == 200:
        print_success("Session retrieved successfully")
    else:
        print_error(f"Failed to get session: {resp.status_code}")
    
    # List sessions
    print_info("Listing all sessions...")
    resp = requests.get(f"{BASE_URL}/sessions")
    if resp.status_code == 200:
        sessions = resp.json()
        print_success(f"Found {len(sessions)} active session(s)")
    else:
        print_error(f"Failed to list sessions: {resp.status_code}")
    
    return session_id


def test_2_file_operations(session_id):
    """Test file upload, list, download"""
    print_test("File Operations")
    
    if not session_id:
        print_error("No session available")
        return False
    
    # Upload file
    print_info("Uploading test file...")
    file_content = b"print('Hello from uploaded file!')\nprint(f'2 + 2 = {2+2}')"
    resp = requests.post(
        f"{BASE_URL}/sessions/{session_id}/files",
        files={"file": ("test_script.py", file_content, "text/x-python")}
    )
    
    if resp.status_code != 200:
        print_error(f"Failed to upload file: {resp.status_code} - {resp.text}")
        return False
    
    upload_result = resp.json()
    print_success(f"File uploaded: {upload_result['filename']} ({upload_result['size']} bytes)")
    
    # List files
    print_info("Listing files...")
    resp = requests.get(f"{BASE_URL}/sessions/{session_id}/files")
    if resp.status_code == 200:
        files = resp.json()
        print_success(f"Found {len(files)} file(s) in workspace")
        for f in files:
            print(f"  - {f['filename']} ({f['size']} bytes)")
    else:
        print_error(f"Failed to list files: {resp.status_code}")
        return False
    
    # Download file
    print_info("Downloading file...")
    resp = requests.get(f"{BASE_URL}/sessions/{session_id}/files/test_script.py")
    if resp.status_code == 200:
        if resp.content == file_content:
            print_success("File downloaded successfully and content matches")
        else:
            print_error("Downloaded file content doesn't match")
            return False
    else:
        print_error(f"Failed to download file: {resp.status_code}")
        return False
    
    return True


def test_3_session_execution(session_id):
    """Test code execution in session"""
    print_test("Session-Based Code Execution")
    
    if not session_id:
        print_error("No session available")
        return False
    
    # Execute Python code
    print_info("Executing Python code...")
    resp = requests.post(
        f"{BASE_URL}/sessions/{session_id}/execute",
        json={
            "code": "import sys\nprint(f'Python version: {sys.version}')\nprint('Execution successful!')",
            "language": "python",
            "working_dir": "/workspace"
        }
    )
    
    if resp.status_code != 200:
        print_error(f"Failed to execute code: {resp.status_code} - {resp.text}")
        return False
    
    result = resp.json()
    print_success("Code executed successfully")
    print(f"  Output: {result['stdout'].strip()}")
    print(f"  Exit code: {result['exit_code']}")
    print(f"  Execution time: {result['execution_time']}s")
    
    return True


def test_4_file_execution(session_id):
    """Test executing uploaded file"""
    print_test("Execute Uploaded File")
    
    if not session_id:
        print_error("No session available")
        return False
    
    # Execute the previously uploaded file
    print_info("Executing uploaded Python file...")
    resp = requests.post(
        f"{BASE_URL}/sessions/{session_id}/execute-file",
        json={
            "filepath": "/workspace/test_script.py",
            "language": "python",
            "args": []
        }
    )
    
    if resp.status_code != 200:
        print_error(f"Failed to execute file: {resp.status_code} - {resp.text}")
        return False
    
    result = resp.json()
    print_success("File executed successfully")
    print(f"  Output:\n{result['stdout']}")
    print(f"  Exit code: {result['exit_code']}")
    
    return True


def test_5_persistent_state(session_id):
    """Test that files persist across executions"""
    print_test("Persistent State Verification")
    
    if not session_id:
        print_error("No session available")
        return False
    
    # First execution: create a file
    print_info("Creating data file in first execution...")
    resp = requests.post(
        f"{BASE_URL}/sessions/{session_id}/execute",
        json={
            "code": "with open('/workspace/persistent.txt', 'w') as f: f.write('Data persists!')",
            "language": "python"
        }
    )
    
    if resp.status_code != 200:
        print_error(f"Failed first execution: {resp.status_code}")
        return False
    
    print_success("File created in first execution")
    
    # Second execution: read the file
    print_info("Reading file in second execution...")
    resp = requests.post(
        f"{BASE_URL}/sessions/{session_id}/execute",
        json={
            "code": "with open('/workspace/persistent.txt', 'r') as f: print(f'Read: {f.read()}')",
            "language": "python"
        }
    )
    
    if resp.status_code != 200:
        print_error(f"Failed second execution: {resp.status_code}")
        return False
    
    result = resp.json()
    if "Data persists!" in result['stdout']:
        print_success("File persisted across executions!")
        print(f"  Output: {result['stdout'].strip()}")
    else:
        print_error("File did not persist")
        return False
    
    return True


def test_6_concurrent_sessions():
    """Test creating multiple concurrent sessions"""
    print_test("Concurrent Sessions")
    
    session_ids = []
    
    # Create 3 sessions
    for i in range(3):
        print_info(f"Creating session {i+1}/3...")
        resp = requests.post(f"{BASE_URL}/sessions", json={
            "metadata": {"test": "concurrent", "index": i}
        })
        
        if resp.status_code == 201:
            session_id = resp.json()["session_id"]
            session_ids.append(session_id)
            print_success(f"Session {i+1} created: {session_id[:8]}...")
        else:
            print_error(f"Failed to create session {i+1}")
    
    # List all sessions
    resp = requests.get(f"{BASE_URL}/sessions")
    if resp.status_code == 200:
        all_sessions = resp.json()
        print_success(f"Total active sessions: {len(all_sessions)}")
    
    # Cleanup
    print_info("Cleaning up concurrent sessions...")
    for sid in session_ids:
        requests.delete(f"{BASE_URL}/sessions/{sid}")
    
    print_success("Concurrent session test completed")
    return True


def test_7_cleanup(session_id):
    """Test session cleanup"""
    print_test("Session Cleanup")
    
    if not session_id:
        print_info("No session to cleanup")
        return True
    
    print_info(f"Destroying session {session_id[:8]}...")
    resp = requests.delete(f"{BASE_URL}/sessions/{session_id}")
    
    if resp.status_code == 200:
        print_success("Session destroyed successfully")
        
        # Verify it's gone
        resp = requests.get(f"{BASE_URL}/sessions/{session_id}")
        if resp.status_code == 404:
            print_success("Session verified as destroyed")
            return True
        else:
            print_error("Session still exists after destruction")
            return False
    else:
        print_error(f"Failed to destroy session: {resp.status_code}")
        return False


def test_8_stateless_execution():
    """Test backward compatibility - stateless execution"""
    print_test("Stateless Execution (Backward Compatibility)")
    
    print_info("Executing code without session...")
    resp = requests.post(f"{BASE_URL}/execute", json={
        "code": "print('Stateless execution works!')\nprint(f'Result: {10 * 10}')",
        "language": "python"
    })
    
    if resp.status_code == 200:
        result = resp.json()
        print_success("Stateless execution successful")
        print(f"  Output: {result['stdout'].strip()}")
        print(f"  Exit code: {result['exit_code']}")
        return True
    else:
        print_error(f"Stateless execution failed: {resp.status_code}")
        return False


def main():
    """Run all verification tests"""
    print("\n" + "="*60)
    print("ENHANCED SANDBOX - MANUAL VERIFICATION")
    print("="*60)
    
    # Check if API is running
    try:
        resp = requests.get(f"{BASE_URL}/health", timeout=2)
        if resp.status_code != 200:
            print_error("API server is not healthy")
            sys.exit(1)
        print_success("API server is running")
    except requests.exceptions.RequestException as e:
        print_error(f"Cannot connect to API server at {BASE_URL}")
        print_info("Please run 'python app.py' first")
        sys.exit(1)
    
    results = {}
    session_id = None
    
    # Run all tests
    try:
        session_id = test_1_session_lifecycle()
        results['session_lifecycle'] = session_id is not None
        
        results['file_operations'] = test_2_file_operations(session_id)
        results['session_execution'] = test_3_session_execution(session_id)
        results['file_execution'] = test_4_file_execution(session_id)
        results['persistent_state'] = test_5_persistent_state(session_id)
        results['concurrent_sessions'] = test_6_concurrent_sessions()
        results['stateless_execution'] = test_8_stateless_execution()
        results['cleanup'] = test_7_cleanup(session_id)
        
    except Exception as e:
        print_error(f"Test failed with exception: {e}")
        import traceback
        traceback.print_exc()
    
    # Summary
    print("\n" + "="*60)
    print("VERIFICATION SUMMARY")
    print("="*60)
    
    passed = sum(1 for v in results.values() if v)
    total = len(results)
    
    for test, result in results.items():
        status = "✅ PASS" if result else "❌ FAIL"
        print(f"{status} - {test.replace('_', ' ').title()}")
    
    print(f"\nResults: {passed}/{total} tests passed")
    
    if passed == total:
        print_success("All verification tests passed!")
        sys.exit(0)
    else:
        print_error(f"{total - passed} test(s) failed")
        sys.exit(1)


if __name__ == "__main__":
    main()