+ """
+
+ @staticmethod
+ def create_form_elements():
+ """Create form elements for testing"""
+ return """
+
+ """
+
+
+# Custom assertion helpers
+class AssertionHelpers:
+ """Custom assertion helpers for Stagehand testing"""
+
+ @staticmethod
+ def assert_valid_selector(selector: str):
+ """Assert selector is valid CSS/XPath"""
+ import re
+
+ # Basic CSS selector validation
+ css_pattern = r'^[#.]?[\w\-\[\]="\':\s,>+~*()]+$'
+ xpath_pattern = r'^\/\/.*$'
+
+ assert (re.match(css_pattern, selector) or
+ re.match(xpath_pattern, selector)), f"Invalid selector: {selector}"
+
+ @staticmethod
+ def assert_schema_compliance(data: dict, schema: dict):
+ """Assert data matches expected schema"""
+ import jsonschema
+
+ try:
+ jsonschema.validate(data, schema)
+ except jsonschema.ValidationError as e:
+ pytest.fail(f"Data does not match schema: {e.message}")
+
+ @staticmethod
+ def assert_act_result_valid(result: ActResult):
+ """Assert ActResult is valid"""
+ assert isinstance(result, ActResult)
+ assert isinstance(result.success, bool)
+ assert isinstance(result.message, str)
+ assert isinstance(result.action, str)
+
+ @staticmethod
+ def assert_observe_results_valid(results: list[ObserveResult]):
+ """Assert ObserveResult list is valid"""
+ assert isinstance(results, list)
+ for result in results:
+ assert isinstance(result, ObserveResult)
+ assert isinstance(result.selector, str)
+ assert isinstance(result.description, str)
+
+
+@pytest.fixture
+def assertion_helpers():
+ """Provide assertion helpers"""
+ return AssertionHelpers()
+
+
+@pytest.fixture
+def test_data_generator():
+ """Provide test data generator"""
+ return TestDataGenerator()
diff --git a/tests/e2e/test_act_integration.py b/tests/e2e/test_act_integration.py
new file mode 100644
index 0000000..c6eb4d4
--- /dev/null
+++ b/tests/e2e/test_act_integration.py
@@ -0,0 +1,391 @@
+"""
+Integration tests for Stagehand act functionality.
+
+These tests are inspired by the act evals and test the page.act() functionality
+for performing actions and interactions in both LOCAL and BROWSERBASE modes.
+"""
+
+import asyncio
+import os
+import pytest
+import pytest_asyncio
+from typing import List, Dict, Any
+
+from stagehand import Stagehand, StagehandConfig
+
+
+class TestActIntegration:
+ """Integration tests for Stagehand act functionality"""
+
+ @pytest.fixture(scope="class")
+ def local_config(self):
+ """Configuration for LOCAL mode testing"""
+ return StagehandConfig(
+ env="LOCAL",
+ model_name="gpt-4o-mini",
+ headless=True,
+ verbose=1,
+ dom_settle_timeout_ms=2000,
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest.fixture(scope="class")
+ def browserbase_config(self):
+ """Configuration for BROWSERBASE mode testing"""
+ return StagehandConfig(
+ env="BROWSERBASE",
+ api_key=os.getenv("BROWSERBASE_API_KEY"),
+ project_id=os.getenv("BROWSERBASE_PROJECT_ID"),
+ model_name="gpt-4o",
+ headless=False,
+ verbose=2,
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest_asyncio.fixture
+ async def local_stagehand(self, local_config):
+ """Create a Stagehand instance for LOCAL testing"""
+ stagehand = Stagehand(config=local_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest_asyncio.fixture
+ async def browserbase_stagehand(self, browserbase_config):
+ """Create a Stagehand instance for BROWSERBASE testing"""
+ if not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")):
+ pytest.skip("Browserbase credentials not available")
+
+ stagehand = Stagehand(config=browserbase_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_form_filling_local(self, local_stagehand):
+ """Test form filling capabilities similar to act_form_filling eval in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Fill various form fields
+ await stagehand.page.act("Fill the customer name field with 'John Doe'")
+ await stagehand.page.act("Fill the telephone field with '555-0123'")
+ await stagehand.page.act("Fill the email field with 'john@example.com'")
+
+ # Verify fields were filled by observing their values
+ filled_name = await stagehand.page.observe("Find the customer name input field")
+ assert filled_name is not None
+ assert len(filled_name) > 0
+
+ # Test dropdown/select interaction
+ await stagehand.page.act("Select 'Large' from the size dropdown")
+
+ # Test checkbox interaction
+ await stagehand.page.act("Check the 'I accept the terms' checkbox")
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_form_filling_browserbase(self, browserbase_stagehand):
+ """Test form filling capabilities similar to act_form_filling eval in BROWSERBASE mode"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Fill various form fields
+ await stagehand.page.act("Fill the customer name field with 'Jane Smith'")
+ await stagehand.page.act("Fill the telephone field with '555-0456'")
+ await stagehand.page.act("Fill the email field with 'jane@example.com'")
+
+ # Verify fields were filled
+ filled_name = await stagehand.page.observe("Find the customer name input field")
+ assert filled_name is not None
+ assert len(filled_name) > 0
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_button_clicking_local(self, local_stagehand):
+ """Test button clicking functionality in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with buttons
+ await stagehand.page.goto("https://httpbin.org")
+
+ # Test clicking various button types
+ # Find and click a navigation button/link
+ buttons = await stagehand.page.observe("Find all clickable buttons or links")
+ assert buttons is not None
+
+ if buttons and len(buttons) > 0:
+ # Try clicking the first button found
+ await stagehand.page.act("Click the first button or link on the page")
+
+ # Wait for any page changes
+ await asyncio.sleep(2)
+
+ # Verify we're still on a valid page
+ new_elements = await stagehand.page.observe("Find any elements on the current page")
+ assert new_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_navigation_actions_local(self, local_stagehand):
+ """Test navigation actions in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Start at example.com
+ await stagehand.page.goto("https://example.com")
+
+ # Test link clicking for navigation
+ links = await stagehand.page.observe("Find all links on the page")
+
+ if links and len(links) > 0:
+ # Click on a link to navigate
+ await stagehand.page.act("Click on the 'More information...' link")
+ await asyncio.sleep(2)
+
+ # Verify navigation occurred
+ current_elements = await stagehand.page.observe("Find the main content on this page")
+ assert current_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_search_workflow_local(self, local_stagehand):
+ """Test search workflow similar to google_jobs eval in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to Google
+ await stagehand.page.goto("https://www.google.com")
+
+ # Perform search actions
+ await stagehand.page.act("Type 'python programming' in the search box")
+ await stagehand.page.act("Press Enter to search")
+
+ # Wait for results
+ await asyncio.sleep(3)
+
+ # Verify search results appeared
+ results = await stagehand.page.observe("Find search result links")
+ assert results is not None
+
+ # Test interacting with search results
+ if results and len(results) > 0:
+ await stagehand.page.act("Click on the first search result")
+ await asyncio.sleep(2)
+
+ # Verify we navigated to a result page
+ content = await stagehand.page.observe("Find the main content of this page")
+ assert content is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_text_input_actions_local(self, local_stagehand):
+ """Test various text input actions in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Test different text input scenarios
+ await stagehand.page.act("Clear the customer name field and type 'Test User'")
+ await stagehand.page.act("Fill the comments field with 'This is a test comment with special characters: @#$%'")
+
+ # Test text modification actions
+ await stagehand.page.act("Select all text in the comments field")
+ await stagehand.page.act("Type 'Replaced text' to replace the selected text")
+
+ # Verify text actions worked
+ filled_fields = await stagehand.page.observe("Find all filled form fields")
+ assert filled_fields is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_keyboard_actions_local(self, local_stagehand):
+ """Test keyboard actions in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to Google for keyboard testing
+ await stagehand.page.goto("https://www.google.com")
+
+ # Test various keyboard actions
+ await stagehand.page.act("Click on the search box")
+ await stagehand.page.act("Type 'hello world'")
+ await stagehand.page.act("Press Ctrl+A to select all")
+ await stagehand.page.act("Press Delete to clear the field")
+ await stagehand.page.act("Type 'new search term'")
+ await stagehand.page.act("Press Enter")
+
+ # Wait for search results
+ await asyncio.sleep(3)
+
+ # Verify keyboard actions resulted in search
+ results = await stagehand.page.observe("Find search results")
+ assert results is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_mouse_actions_local(self, local_stagehand):
+ """Test mouse actions in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with various clickable elements
+ await stagehand.page.goto("https://httpbin.org")
+
+ # Test different mouse actions
+ await stagehand.page.act("Right-click on the main heading")
+ await stagehand.page.act("Click outside the page to dismiss any context menu")
+ await stagehand.page.act("Double-click on the main heading")
+
+ # Test hover actions
+ links = await stagehand.page.observe("Find all links on the page")
+ if links and len(links) > 0:
+ await stagehand.page.act("Hover over the first link")
+ await asyncio.sleep(1)
+ await stagehand.page.act("Click the hovered link")
+ await asyncio.sleep(2)
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_complex_form_workflow_local(self, local_stagehand):
+ """Test complex form workflow in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a comprehensive form
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Complete multi-step form filling
+ await stagehand.page.act("Fill the customer name field with 'Integration Test User'")
+ await stagehand.page.act("Fill the telephone field with '+1-555-123-4567'")
+ await stagehand.page.act("Fill the email field with 'integration.test@example.com'")
+ await stagehand.page.act("Select 'Medium' from the size dropdown if available")
+ await stagehand.page.act("Fill the comments field with 'This is an automated integration test submission'")
+
+ # Submit the form
+ await stagehand.page.act("Click the Submit button")
+
+ # Wait for submission and verify
+ await asyncio.sleep(3)
+
+ # Check if form was submitted (page changed or success message)
+ result_content = await stagehand.page.observe("Find any confirmation or result content")
+ assert result_content is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_error_recovery_local(self, local_stagehand):
+ """Test error recovery in act operations in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a simple page
+ await stagehand.page.goto("https://example.com")
+
+ # Test acting on non-existent elements (should handle gracefully)
+ try:
+ await stagehand.page.act("Click the non-existent button with id 'impossible-button-12345'")
+ # If it doesn't raise an exception, that's also acceptable
+ except Exception:
+ # Expected for non-existent elements
+ pass
+
+ # Verify page is still functional after error
+ elements = await stagehand.page.observe("Find any elements on the page")
+ assert elements is not None
+
+ # Test successful action after failed attempt
+ await stagehand.page.act("Click on the main heading of the page")
+
+ @pytest.mark.asyncio
+ @pytest.mark.slow
+ @pytest.mark.local
+ async def test_performance_multiple_actions_local(self, local_stagehand):
+ """Test performance of multiple sequential actions in LOCAL mode"""
+ import time
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Time multiple actions
+ start_time = time.time()
+
+ await stagehand.page.act("Fill the customer name field with 'Speed Test'")
+ await stagehand.page.act("Fill the telephone field with '555-SPEED'")
+ await stagehand.page.act("Fill the email field with 'speed@test.com'")
+ await stagehand.page.act("Click in the comments field")
+ await stagehand.page.act("Type 'Performance testing in progress'")
+
+ total_time = time.time() - start_time
+
+ # Multiple actions should complete within reasonable time
+ assert total_time < 120.0 # 2 minutes for 5 actions
+
+ # Verify all actions were successful
+ filled_fields = await stagehand.page.observe("Find all filled form fields")
+ assert filled_fields is not None
+ assert len(filled_fields) > 0
+
+ @pytest.mark.asyncio
+ @pytest.mark.e2e
+ @pytest.mark.local
+ async def test_end_to_end_user_journey_local(self, local_stagehand):
+ """End-to-end test simulating complete user journey in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Step 1: Start at homepage
+ await stagehand.page.goto("https://httpbin.org")
+
+ # Step 2: Navigate to forms section
+ await stagehand.page.act("Click on any link that leads to forms or testing")
+ await asyncio.sleep(2)
+
+ # Step 3: Fill out a form completely
+ forms = await stagehand.page.observe("Find any form elements")
+ if forms and len(forms) > 0:
+ # Navigate to forms page if not already there
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Complete the form
+ await stagehand.page.act("Fill the customer name field with 'E2E Test User'")
+ await stagehand.page.act("Fill the telephone field with '555-E2E-TEST'")
+ await stagehand.page.act("Fill the email field with 'e2e@test.com'")
+ await stagehand.page.act("Fill the comments with 'End-to-end integration test'")
+
+ # Submit the form
+ await stagehand.page.act("Click the Submit button")
+ await asyncio.sleep(3)
+
+ # Verify successful completion
+ result = await stagehand.page.observe("Find any result or confirmation content")
+ assert result is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_browserbase_specific_actions(self, browserbase_stagehand):
+ """Test Browserbase-specific action capabilities"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Test actions in Browserbase environment
+ await stagehand.page.act("Fill the customer name field with 'Browserbase Test'")
+ await stagehand.page.act("Fill the email field with 'browserbase@test.com'")
+
+ # Verify actions worked
+ filled_fields = await stagehand.page.observe("Find filled form fields")
+ assert filled_fields is not None
+
+ # Verify Browserbase session is active
+ assert hasattr(stagehand, 'session_id')
+ assert stagehand.session_id is not None
\ No newline at end of file
diff --git a/tests/e2e/test_extract_integration.py b/tests/e2e/test_extract_integration.py
new file mode 100644
index 0000000..d88b51a
--- /dev/null
+++ b/tests/e2e/test_extract_integration.py
@@ -0,0 +1,482 @@
+"""
+Integration tests for Stagehand extract functionality.
+
+These tests are inspired by the extract evals and test the page.extract() functionality
+for extracting structured data from web pages in both LOCAL and BROWSERBASE modes.
+"""
+
+import asyncio
+import os
+import pytest
+import pytest_asyncio
+from typing import List, Dict, Any
+from pydantic import BaseModel, Field, HttpUrl
+
+from stagehand import Stagehand, StagehandConfig
+from stagehand.schemas import ExtractOptions
+
+
+class Article(BaseModel):
+ """Schema for article extraction tests"""
+ title: str = Field(..., description="The title of the article")
+ summary: str = Field(None, description="A brief summary or description of the article")
+ author: str = Field(None, description="The author of the article")
+ date: str = Field(None, description="The publication date")
+ url: HttpUrl = Field(None, description="The URL of the article")
+
+
+class Articles(BaseModel):
+ """Schema for multiple articles extraction"""
+ articles: List[Article] = Field(..., description="List of articles extracted from the page")
+
+
+class PressRelease(BaseModel):
+ """Schema for press release extraction tests"""
+ title: str = Field(..., description="The title of the press release")
+ date: str = Field(..., description="The publication date")
+ content: str = Field(..., description="The main content or summary")
+ company: str = Field(None, description="The company name")
+
+
+class SearchResult(BaseModel):
+ """Schema for search result extraction"""
+ title: str = Field(..., description="The title of the search result")
+ url: HttpUrl = Field(..., description="The URL of the search result")
+ snippet: str = Field(None, description="The snippet or description")
+
+
+class FormData(BaseModel):
+ """Schema for form data extraction"""
+ customer_name: str = Field(None, description="Customer name field value")
+ telephone: str = Field(None, description="Telephone field value")
+ email: str = Field(None, description="Email field value")
+ comments: str = Field(None, description="Comments field value")
+
+
+class TestExtractIntegration:
+ """Integration tests for Stagehand extract functionality"""
+
+ @pytest.fixture(scope="class")
+ def local_config(self):
+ """Configuration for LOCAL mode testing"""
+ return StagehandConfig(
+ env="LOCAL",
+ model_name="gpt-4o-mini",
+ headless=True,
+ verbose=1,
+ dom_settle_timeout_ms=2000,
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest.fixture(scope="class")
+ def browserbase_config(self):
+ """Configuration for BROWSERBASE mode testing"""
+ return StagehandConfig(
+ env="BROWSERBASE",
+ api_key=os.getenv("BROWSERBASE_API_KEY"),
+ project_id=os.getenv("BROWSERBASE_PROJECT_ID"),
+ model_name="gpt-4o",
+ headless=False,
+ verbose=2,
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest_asyncio.fixture
+ async def local_stagehand(self, local_config):
+ """Create a Stagehand instance for LOCAL testing"""
+ stagehand = Stagehand(config=local_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest_asyncio.fixture
+ async def browserbase_stagehand(self, browserbase_config):
+ """Create a Stagehand instance for BROWSERBASE testing"""
+ if not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")):
+ pytest.skip("Browserbase credentials not available")
+
+ stagehand = Stagehand(config=browserbase_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_news_articles_local(self, local_stagehand):
+ """Test extracting news articles similar to extract_news_articles eval in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a news site
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Test simple string-based extraction
+ titles_text = await stagehand.page.extract(
+ "Extract the titles of the first 5 articles on the page as a JSON array"
+ )
+ assert titles_text is not None
+
+ # Test schema-based extraction
+ extract_options = ExtractOptions(
+ instruction="Extract the first article's title, summary, and any available metadata",
+ schema_definition=Article
+ )
+
+ article_data = await stagehand.page.extract(extract_options)
+ assert article_data is not None
+
+ # Validate the extracted data structure
+ if hasattr(article_data, 'data') and article_data.data:
+ # BROWSERBASE mode format
+ article = Article.model_validate(article_data.data)
+ assert article.title
+ elif hasattr(article_data, 'title'):
+ # LOCAL mode format
+ article = Article.model_validate(article_data.model_dump())
+ assert article.title
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_extract_news_articles_browserbase(self, browserbase_stagehand):
+ """Test extracting news articles similar to extract_news_articles eval in BROWSERBASE mode"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a news site
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Test schema-based extraction
+ extract_options = ExtractOptions(
+ instruction="Extract the first article's title, summary, and any available metadata",
+ schema_definition=Article
+ )
+
+ article_data = await stagehand.page.extract(extract_options)
+ assert article_data is not None
+
+ # Validate the extracted data structure
+ if hasattr(article_data, 'data') and article_data.data:
+ article = Article.model_validate(article_data.data)
+ assert article.title
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_multiple_articles_local(self, local_stagehand):
+ """Test extracting multiple articles in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a news site
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Extract multiple articles with schema
+ extract_options = ExtractOptions(
+ instruction="Extract the top 3 articles with their titles and any available metadata",
+ schema_definition=Articles
+ )
+
+ articles_data = await stagehand.page.extract(extract_options)
+ assert articles_data is not None
+
+ # Validate the extracted data
+ if hasattr(articles_data, 'data') and articles_data.data:
+ articles = Articles.model_validate(articles_data.data)
+ assert len(articles.articles) > 0
+ for article in articles.articles:
+ assert article.title
+ elif hasattr(articles_data, 'articles'):
+ articles = Articles.model_validate(articles_data.model_dump())
+ assert len(articles.articles) > 0
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_search_results_local(self, local_stagehand):
+ """Test extracting search results in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to Google and perform a search
+ await stagehand.page.goto("https://www.google.com")
+ await stagehand.page.act("Type 'python programming' in the search box")
+ await stagehand.page.act("Press Enter")
+
+ # Wait for results
+ await asyncio.sleep(3)
+
+ # Extract search results
+ search_results = await stagehand.page.extract(
+ "Extract the first 3 search results with their titles, URLs, and snippets as a JSON array"
+ )
+
+ assert search_results is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_form_data_local(self, local_stagehand):
+ """Test extracting form data after filling it in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Fill the form first
+ await stagehand.page.act("Fill the customer name field with 'Extract Test User'")
+ await stagehand.page.act("Fill the telephone field with '555-EXTRACT'")
+ await stagehand.page.act("Fill the email field with 'extract@test.com'")
+ await stagehand.page.act("Fill the comments field with 'Testing extraction functionality'")
+
+ # Extract the form data
+ extract_options = ExtractOptions(
+ instruction="Extract all the filled form field values",
+ schema_definition=FormData
+ )
+
+ form_data = await stagehand.page.extract(extract_options)
+ assert form_data is not None
+
+ # Validate extracted form data
+ if hasattr(form_data, 'data') and form_data.data:
+ data = FormData.model_validate(form_data.data)
+ assert data.customer_name or data.email # At least one field should be extracted
+ elif hasattr(form_data, 'customer_name'):
+ data = FormData.model_validate(form_data.model_dump())
+ assert data.customer_name or data.email
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_structured_content_local(self, local_stagehand):
+ """Test extracting structured content from complex pages in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with structured content
+ await stagehand.page.goto("https://httpbin.org")
+
+ # Extract page structure information
+ page_info = await stagehand.page.extract(
+ "Extract the main sections and navigation elements of this page as structured JSON"
+ )
+
+ assert page_info is not None
+
+ # Extract specific elements
+ navigation_data = await stagehand.page.extract(
+ "Extract all the navigation links with their text and destinations"
+ )
+
+ assert navigation_data is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_table_data_local(self, local_stagehand):
+ """Test extracting tabular data in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with tables (using HTTP status codes page)
+ await stagehand.page.goto("https://httpbin.org/status/200")
+
+ # Extract any structured data available
+ structured_data = await stagehand.page.extract(
+ "Extract any structured data, lists, or key-value pairs from this page"
+ )
+
+ assert structured_data is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_metadata_local(self, local_stagehand):
+ """Test extracting page metadata in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with rich metadata
+ await stagehand.page.goto("https://example.com")
+
+ # Extract page metadata
+ metadata = await stagehand.page.extract(
+ "Extract the page title, description, and any other metadata"
+ )
+
+ assert metadata is not None
+
+ # Extract specific content
+ content_info = await stagehand.page.extract(
+ "Extract the main heading and paragraph content from this page"
+ )
+
+ assert content_info is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_error_handling_local(self, local_stagehand):
+ """Test extract error handling in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a simple page
+ await stagehand.page.goto("https://example.com")
+
+ # Test extracting non-existent data
+ nonexistent_data = await stagehand.page.extract(
+ "Extract all purple elephants and unicorns from this page"
+ )
+ # Should return something (even if empty) rather than crash
+ assert nonexistent_data is not None
+
+ # Test with very specific schema that might not match
+ class ImpossibleSchema(BaseModel):
+ unicorn_name: str = Field(..., description="Name of the unicorn")
+ magic_level: int = Field(..., description="Level of magic")
+
+ try:
+ extract_options = ExtractOptions(
+ instruction="Extract unicorn information",
+ schema_definition=ImpossibleSchema
+ )
+ impossible_data = await stagehand.page.extract(extract_options)
+ # If it doesn't crash, that's acceptable
+ assert impossible_data is not None
+ except Exception:
+ # Expected for impossible schemas
+ pass
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_json_validation_local(self, local_stagehand):
+ """Test that extracted data validates against schemas in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a content-rich page
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Define a strict schema
+ class StrictArticle(BaseModel):
+ title: str = Field(..., description="Article title", min_length=1)
+ has_content: bool = Field(..., description="Whether the article has visible content")
+
+ extract_options = ExtractOptions(
+ instruction="Extract the first article with its title and whether it has content",
+ schema_definition=StrictArticle
+ )
+
+ article_data = await stagehand.page.extract(extract_options)
+ assert article_data is not None
+
+ # Validate against the strict schema
+ if hasattr(article_data, 'data') and article_data.data:
+ strict_article = StrictArticle.model_validate(article_data.data)
+ assert len(strict_article.title) > 0
+ assert isinstance(strict_article.has_content, bool)
+
+ @pytest.mark.asyncio
+ @pytest.mark.slow
+ @pytest.mark.local
+ async def test_extract_performance_local(self, local_stagehand):
+ """Test extract performance characteristics in LOCAL mode"""
+ import time
+ stagehand = local_stagehand
+
+ # Navigate to a content-rich page
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Time simple extraction
+ start_time = time.time()
+ simple_extract = await stagehand.page.extract(
+ "Extract the titles of the first 3 articles"
+ )
+ simple_time = time.time() - start_time
+
+ assert simple_time < 30.0 # Should complete within 30 seconds
+ assert simple_extract is not None
+
+ # Time schema-based extraction
+ start_time = time.time()
+ extract_options = ExtractOptions(
+ instruction="Extract the first article with metadata",
+ schema_definition=Article
+ )
+ schema_extract = await stagehand.page.extract(extract_options)
+ schema_time = time.time() - start_time
+
+ assert schema_time < 45.0 # Schema extraction might take a bit longer
+ assert schema_extract is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.e2e
+ @pytest.mark.local
+ async def test_extract_end_to_end_workflow_local(self, local_stagehand):
+ """End-to-end test combining actions and extraction in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Step 1: Navigate and search
+ await stagehand.page.goto("https://www.google.com")
+ await stagehand.page.act("Type 'news python programming' in the search box")
+ await stagehand.page.act("Press Enter")
+ await asyncio.sleep(3)
+
+ # Step 2: Extract search results
+ search_results = await stagehand.page.extract(
+ "Extract the first 3 search results with titles and URLs"
+ )
+ assert search_results is not None
+
+ # Step 3: Navigate to first result (if available)
+ first_result = await stagehand.page.observe("Find the first search result link")
+ if first_result and len(first_result) > 0:
+ await stagehand.page.act("Click on the first search result")
+ await asyncio.sleep(3)
+
+ # Step 4: Extract content from the result page
+ page_content = await stagehand.page.extract(
+ "Extract the main title and content summary from this page"
+ )
+ assert page_content is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extract_with_text_extract_mode_local(self, local_stagehand):
+ """Test extraction with text extract mode in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a content page
+ await stagehand.page.goto("https://example.com")
+
+ # Test text-based extraction (no schema)
+ text_content = await stagehand.page.extract(
+ "Extract all the text content from this page as plain text"
+ )
+ assert text_content is not None
+
+ # Test structured text extraction
+ structured_text = await stagehand.page.extract(
+ "Extract the heading and paragraph text as separate fields in JSON format"
+ )
+ assert structured_text is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_extract_browserbase_specific_features(self, browserbase_stagehand):
+ """Test Browserbase-specific extract capabilities"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a content-rich page
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Test extraction in Browserbase environment
+ extract_options = ExtractOptions(
+ instruction="Extract the first 2 articles with all available metadata",
+ schema_definition=Articles
+ )
+
+ articles_data = await stagehand.page.extract(extract_options)
+ assert articles_data is not None
+
+ # Verify Browserbase session is active
+ assert hasattr(stagehand, 'session_id')
+ assert stagehand.session_id is not None
+
+ # Validate the extracted data structure (Browserbase format)
+ if hasattr(articles_data, 'data') and articles_data.data:
+ articles = Articles.model_validate(articles_data.data)
+ assert len(articles.articles) > 0
\ No newline at end of file
diff --git a/tests/e2e/test_observe_integration.py b/tests/e2e/test_observe_integration.py
new file mode 100644
index 0000000..7e143d3
--- /dev/null
+++ b/tests/e2e/test_observe_integration.py
@@ -0,0 +1,329 @@
+"""
+Integration tests for Stagehand observe functionality.
+
+These tests are inspired by the observe evals and test the page.observe() functionality
+for finding and identifying elements on web pages in both LOCAL and BROWSERBASE modes.
+"""
+
+import asyncio
+import os
+import pytest
+import pytest_asyncio
+from typing import List, Dict, Any
+
+from stagehand import Stagehand, StagehandConfig
+
+
+class TestObserveIntegration:
+ """Integration tests for Stagehand observe functionality"""
+
+ @pytest.fixture(scope="class")
+ def local_config(self):
+ """Configuration for LOCAL mode testing"""
+ return StagehandConfig(
+ env="LOCAL",
+ model_name="gpt-4o-mini",
+ headless=True,
+ verbose=1,
+ dom_settle_timeout_ms=2000,
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest.fixture(scope="class")
+ def browserbase_config(self):
+ """Configuration for BROWSERBASE mode testing"""
+ return StagehandConfig(
+ env="BROWSERBASE",
+ api_key=os.getenv("BROWSERBASE_API_KEY"),
+ project_id=os.getenv("BROWSERBASE_PROJECT_ID"),
+ model_name="gpt-4o",
+ headless=False,
+ verbose=2,
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest_asyncio.fixture
+ async def local_stagehand(self, local_config):
+ """Create a Stagehand instance for LOCAL testing"""
+ stagehand = Stagehand(config=local_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest_asyncio.fixture
+ async def browserbase_stagehand(self, browserbase_config):
+ """Create a Stagehand instance for BROWSERBASE testing"""
+ if not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")):
+ pytest.skip("Browserbase credentials not available")
+
+ stagehand = Stagehand(config=browserbase_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_form_elements_local(self, local_stagehand):
+ """Test observing form elements similar to observe_taxes eval in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Observe form input elements
+ observations = await stagehand.page.observe("Find all form input elements")
+
+ # Verify observations
+ assert observations is not None
+ assert len(observations) > 0
+
+ # Check observation structure
+ for obs in observations:
+ assert hasattr(obs, "selector")
+ assert obs.selector # Not empty
+
+ # Test finding specific labeled elements
+ labeled_observations = await stagehand.page.observe("Find all form elements with labels")
+ assert labeled_observations is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_observe_form_elements_browserbase(self, browserbase_stagehand):
+ """Test observing form elements similar to observe_taxes eval in BROWSERBASE mode"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Observe form input elements
+ observations = await stagehand.page.observe("Find all form input elements")
+
+ # Verify observations
+ assert observations is not None
+ assert len(observations) > 0
+
+ # Check observation structure
+ for obs in observations:
+ assert hasattr(obs, "selector")
+ assert obs.selector # Not empty
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_search_results_local(self, local_stagehand):
+ """Test observing search results similar to observe_search_results eval in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to Google
+ await stagehand.page.goto("https://www.google.com")
+
+ # Find search box
+ search_box = await stagehand.page.observe("Find the search input field")
+ assert search_box is not None
+ assert len(search_box) > 0
+
+ # Perform search
+ await stagehand.page.act("Type 'python' in the search box")
+ await stagehand.page.act("Press Enter")
+
+ # Wait for results
+ await asyncio.sleep(3)
+
+ # Observe search results
+ results = await stagehand.page.observe("Find all search result links")
+ assert results is not None
+ # Note: Results may vary, so we just check that we got some response
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_navigation_elements_local(self, local_stagehand):
+ """Test observing navigation elements in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a site with navigation
+ await stagehand.page.goto("https://example.com")
+
+ # Observe all links
+ links = await stagehand.page.observe("Find all links on the page")
+ assert links is not None
+
+ # Observe clickable elements
+ clickable = await stagehand.page.observe("Find all clickable elements")
+ assert clickable is not None
+
+ # Test specific element observation
+ specific_elements = await stagehand.page.observe("Find the main heading on the page")
+ assert specific_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_complex_selectors_local(self, local_stagehand):
+ """Test observing elements with complex selectors in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with various elements
+ await stagehand.page.goto("https://httpbin.org")
+
+ # Test observing by element type
+ buttons = await stagehand.page.observe("Find all buttons on the page")
+ assert buttons is not None
+
+ # Test observing by text content
+ text_elements = await stagehand.page.observe("Find elements containing the word 'testing'")
+ assert text_elements is not None
+
+ # Test observing by position/layout
+ visible_elements = await stagehand.page.observe("Find all visible interactive elements")
+ assert visible_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_element_validation_local(self, local_stagehand):
+ """Test that observed elements can be interacted with in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Observe form elements
+ form_elements = await stagehand.page.observe("Find all input fields in the form")
+ assert form_elements is not None
+ assert len(form_elements) > 0
+
+ # Validate that we can get element info for each observed element
+ for element in form_elements[:3]: # Test first 3 to avoid timeout
+ selector = element.selector
+ if selector:
+ try:
+ # Try to check if element exists and is visible
+ element_info = await stagehand.page.locator(selector).first.is_visible()
+ # Element should be found (visible or not)
+ assert element_info is not None
+ except Exception:
+ # Some elements might not be accessible, which is okay
+ pass
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_accessibility_features_local(self, local_stagehand):
+ """Test observing elements by accessibility features in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page with labels
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Observe by accessibility labels
+ labeled_elements = await stagehand.page.observe("Find all form fields with proper labels")
+ assert labeled_elements is not None
+
+ # Observe interactive elements
+ interactive = await stagehand.page.observe("Find all interactive elements accessible to screen readers")
+ assert interactive is not None
+
+ # Test role-based observation
+ form_controls = await stagehand.page.observe("Find all form control elements")
+ assert form_controls is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_observe_error_handling_local(self, local_stagehand):
+ """Test observe error handling in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a simple page
+ await stagehand.page.goto("https://example.com")
+
+ # Test observing non-existent elements
+ nonexistent = await stagehand.page.observe("Find elements with class 'nonexistent-class-12345'")
+ # Should return empty list or None, not crash
+ assert nonexistent is not None or nonexistent == []
+
+ # Test with ambiguous instructions
+ ambiguous = await stagehand.page.observe("Find stuff")
+ assert ambiguous is not None
+
+ # Test with very specific instructions that might not match
+ specific = await stagehand.page.observe("Find a purple button with the text 'Impossible Button'")
+ assert specific is not None or specific == []
+
+ @pytest.mark.asyncio
+ @pytest.mark.slow
+ @pytest.mark.local
+ async def test_observe_performance_local(self, local_stagehand):
+ """Test observe performance characteristics in LOCAL mode"""
+ import time
+ stagehand = local_stagehand
+
+ # Navigate to a complex page
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Time observation operation
+ start_time = time.time()
+ observations = await stagehand.page.observe("Find all story titles on the page")
+ observation_time = time.time() - start_time
+
+ # Should complete within reasonable time
+ assert observation_time < 30.0 # 30 seconds max
+ assert observations is not None
+
+ # Test multiple rapid observations
+ start_time = time.time()
+ await stagehand.page.observe("Find all links")
+ await stagehand.page.observe("Find all comments")
+ await stagehand.page.observe("Find the navigation")
+ total_time = time.time() - start_time
+
+ # Multiple observations should still be reasonable
+ assert total_time < 120.0 # 2 minutes max for 3 operations
+
+ @pytest.mark.asyncio
+ @pytest.mark.e2e
+ @pytest.mark.local
+ async def test_observe_end_to_end_workflow_local(self, local_stagehand):
+ """End-to-end test with observe as part of larger workflow in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a news site
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Step 1: Observe the page structure
+ structure = await stagehand.page.observe("Find the main content areas")
+ assert structure is not None
+
+ # Step 2: Observe specific content
+ stories = await stagehand.page.observe("Find the first 5 story titles")
+ assert stories is not None
+
+ # Step 3: Use observation results to guide next actions
+ if stories and len(stories) > 0:
+ # Try to interact with the first story
+ await stagehand.page.act("Click on the first story title")
+ await asyncio.sleep(2)
+
+ # Observe elements on the new page
+ new_page_elements = await stagehand.page.observe("Find the main content of this page")
+ assert new_page_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_observe_browserbase_specific_features(self, browserbase_stagehand):
+ """Test Browserbase-specific observe features"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a page
+ await stagehand.page.goto("https://example.com")
+
+ # Test observe with Browserbase capabilities
+ observations = await stagehand.page.observe("Find all interactive elements on the page")
+ assert observations is not None
+
+ # Verify we can access Browserbase session info
+ assert hasattr(stagehand, 'session_id')
+ assert stagehand.session_id is not None
\ No newline at end of file
diff --git a/tests/e2e/test_stagehand_integration.py b/tests/e2e/test_stagehand_integration.py
new file mode 100644
index 0000000..0150cfa
--- /dev/null
+++ b/tests/e2e/test_stagehand_integration.py
@@ -0,0 +1,454 @@
+"""
+Integration tests for Stagehand Python SDK.
+
+These tests verify the end-to-end functionality of Stagehand in both LOCAL and BROWSERBASE modes.
+Inspired by the evals and examples in the project.
+"""
+
+import asyncio
+import os
+import pytest
+import pytest_asyncio
+from typing import Dict, Any
+from pydantic import BaseModel, Field, HttpUrl
+
+from stagehand import Stagehand, StagehandConfig
+from stagehand.schemas import ExtractOptions
+
+
+class Company(BaseModel):
+ """Schema for company extraction tests"""
+ name: str = Field(..., description="The name of the company")
+ url: HttpUrl = Field(..., description="The URL of the company website or relevant page")
+
+
+class Companies(BaseModel):
+ """Schema for companies list extraction tests"""
+ companies: list[Company] = Field(..., description="List of companies extracted from the page, maximum of 5 companies")
+
+
+class NewsArticle(BaseModel):
+ """Schema for news article extraction tests"""
+ title: str = Field(..., description="The title of the article")
+ summary: str = Field(..., description="A brief summary of the article")
+ author: str = Field(None, description="The author of the article")
+ date: str = Field(None, description="The publication date")
+
+
+class TestStagehandIntegration:
+ """
+ Integration tests for Stagehand Python SDK.
+
+ These tests verify the complete workflow of Stagehand operations
+ including initialization, navigation, observation, action, and extraction.
+ """
+
+ @pytest.fixture(scope="class")
+ def local_config(self):
+ """Configuration for LOCAL mode testing"""
+ return StagehandConfig(
+ env="LOCAL",
+ model_name="gpt-4o-mini",
+ headless=True, # Use headless mode for CI
+ verbose=1,
+ dom_settle_timeout_ms=2000,
+ self_heal=True,
+ wait_for_captcha_solves=False,
+ system_prompt="You are a browser automation assistant for testing purposes.",
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest.fixture(scope="class")
+ def browserbase_config(self):
+ """Configuration for BROWSERBASE mode testing"""
+ return StagehandConfig(
+ env="BROWSERBASE",
+ api_key=os.getenv("BROWSERBASE_API_KEY"),
+ project_id=os.getenv("BROWSERBASE_PROJECT_ID"),
+ model_name="gpt-4o",
+ verbose=2,
+ dom_settle_timeout_ms=3000,
+ self_heal=True,
+ wait_for_captcha_solves=True,
+ system_prompt="You are a browser automation assistant for integration testing.",
+ model_client_options={"apiKey": os.getenv("MODEL_API_KEY") or os.getenv("OPENAI_API_KEY")},
+ )
+
+ @pytest_asyncio.fixture
+ async def local_stagehand(self, local_config):
+ """Create a Stagehand instance for LOCAL testing"""
+ stagehand = Stagehand(config=local_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest_asyncio.fixture
+ async def browserbase_stagehand(self, browserbase_config):
+ """Create a Stagehand instance for BROWSERBASE testing"""
+ # Skip if Browserbase credentials are not available
+ if not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")):
+ pytest.skip("Browserbase credentials not available")
+
+ stagehand = Stagehand(config=browserbase_config)
+ await stagehand.init()
+ yield stagehand
+ await stagehand.close()
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_basic_navigation_and_observe_local(self, local_stagehand):
+ """Test basic navigation and observe functionality in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a simple page
+ await stagehand.page.goto("https://example.com")
+
+ # Observe elements on the page
+ observations = await stagehand.page.observe("Find all the links on the page")
+
+ # Verify we got some observations
+ assert observations is not None
+ assert len(observations) > 0
+
+ # Verify observation structure
+ for obs in observations:
+ assert hasattr(obs, "selector")
+ assert obs.selector # Not empty
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_basic_navigation_and_observe_browserbase(self, browserbase_stagehand):
+ """Test basic navigation and observe functionality in BROWSERBASE mode"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a simple page
+ await stagehand.page.goto("https://example.com")
+
+ # Observe elements on the page
+ observations = await stagehand.page.observe("Find all the links on the page")
+
+ # Verify we got some observations
+ assert observations is not None
+ assert len(observations) > 0
+
+ # Verify observation structure
+ for obs in observations:
+ assert hasattr(obs, "selector")
+ assert obs.selector # Not empty
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_form_interaction_local(self, local_stagehand):
+ """Test form interaction capabilities in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with forms
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Observe form elements
+ form_elements = await stagehand.page.observe("Find all form input elements")
+
+ # Verify we found form elements
+ assert form_elements is not None
+ assert len(form_elements) > 0
+
+ # Try to interact with a form field
+ await stagehand.page.act("Fill the customer name field with 'Test User'")
+
+ # Verify the field was filled by observing its value
+ filled_elements = await stagehand.page.observe("Find the customer name input field")
+ assert filled_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_form_interaction_browserbase(self, browserbase_stagehand):
+ """Test form interaction capabilities in BROWSERBASE mode"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a page with forms
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Observe form elements
+ form_elements = await stagehand.page.observe("Find all form input elements")
+
+ # Verify we found form elements
+ assert form_elements is not None
+ assert len(form_elements) > 0
+
+ # Try to interact with a form field
+ await stagehand.page.act("Fill the customer name field with 'Test User'")
+
+ # Verify the field was filled by observing its value
+ filled_elements = await stagehand.page.observe("Find the customer name input field")
+ assert filled_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_search_functionality_local(self, local_stagehand):
+ """Test search functionality similar to examples in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a search page
+ await stagehand.page.goto("https://www.google.com")
+
+ # Find and interact with search box
+ search_elements = await stagehand.page.observe("Find the search input field")
+ assert search_elements is not None
+ assert len(search_elements) > 0
+
+ # Perform a search
+ await stagehand.page.act("Type 'python automation' in the search box")
+
+ # Submit the search (press Enter or click search button)
+ await stagehand.page.act("Press Enter or click the search button")
+
+ # Wait for results and observe them
+ await asyncio.sleep(2) # Give time for results to load
+
+ # Observe search results
+ results = await stagehand.page.observe("Find search result links")
+ assert results is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_extraction_functionality_local(self, local_stagehand):
+ """Test extraction functionality with schema validation in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a news site
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Extract article titles using simple string instruction
+ articles_text = await stagehand.page.extract(
+ "Extract the titles of the first 3 articles on the page as a JSON list"
+ )
+
+ # Verify extraction worked
+ assert articles_text is not None
+
+ # Test with schema-based extraction
+ extract_options = ExtractOptions(
+ instruction="Extract the first article's title and a brief summary",
+ schema_definition=NewsArticle
+ )
+
+ article_data = await stagehand.page.extract(extract_options)
+ assert article_data is not None
+
+ # Validate the extracted data structure
+ if hasattr(article_data, 'data') and article_data.data:
+ # BROWSERBASE mode format
+ article = NewsArticle.model_validate(article_data.data)
+ assert article.title
+ elif hasattr(article_data, 'title'):
+ # LOCAL mode format
+ article = NewsArticle.model_validate(article_data.model_dump())
+ assert article.title
+
+ @pytest.mark.asyncio
+ @pytest.mark.browserbase
+ @pytest.mark.skipif(
+ not (os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID")),
+ reason="Browserbase credentials not available"
+ )
+ async def test_extraction_functionality_browserbase(self, browserbase_stagehand):
+ """Test extraction functionality with schema validation in BROWSERBASE mode"""
+ stagehand = browserbase_stagehand
+
+ # Navigate to a news site
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Extract article titles using simple string instruction
+ articles_text = await stagehand.page.extract(
+ "Extract the titles of the first 3 articles on the page as a JSON list"
+ )
+
+ # Verify extraction worked
+ assert articles_text is not None
+
+ # Test with schema-based extraction
+ extract_options = ExtractOptions(
+ instruction="Extract the first article's title and a brief summary",
+ schema_definition=NewsArticle
+ )
+
+ article_data = await stagehand.page.extract(extract_options)
+ assert article_data is not None
+
+ # Validate the extracted data structure
+ if hasattr(article_data, 'data') and article_data.data:
+ # BROWSERBASE mode format
+ article = NewsArticle.model_validate(article_data.data)
+ assert article.title
+ elif hasattr(article_data, 'title'):
+ # LOCAL mode format
+ article = NewsArticle.model_validate(article_data.model_dump())
+ assert article.title
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_multi_page_workflow_local(self, local_stagehand):
+ """Test multi-page workflow similar to examples in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Start at a homepage
+ await stagehand.page.goto("https://example.com")
+
+ # Observe initial page
+ initial_observations = await stagehand.page.observe("Find all navigation links")
+ assert initial_observations is not None
+
+ # Create a new page in the same context
+ new_page = await stagehand.context.new_page()
+ await new_page.goto("https://httpbin.org")
+
+ # Observe elements on the new page
+ new_page_observations = await new_page.observe("Find the main content area")
+ assert new_page_observations is not None
+
+ # Verify both pages are working independently
+ assert stagehand.page != new_page
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_accessibility_features_local(self, local_stagehand):
+ """Test accessibility tree extraction in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a page with form elements
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Test accessibility tree extraction by finding labeled elements
+ labeled_elements = await stagehand.page.observe("Find all form elements with labels")
+ assert labeled_elements is not None
+
+ # Test finding elements by accessibility properties
+ accessible_elements = await stagehand.page.observe(
+ "Find all interactive elements that are accessible to screen readers"
+ )
+ assert accessible_elements is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_error_handling_local(self, local_stagehand):
+ """Test error handling and recovery in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Test with a non-existent page (should handle gracefully)
+ with pytest.raises(Exception):
+ await stagehand.page.goto("https://thisdomaindoesnotexist12345.com")
+
+ # Test with a valid page after error
+ await stagehand.page.goto("https://example.com")
+ observations = await stagehand.page.observe("Find any elements on the page")
+ assert observations is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.local
+ async def test_performance_basic_local(self, local_stagehand):
+ """Test basic performance characteristics in LOCAL mode"""
+ import time
+
+ stagehand = local_stagehand
+
+ # Time navigation
+ start_time = time.time()
+ await stagehand.page.goto("https://example.com")
+ navigation_time = time.time() - start_time
+
+ # Navigation should complete within reasonable time (30 seconds)
+ assert navigation_time < 30.0
+
+ # Time observation
+ start_time = time.time()
+ observations = await stagehand.page.observe("Find all links on the page")
+ observation_time = time.time() - start_time
+
+ # Observation should complete within reasonable time (20 seconds)
+ assert observation_time < 20.0
+ assert observations is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.slow
+ @pytest.mark.local
+ async def test_complex_workflow_local(self, local_stagehand):
+ """Test complex multi-step workflow in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to a form page
+ await stagehand.page.goto("https://httpbin.org/forms/post")
+
+ # Step 1: Observe the form structure
+ form_structure = await stagehand.page.observe("Find all form fields and their labels")
+ assert form_structure is not None
+ assert len(form_structure) > 0
+
+ # Step 2: Fill multiple form fields
+ await stagehand.page.act("Fill the customer name field with 'Integration Test User'")
+ await stagehand.page.act("Fill the telephone field with '555-1234'")
+ await stagehand.page.act("Fill the email field with 'test@example.com'")
+
+ # Step 3: Observe filled fields to verify
+ filled_fields = await stagehand.page.observe("Find all filled form input fields")
+ assert filled_fields is not None
+
+ # Step 4: Extract the form data
+ form_data = await stagehand.page.extract(
+ "Extract all the form field values as a JSON object"
+ )
+ assert form_data is not None
+
+ @pytest.mark.asyncio
+ @pytest.mark.e2e
+ @pytest.mark.local
+ async def test_end_to_end_search_and_extract_local(self, local_stagehand):
+ """End-to-end test: search and extract results in LOCAL mode"""
+ stagehand = local_stagehand
+
+ # Navigate to search page
+ await stagehand.page.goto("https://news.ycombinator.com")
+
+ # Extract top stories
+ stories = await stagehand.page.extract(
+ "Extract the titles and points of the top 5 stories as a JSON array with title and points fields"
+ )
+
+ assert stories is not None
+
+ # Navigate to first story (if available)
+ story_links = await stagehand.page.observe("Find the first story link")
+ if story_links and len(story_links) > 0:
+ await stagehand.page.act("Click on the first story title link")
+
+ # Wait for page load
+ await asyncio.sleep(3)
+
+ # Extract content from the story page
+ content = await stagehand.page.extract("Extract the main content or title from this page")
+ assert content is not None
+
+ # Test Configuration and Environment Detection
+ def test_environment_detection(self):
+ """Test that environment is correctly detected based on available credentials"""
+ # Test LOCAL mode detection
+ local_config = StagehandConfig(env="LOCAL")
+ assert local_config.env == "LOCAL"
+
+ # Test BROWSERBASE mode configuration
+ if os.getenv("BROWSERBASE_API_KEY") and os.getenv("BROWSERBASE_PROJECT_ID"):
+ browserbase_config = StagehandConfig(
+ env="BROWSERBASE",
+ api_key=os.getenv("BROWSERBASE_API_KEY"),
+ project_id=os.getenv("BROWSERBASE_PROJECT_ID")
+ )
+ assert browserbase_config.env == "BROWSERBASE"
+ assert browserbase_config.api_key is not None
+ assert browserbase_config.project_id is not None
\ No newline at end of file
diff --git a/tests/e2e/test_workflows.py b/tests/e2e/test_workflows.py
new file mode 100644
index 0000000..a03a06f
--- /dev/null
+++ b/tests/e2e/test_workflows.py
@@ -0,0 +1,733 @@
+"""End-to-end integration tests for complete Stagehand workflows"""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+from pydantic import BaseModel
+
+from stagehand import Stagehand, StagehandConfig
+from stagehand.schemas import ActResult, ObserveResult, ExtractResult
+from tests.mocks.mock_llm import MockLLMClient
+from tests.mocks.mock_browser import create_mock_browser_stack, setup_page_with_content
+from tests.mocks.mock_server import create_mock_server_with_client, setup_successful_session_flow
+
+
+class TestCompleteWorkflows:
+ """Test complete automation workflows end-to-end"""
+
+ @pytest.mark.asyncio
+ async def test_search_and_extract_workflow(self, mock_stagehand_config, sample_html_content):
+ """Test complete workflow: navigate → search → extract results"""
+
+ # Create mock components
+ playwright, browser, context, page = create_mock_browser_stack()
+ setup_page_with_content(page, sample_html_content, "https://example.com")
+
+ # Setup mock LLM client
+ mock_llm = MockLLMClient()
+
+ # Configure specific responses for each step
+ mock_llm.set_custom_response("act", {
+ "success": True,
+ "message": "Search executed successfully",
+ "action": "search for openai"
+ })
+
+ mock_llm.set_custom_response("extract", {
+ "title": "OpenAI Search Results",
+ "results": [
+ {"title": "OpenAI Official Website", "url": "https://openai.com"},
+ {"title": "OpenAI API Documentation", "url": "https://platform.openai.com"}
+ ]
+ })
+
+ with patch('stagehand.main.async_playwright') as mock_playwright_func, \
+ patch('stagehand.main.LLMClient') as mock_llm_class:
+
+ mock_playwright_func.return_value.start = AsyncMock(return_value=playwright)
+ mock_llm_class.return_value = mock_llm
+
+ # Initialize Stagehand
+ stagehand = Stagehand(config=mock_stagehand_config)
+ stagehand._playwright = playwright
+ stagehand._browser = browser
+ stagehand._context = context
+ stagehand.page = MagicMock()
+ stagehand.page.goto = AsyncMock()
+ stagehand.page.act = AsyncMock(return_value=ActResult(
+ success=True,
+ message="Search executed",
+ action="search"
+ ))
+ stagehand.page.extract = AsyncMock(return_value={
+ "title": "OpenAI Search Results",
+ "results": [
+ {"title": "OpenAI Official Website", "url": "https://openai.com"},
+ {"title": "OpenAI API Documentation", "url": "https://platform.openai.com"}
+ ]
+ })
+ stagehand._initialized = True
+
+ try:
+ # Execute workflow
+ await stagehand.page.goto("https://google.com")
+
+ # Perform search
+ search_result = await stagehand.page.act("search for openai")
+ assert search_result.success is True
+
+ # Extract results
+ extracted_data = await stagehand.page.extract("extract search results")
+ assert extracted_data["title"] == "OpenAI Search Results"
+ assert len(extracted_data["results"]) == 2
+ assert extracted_data["results"][0]["title"] == "OpenAI Official Website"
+
+ # Verify calls were made
+ stagehand.page.goto.assert_called_with("https://google.com")
+ stagehand.page.act.assert_called_with("search for openai")
+ stagehand.page.extract.assert_called_with("extract search results")
+
+ finally:
+ stagehand._closed = True
+
+ @pytest.mark.asyncio
+ async def test_form_filling_workflow(self, mock_stagehand_config):
+ """Test workflow: navigate → fill form → submit → verify"""
+
+ form_html = """
+
+
+
+