MCP Operator
by willer
Verified
- mcp-operator
- tests
#!/usr/bin/env python3
"""
End-to-end test for multi-step browser operations using a real browser.
This directly tests the operate-browser functionality with complex multi-step tasks.
"""
import os
import time
import sys
import unittest
import asyncio
from pathlib import Path
# Ensure the src directory is in the path
src_dir = Path(__file__).parent.parent / "src"
sys.path.insert(0, str(src_dir))
# Import the modules to test
from mcp_operator.browser import BrowserOperator, BrowserInstance
# Default test settings with environment variable override
import os
HEADLESS_MODE = os.environ.get("TEST_HEADLESS", "true").lower() != "false"
EXAMPLE_URL = "https://example.com"
class TestRealMultistep(unittest.TestCase):
"""Test real multi-step browser operations"""
def setUp(self):
"""Set up the test environment"""
# Create a unique project name
self.project_name = f"test-{int(time.time())}"
# Create directories needed for the test
for directory in ["logs", "screenshots", "notes"]:
os.makedirs(os.path.join(Path(__file__).parent.parent, directory), exist_ok=True)
# Initialize the event loop for async tests
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
def tearDown(self):
"""Clean up after the test"""
self.loop.close()
def test_real_multistep(self):
"""Test a real multi-step browser operation using process_message"""
# Run the async test in the loop
mode_str = "HEADLESS" if HEADLESS_MODE else "VISIBLE"
print(f"\n🔍 Running browser test in {mode_str} mode - {'browser will be invisible' if HEADLESS_MODE else 'watch your screen!'}")
print(f"💡 Tip: Set TEST_HEADLESS=false to see the browser window")
result = self.loop.run_until_complete(self._async_test_real_multistep())
self.assertTrue(result, "Multi-step test should complete successfully")
async def _async_test_real_multistep(self):
"""Async implementation of the test"""
print(f"\n🌐 Creating browser '{self.project_name}' to test multi-step operations...")
# Create and initialize the operator
operator = BrowserOperator(self.project_name)
try:
# Create the browser
result = await operator.create_browser()
job_id = result.get("job_id")
print(f"Browser created with job ID: {job_id}")
# Make sure browser_instance has headless mode set correctly
operator.browser_instance.headless = HEADLESS_MODE
print(f"Browser headless mode set to: {HEADLESS_MODE}")
# Define a simple task that works reliably
instruction = "Go to example.com and tell me the title of the page and what links are available"
print(f"🚀 Starting multi-step task: {instruction}")
# Execute the instruction
start_time = time.time()
result = await operator.process_message(instruction)
elapsed_time = time.time() - start_time
# Display results
print(f"⏱️ Task completed in {elapsed_time:.2f} seconds")
print(f"📝 Actions executed: {result.get('actions_executed', 0)}")
# Check for success
actions_executed = result.get('actions_executed', 0)
success = actions_executed > 0
# Get some result text to verify
result_text = result.get("text", "")
# Print the text for debugging
print(f"\nResult text: {result_text[:500]}...\n")
# Look for evidence of page analysis
if "Example Domain" in result_text and "More information" in result_text:
print("✅ Found expected page title and link information in result")
success = True
else:
print("❌ Did not find expected page information in result")
# Clean up
await operator.close()
print("✅ Browser closed")
return success
except Exception as e:
import traceback
print(f"❌ Error in multi-step test: {str(e)}")
traceback.print_exc()
# Try to clean up even if there's an error
try:
if operator:
await operator.close()
except:
pass
return False
if __name__ == "__main__":
unittest.main()