Filip Michalsky
1 month ago
parent
b04e60ca17
commit
2d7dfca4a4
@ -0,0 +1,304 @@
|
||||
"""
|
||||
Simple tests for Stagehand Integration with Swarms
|
||||
=================================================
|
||||
|
||||
These tests verify the basic structure and functionality of the
|
||||
Stagehand integration without requiring external dependencies.
|
||||
"""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
class TestStagehandIntegrationStructure:
|
||||
"""Test that integration files have correct structure."""
|
||||
|
||||
def test_examples_directory_exists(self):
|
||||
"""Test that examples directory structure is correct."""
|
||||
import os
|
||||
|
||||
base_path = "examples/stagehand"
|
||||
assert os.path.exists(base_path)
|
||||
|
||||
expected_files = [
|
||||
"1_stagehand_wrapper_agent.py",
|
||||
"2_stagehand_tools_agent.py",
|
||||
"3_stagehand_mcp_agent.py",
|
||||
"4_stagehand_multi_agent_workflow.py",
|
||||
"README.md",
|
||||
"requirements.txt",
|
||||
]
|
||||
|
||||
for file in expected_files:
|
||||
file_path = os.path.join(base_path, file)
|
||||
assert os.path.exists(file_path), f"Missing file: {file}"
|
||||
|
||||
def test_wrapper_agent_imports(self):
|
||||
"""Test that wrapper agent has correct imports."""
|
||||
with open(
|
||||
"examples/stagehand/1_stagehand_wrapper_agent.py", "r"
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check for required imports
|
||||
assert "from swarms import Agent" in content
|
||||
assert "import asyncio" in content
|
||||
assert "import json" in content
|
||||
assert "class StagehandAgent" in content
|
||||
|
||||
def test_tools_agent_imports(self):
|
||||
"""Test that tools agent has correct imports."""
|
||||
with open(
|
||||
"examples/stagehand/2_stagehand_tools_agent.py", "r"
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check for required imports
|
||||
assert (
|
||||
"from swarms.tools.base_tool import BaseTool" in content
|
||||
)
|
||||
assert "class NavigateTool" in content
|
||||
assert "class ActTool" in content
|
||||
assert "class ExtractTool" in content
|
||||
|
||||
def test_mcp_agent_imports(self):
|
||||
"""Test that MCP agent has correct imports."""
|
||||
with open(
|
||||
"examples/stagehand/3_stagehand_mcp_agent.py", "r"
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check for required imports
|
||||
assert "from swarms import Agent" in content
|
||||
assert "class StagehandMCPAgent" in content
|
||||
assert "mcp_url" in content
|
||||
|
||||
def test_workflow_agent_imports(self):
|
||||
"""Test that workflow agent has correct imports."""
|
||||
with open(
|
||||
"examples/stagehand/4_stagehand_multi_agent_workflow.py",
|
||||
"r",
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check for required imports
|
||||
assert (
|
||||
"from swarms import Agent, SequentialWorkflow, ConcurrentWorkflow"
|
||||
in content
|
||||
)
|
||||
assert (
|
||||
"from swarms.structs.agent_rearrange import AgentRearrange"
|
||||
in content
|
||||
)
|
||||
|
||||
|
||||
class TestStagehandMockIntegration:
|
||||
"""Test Stagehand integration with mocked dependencies."""
|
||||
|
||||
def test_mock_stagehand_initialization(self):
|
||||
"""Test that Stagehand can be mocked and initialized."""
|
||||
|
||||
# Setup mock without importing actual stagehand
|
||||
mock_stagehand = MagicMock()
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.init = MagicMock()
|
||||
mock_stagehand.return_value = mock_instance
|
||||
|
||||
# Mock config creation
|
||||
config = MagicMock()
|
||||
stagehand_instance = mock_stagehand(config)
|
||||
|
||||
# Verify mock works
|
||||
assert stagehand_instance is not None
|
||||
assert hasattr(stagehand_instance, "init")
|
||||
|
||||
def test_json_serialization(self):
|
||||
"""Test JSON serialization for agent responses."""
|
||||
|
||||
# Test data that would come from browser automation
|
||||
test_data = {
|
||||
"task": "Navigate to example.com",
|
||||
"status": "completed",
|
||||
"data": {
|
||||
"navigated_to": "https://example.com",
|
||||
"extracted": ["item1", "item2"],
|
||||
"action": "navigate",
|
||||
},
|
||||
}
|
||||
|
||||
# Test serialization
|
||||
json_result = json.dumps(test_data, indent=2)
|
||||
assert isinstance(json_result, str)
|
||||
|
||||
# Test deserialization
|
||||
parsed_data = json.loads(json_result)
|
||||
assert parsed_data["task"] == "Navigate to example.com"
|
||||
assert parsed_data["status"] == "completed"
|
||||
assert len(parsed_data["data"]["extracted"]) == 2
|
||||
|
||||
def test_url_extraction_logic(self):
|
||||
"""Test URL extraction logic from task strings."""
|
||||
import re
|
||||
|
||||
# Test cases
|
||||
test_cases = [
|
||||
(
|
||||
"Navigate to https://example.com",
|
||||
["https://example.com"],
|
||||
),
|
||||
("Go to google.com and search", ["google.com"]),
|
||||
(
|
||||
"Visit https://github.com/repo",
|
||||
["https://github.com/repo"],
|
||||
),
|
||||
("Open example.org", ["example.org"]),
|
||||
]
|
||||
|
||||
url_pattern = r"https?://[^\s]+"
|
||||
domain_pattern = r"(\w+\.\w+)"
|
||||
|
||||
for task, expected in test_cases:
|
||||
# Extract full URLs
|
||||
urls = re.findall(url_pattern, task)
|
||||
|
||||
# If no full URLs, extract domains
|
||||
if not urls:
|
||||
domains = re.findall(domain_pattern, task)
|
||||
if domains:
|
||||
urls = domains
|
||||
|
||||
assert (
|
||||
len(urls) > 0
|
||||
), f"Failed to extract URL from: {task}"
|
||||
assert (
|
||||
urls[0] in expected
|
||||
), f"Expected {expected}, got {urls}"
|
||||
|
||||
|
||||
class TestSwarmsPatternsCompliance:
|
||||
"""Test compliance with Swarms framework patterns."""
|
||||
|
||||
def test_agent_inheritance_pattern(self):
|
||||
"""Test that wrapper agent follows Swarms Agent inheritance pattern."""
|
||||
|
||||
# Read the wrapper agent file
|
||||
with open(
|
||||
"examples/stagehand/1_stagehand_wrapper_agent.py", "r"
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check inheritance pattern
|
||||
assert "class StagehandAgent(SwarmsAgent):" in content
|
||||
assert "def run(self, task: str" in content
|
||||
assert "return" in content
|
||||
|
||||
def test_tools_pattern(self):
|
||||
"""Test that tools follow Swarms BaseTool pattern."""
|
||||
|
||||
# Read the tools agent file
|
||||
with open(
|
||||
"examples/stagehand/2_stagehand_tools_agent.py", "r"
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check tool pattern
|
||||
assert "class NavigateTool(BaseTool):" in content
|
||||
assert "def run(self," in content
|
||||
assert "name=" in content
|
||||
assert "description=" in content
|
||||
|
||||
def test_mcp_integration_pattern(self):
|
||||
"""Test MCP integration follows Swarms pattern."""
|
||||
|
||||
# Read the MCP agent file
|
||||
with open(
|
||||
"examples/stagehand/3_stagehand_mcp_agent.py", "r"
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check MCP pattern
|
||||
assert "mcp_url=" in content
|
||||
assert "Agent(" in content
|
||||
|
||||
def test_workflow_patterns(self):
|
||||
"""Test workflow patterns are properly used."""
|
||||
|
||||
# Read the workflow file
|
||||
with open(
|
||||
"examples/stagehand/4_stagehand_multi_agent_workflow.py",
|
||||
"r",
|
||||
) as f:
|
||||
content = f.read()
|
||||
|
||||
# Check workflow patterns
|
||||
assert "SequentialWorkflow" in content
|
||||
assert "ConcurrentWorkflow" in content
|
||||
assert "AgentRearrange" in content
|
||||
|
||||
|
||||
class TestDocumentationAndExamples:
|
||||
"""Test documentation and example completeness."""
|
||||
|
||||
def test_readme_completeness(self):
|
||||
"""Test that README contains essential information."""
|
||||
|
||||
with open("examples/stagehand/README.md", "r") as f:
|
||||
content = f.read()
|
||||
|
||||
required_sections = [
|
||||
"# Stagehand Browser Automation Integration",
|
||||
"## Overview",
|
||||
"## Examples",
|
||||
"## Setup",
|
||||
"## Use Cases",
|
||||
"## Best Practices",
|
||||
]
|
||||
|
||||
for section in required_sections:
|
||||
assert section in content, f"Missing section: {section}"
|
||||
|
||||
def test_requirements_file(self):
|
||||
"""Test that requirements file has necessary dependencies."""
|
||||
|
||||
with open("examples/stagehand/requirements.txt", "r") as f:
|
||||
content = f.read()
|
||||
|
||||
required_deps = [
|
||||
"swarms",
|
||||
"stagehand",
|
||||
"python-dotenv",
|
||||
"pydantic",
|
||||
"loguru",
|
||||
]
|
||||
|
||||
for dep in required_deps:
|
||||
assert dep in content, f"Missing dependency: {dep}"
|
||||
|
||||
def test_example_files_have_docstrings(self):
|
||||
"""Test that example files have proper docstrings."""
|
||||
|
||||
example_files = [
|
||||
"examples/stagehand/1_stagehand_wrapper_agent.py",
|
||||
"examples/stagehand/2_stagehand_tools_agent.py",
|
||||
"examples/stagehand/3_stagehand_mcp_agent.py",
|
||||
"examples/stagehand/4_stagehand_multi_agent_workflow.py",
|
||||
]
|
||||
|
||||
for file_path in example_files:
|
||||
with open(file_path, "r") as f:
|
||||
content = f.read()
|
||||
|
||||
# Check for module docstring
|
||||
assert (
|
||||
'"""' in content[:500]
|
||||
), f"Missing docstring in {file_path}"
|
||||
|
||||
# Check for main execution block
|
||||
assert (
|
||||
'if __name__ == "__main__":' in content
|
||||
), f"Missing main block in {file_path}"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Loading…
Reference in new issue