Spaces:
Running
Running
File size: 2,736 Bytes
0646b18 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
#!/usr/bin/env python3
"""
Run the ngrok connectivity test inside an E2B sandbox.
"""
import asyncio
import os
import sys
from dotenv import load_dotenv
from e2b_code_interpreter import Sandbox
# Load environment variables
load_dotenv()
async def run_code_async(sandbox, test_code):
"""Run code in sandbox asynchronously."""
loop = asyncio.get_event_loop()
execution = await loop.run_in_executor(None, sandbox.run_code, test_code)
return execution
async def main():
# Read the test script
with open('test_e2b_ngrok.py', 'r') as f:
test_code = f.read()
print("=" * 80)
print("USING EXISTING E2B SANDBOX TO TEST NGROK CONNECTIVITY (ASYNC)")
print("=" * 80)
print()
# Get sandbox ID from environment or use default
sandbox_id = os.getenv('E2B_SANDBOX_ID')
try:
if sandbox_id:
print(f"Connecting to existing sandbox: {sandbox_id}")
sandbox = Sandbox.connect(sandbox_id)
else:
# Create E2B sandbox with the cuga-langchain template
print("Creating E2B sandbox with 'cuga-langchain' template...")
sandbox = Sandbox.create("cuga-langchain")
print("Sandbox created")
try:
print()
print("Executing test script in sandbox asynchronously...")
print("-" * 80)
# Run the test code asynchronously
execution = await run_code_async(sandbox, test_code)
# Print all output
if execution.logs.stdout:
print("STDOUT:")
for line in execution.logs.stdout:
print(line)
if execution.logs.stderr:
print("\nSTDERR:")
for line in execution.logs.stderr:
print(line)
if execution.error:
print(f"\nEXECUTION ERROR: {execution.error}")
print("-" * 80)
print()
if execution.error:
print("❌ TEST FAILED IN E2B SANDBOX")
sys.exit(1)
else:
print("✅ TEST COMPLETED IN E2B SANDBOX")
sys.exit(0)
finally:
# Only close if we created it (not if we connected to existing)
if not sandbox_id and sandbox:
print("\nClosing sandbox...")
try:
sandbox.kill()
except Exception as e:
print(f"Warning: Could not close sandbox: {e}")
except Exception as e:
print(f"\n❌ ERROR: {type(e).__name__}: {str(e)}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
|