test_browser_fastmcp_client.py•63.6 kB
import unittest
import json
from fastmcp import Client
from browser_fastmcp_server import mcp as browsers_mcp
class TestBrowserFastMCPClient(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.client = Client(browsers_mcp)
async def asyncTearDown(self):
await self.client.__aexit__(None, None, None)
async def test_create_chrome_instance(self):
async with self.client:
result = await self.client.call_tool("create_chrome_instance", {"headless": True})
print("create_chrome_instance:", result)
self.assertTrue("session_id" in str(result))
await self.client.call_tool("close_instance", {"session_id": result.data.session_id})
async def test_close_instance(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("close_instance", {"session_id": session_id})
self.assertIn(f"Chrome instance {session_id} closed successfully.", result.data)
async def test_get_instance_info(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("get_instance_info", {"session_id": session_id})
print("get_instance_info:", result)
self.assertEqual(result.data.session_id, session_id)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_get_browser_status(self):
async with self.client:
result = await self.client.call_tool("get_browser_status", {})
print("get_browser_status:", result)
self.assertEqual(result.data, [])
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
result = await self.client.call_tool("get_browser_status", {})
print("get_browser_status:", result)
self.assertEqual(len(result.data), 1)
self.assertEqual(result.data[0]['session_id'], session.data.session_id)
await self.client.call_tool("close_instance", {"session_id": session.data.session_id})
async def test_close_all_instances(self):
async with self.client:
await self.client.call_tool("create_chrome_instance", {"headless": True})
result = await self.client.call_tool("close_all_instances", {})
print("close_all_instances:", result)
self.assertIn("All browser instances closed.", str(result))
async def test_set_and_get_browser_config(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
await self.client.call_tool("wait", {"seconds": 3})
session_id = session.data.session_id
# Set headless to False and change viewport size
await self.client.call_tool("set_browser_config", {
"session_id": session_id,
"headless": True,
"viewport_width": 250,
"viewport_height": 720
})
await self.client.call_tool("wait", {"seconds": 3})
result = await self.client.call_tool("get_browser_config", {"session_id": session_id})
print("get_browser_config:", result)
self.assertEqual(result.data.session_id, session_id)
self.assertEqual(result.data.config.viewport_width, 250)
self.assertEqual(result.data.config.viewport_height, 720)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_navigate_to_and_back_forward(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
urls = [
"https://httpbin.org/html",
"https://example.com/",
"https://www.baidu.com/"
]
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": urls[0]})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue(urls[0].rstrip('/') in state.data.url)
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": urls[1]})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue(urls[1].rstrip('/') in state.data.url)
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": urls[2]})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue("baidu.com" in state.data.url)
await self.client.call_tool("navigate_back", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue(urls[1].rstrip('/') in state.data.url)
await self.client.call_tool("navigate_back", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue(urls[0].rstrip('/') in state.data.url)
await self.client.call_tool("navigate_forward", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue(urls[1].rstrip('/') in state.data.url)
await self.client.call_tool("navigate_forward", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertTrue("baidu.com" in state.data.url)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_get_tabs_info_and_switch_close_tab(self):
async with self.client:
# 1. create browser instance
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
# 2. open first tab, visit page A
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 3})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
self.assertEqual(len(tabs.data), 1, "should have one tab after first navigation")
page_id1 = tabs.data[0]['id']
# 3. open second tab, visit page B
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://example.com/", "new_tab": True})
await self.client.call_tool("wait", {"seconds": 3})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
self.assertEqual(len(tabs.data), 2, "should have two tabs after opening new tab")
page_id2 = tabs.data[1]['id']
# 4. switch to first tab, verify page A
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id1})
await self.client.call_tool("wait", {"seconds": 3})
state1 = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertIn("httpbin.org/html", state1.data.url)
# 5. switch to second tab, verify page B
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id2})
await self.client.call_tool("wait", {"seconds": 3})
state2 = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertIn("example.com", state2.data.url)
# 6. close second tab, verify first tab still exists and is available
await self.client.call_tool("close_tab", {"session_id": session_id, "page_id": page_id2})
await self.client.call_tool("wait", {"seconds": 3})
tabs_after = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
self.assertEqual(len(tabs_after.data), 1, "should have one tab after closing second tab")
self.assertEqual(tabs_after.data[0]['id'], page_id1, "remaining tab should be the first one")
# verify page A again
state1_again = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertIn("httpbin.org/html", state1_again.data.url)
# 7. close first tab, verify all tabs are closed or become about:blank
await self.client.call_tool("close_tab", {"session_id": session_id, "page_id": page_id1})
await self.client.call_tool("wait", {"seconds": 3})
tabs_final = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
self.assertEqual(len(tabs_final.data), 1, "should have one tab (about:blank) after closing all tabs")
self.assertEqual(tabs_final.data[0]['url'], 'about:blank', "tab should be blank after all closed")
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_multi_tab_navigation_and_switch(self):
async with self.client:
# 1. create browser instance
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
# 2. open first tab, visit page A
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/image/png"})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
page_id1 = tabs.data[0]['id']
tab1 = next(tab for tab in tabs.data if tab['id'] == page_id1)
self.assertTrue(tab1['url'].startswith("https://httpbin.org/image/png"), f"Tab1 url should start with https://httpbin.org/image/png, got {tab1['url']}")
# 3. open second tab, visit page B
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/json", "new_tab": True})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
page_id2 = tabs.data[-1]['id']
tab2 = next(tab for tab in tabs.data if tab['id'] == page_id2)
self.assertEqual(tab2['url'], "https://httpbin.org/json", f"Tab2 url should be https://httpbin.org/json, got {tab2['url']}")
# switch to tab1, verify tab1's history is independent
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id1})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
tab1 = next(tab for tab in tabs.data if tab['id'] == page_id1)
self.assertTrue(tab1['url'].startswith("https://httpbin.org/image/png"), f"Tab1 url should start with https://httpbin.org/image/png after switch, got {tab1['url']}")
# 4. switch to first tab, forward/backward navigation
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id1})
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/image/png?2"})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
tab1 = next(tab for tab in tabs.data if tab['id'] == page_id1)
self.assertTrue(tab1['url'].startswith("https://httpbin.org/image/png?2"), f"Tab1 url should start with https://httpbin.org/image/png?2, got {tab1['url']}")
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id1})
await self.client.call_tool("navigate_back", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
tab1 = next(tab for tab in tabs.data if tab['id'] == page_id1)
self.assertTrue(tab1['url'].startswith("https://httpbin.org/image/png"), f"Tab1 url should start with https://httpbin.org/image/png after back, got {tab1['url']}")
# 5. switch to second tab, forward/backward navigation
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id2})
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/json?2"})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
tab2 = next(tab for tab in tabs.data if tab['id'] == page_id2)
self.assertTrue(tab2['url'].startswith("https://httpbin.org/json?2"), f"Tab2 url should start with https://httpbin.org/json?2, got {tab2['url']}")
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": page_id2})
await self.client.call_tool("navigate_back", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
tab2 = next(tab for tab in tabs.data if tab['id'] == page_id2)
self.assertEqual(tab2['url'], "https://httpbin.org/json", f"Tab2 url should be https://httpbin.org/json after back navigation, got {tab2['url']}")
# 7. close one tab, verify remaining tab status
await self.client.call_tool("close_tab", {"session_id": session_id, "page_id": page_id2})
tabs_after = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
self.assertEqual(len(tabs_after.data), 1, "Only one tab should remain after closing tab2")
remaining_tab = tabs_after.data[0]
self.assertEqual(remaining_tab['id'], page_id1, "The remaining tab should be tab1")
self.assertTrue(remaining_tab['url'].startswith("https://httpbin.org/image/png"), f"Remaining tab url should start with https://httpbin.org/image/png, got {remaining_tab['url']}")
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_get_page_state(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 2})
result = await self.client.call_tool("get_page_state", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 2})
print("get_page_state:", result)
self.assertTrue(result.data.url.startswith("https://httpbin.org/html"))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_click_news_link(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://www.baidu.com"})
await self.client.call_tool("wait", {"seconds": 10})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
news_element = next((el for el in state.data.elements if "新闻" in el.text and el.tag == "a"), None)
self.assertIsNotNone(news_element, "not found '新闻' link")
print("news_element:", news_element)
await self.client.call_tool("click_element_by_xpath", {"session_id": session_id, "xpath": news_element.xpath})
await self.client.call_tool("wait", {"seconds": 10})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
new_tab = None
for tab in tabs.data:
if "news.baidu.com" in tab['url']:
new_tab = tab
break
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": new_tab['id']})
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertIn("news.baidu.com", new_state.data.url)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_click_bing_link(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://www.bing.com"})
await self.client.call_tool("wait", {"seconds": 10})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
import time
timeout = 20 # wait for 10 seconds
start = time.time()
while True:
await self.client.call_tool("wait", {"seconds": 5})
result = await self.client.call_tool("take_screenshot", {"session_id": session_id})
print("take_screenshot:", result)
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
news_element = next((el for el in new_state.data.elements if ( "图片" in el.text or "images" in el.text.lower()) and el.tag == "a"), None)
if news_element:
break
if time.time() - start > timeout:
raise TimeoutError("wait for image page timeout")
await self.client.call_tool("wait", {"seconds": 2})
await self.client.call_tool("click_element", {"session_id": session_id, "index": news_element.index})
await self.client.call_tool("wait", {"seconds": 1})
tabs = await self.client.call_tool("get_tabs_info", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
new_tab = None
for tab in tabs.data:
if "bing.com/images" in tab['url']:
new_tab = tab
break
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("switch_tab", {"session_id": session_id, "page_id": new_tab['id']})
await self.client.call_tool("wait", {"seconds": 1})
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
self.assertIn("bing.com/images", new_state.data.url)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_input_text(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id if hasattr(session, 'data') else session.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://www.baidu.com"})
import time
timeout = 10
start = time.time()
input_idx = None
while True:
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
for el in state.data.elements:
if el.tag.lower() == "textarea" or el.tag.lower() == "input":
input_idx = el.index
break
if input_idx is not None:
break
if time.time() - start > timeout:
raise TimeoutError("wait for input element timeout")
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("input_text", {"session_id": session_id, "index": input_idx, "text": "Playwright Python"})
self.assertIn("Input text", str(result))
await self.client.call_tool("click_element_by_xpath", {"session_id": session_id, "xpath": el.xpath})
await self.client.call_tool("wait", {"seconds": 1})
timeout = 10
start = time.time()
found_result = False
while True:
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
for el in state.data.elements:
if el.tag.lower() == "li" or el.tag.lower() == "div":
if "playwright" in el.text.lower() and "python" in el.text.lower():
found_result = True
break
if found_result:
break
if time.time() - start > timeout:
raise TimeoutError("wait for search result timeout")
await self.client.call_tool("wait", {"seconds": 2})
self.assertTrue(found_result, "not found search result")
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_send_keys(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id if hasattr(session, 'data') else session.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://www.baidu.com"})
import time
timeout = 10
start = time.time()
input_idx = None
while True:
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
for el in state.data.elements:
if el.tag.lower() == "textarea" or el.tag.lower() == "input":
input_idx = el.index
break
if input_idx is not None:
break
if time.time() - start > timeout:
raise TimeoutError("wait for input element timeout")
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("input_text", {"session_id": session_id, "index": input_idx, "text": "Playwright Python"})
self.assertIn("Input text", str(result))
await self.client.call_tool("click_element_by_xpath", {"session_id": session_id, "xpath": el.xpath})
result = await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "Enter"})
self.assertIn("Sent keys", str(result))
timeout = 10
start = time.time()
found_result = False
while True:
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
for el in state.data.elements:
print(el.xpath)
if "div/div/div[1]/div/h3/a/div/div/p/span/span/span" in el.xpath:
found_result = True
break
if time.time() - start > timeout:
break
await self.client.call_tool("wait", {"seconds": 1})
self.assertTrue(found_result, "not found search result page")
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "PageDown"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "PageUp"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "End"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "Home"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_set_element_value(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id if hasattr(session, 'data') else session.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://www.baidu.com"})
import time
timeout = 10
start = time.time()
input_idx = None
while True:
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
for el in state.data.elements:
if el.tag.lower() == "textarea" or el.tag.lower() == "input":
input_idx = el.index
break
if input_idx is not None:
break
if time.time() - start > timeout:
raise TimeoutError("wait for input element timeout")
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("set_element_value", {"session_id": session_id, "index": input_idx, "value": "playwright python"})
self.assertIn("Set value", str(result))
await self.client.call_tool("click_element_by_xpath", {"session_id": session_id, "xpath": el.xpath})
result = await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "Enter"})
self.assertIn("Sent keys", str(result))
timeout = 10
start = time.time()
found_result = False
while True:
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
for el in state.data.elements:
print(el.xpath)
if "div/div/div[1]/div/h3/a/div/div/p/span/span/span" in el.xpath:
found_result = True
break
if time.time() - start > timeout:
break
await self.client.call_tool("wait", {"seconds": 1})
self.assertTrue(found_result, "not found search result page")
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "PageDown"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "PageUp"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "End"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "Home"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_url(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
url = "https://httpbin.org/html"
result = await self.client.call_tool("generate_pdf", {"session_id": session_id, "url": url})
print("generate_pdf with url:", result)
self.assertTrue(result.data.success)
self.assertTrue(result.data.file_path)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_html_content(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
html_content = "<html><body><h1>Test PDF</h1></body></html>"
result = await self.client.call_tool("generate_pdf", {"session_id": session_id, "html_content": html_content})
print("generate_pdf with html_content:", result)
self.assertTrue(result.data.success)
self.assertTrue(result.data.file_path)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_output_filename(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
output_filename = "custom_name.pdf"
result = await self.client.call_tool("generate_pdf", {"session_id": session_id, "output_filename": output_filename})
print("generate_pdf with output_filename:", result)
self.assertTrue(result.data.success)
self.assertEqual(result.data.file_name, output_filename)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_print_background(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("generate_pdf", {"session_id": session_id, "print_background": False})
print("generate_pdf with print_background=False:", result)
self.assertTrue(result.data.success)
self.assertTrue(result.data.file_path)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_margins(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("generate_pdf", {
"session_id": session_id,
"margin_top": 1.0,
"margin_bottom": 1.0,
"margin_left": 1.0,
"margin_right": 1.0
})
print("generate_pdf with custom margins:", result)
self.assertTrue(result.data.success)
self.assertTrue(result.data.file_path)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_paper_size(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("generate_pdf", {
"session_id": session_id,
"paper_width": 10.0,
"paper_height": 15.0
})
print("generate_pdf with custom paper size:", result)
self.assertTrue(result.data.success)
self.assertTrue(result.data.file_path)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_generate_pdf_with_all_params(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("generate_pdf", {
"session_id": session_id,
"url": "https://httpbin.org/html",
"output_filename": "all_params.pdf",
"print_background": True,
"margin_top": 2,
"margin_bottom": 2,
"margin_left": 4,
"margin_right": 0.5,
"paper_width": 9.0,
"paper_height": 12.0
})
print("generate_pdf with all params:", result)
self.assertTrue(result.data.success)
self.assertEqual(result.data.file_name, "all_params.pdf")
self.assertTrue(result.data.file_path)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_download_file(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
url = "https://proof.ovh.net/files/"
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": url})
await self.client.call_tool("wait", {"seconds": 3})
await self.client.call_tool("scroll_page", {"session_id": session_id, "direction": "down"})
await self.client.call_tool("wait", {"seconds": 1})
import time
timeout = 20 # 增加超时时间
start = time.time()
retry_count = 0
max_retries = 3
while True:
try:
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
# 尝试获取页面状态
try:
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
news_element = next((el for el in new_state.data.elements if "1 mio file" in el.text.lower() and el.tag == "a"), None)
if news_element:
print("download link:", news_element)
break
except Exception as e:
# 如果get_page_state失败，检查浏览器健康状态
if "Page is not accessible" in str(e) or "Failed to get page state" in str(e):
print(f"⚠️ Page accessibility issue detected: {e}")
# 检查浏览器健康状态
try:
health_status = await self.client.call_tool("check_browser_health", {"session_id": session_id})
print(f"🔍 Browser health check: {health_status}")
except Exception as health_error:
print(f"❌ Health check failed: {health_error}")
# 尝试刷新页面
if retry_count < max_retries:
retry_count += 1
print(f"🔄 Attempting to refresh page (attempt {retry_count}/{max_retries})")
try:
await self.client.call_tool("refresh_page", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 3})
continue
except Exception as refresh_error:
print(f"❌ Page refresh failed: {refresh_error}")
else:
print("❌ Max retries reached, creating new instance")
# 关闭当前实例并创建新实例
await self.client.call_tool("close_instance", {"session_id": session_id})
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": url})
await self.client.call_tool("wait", {"seconds": 3})
retry_count = 0
continue
else:
# 其他类型的错误，直接抛出
raise e
except Exception as e:
print(f"❌ Error in main loop: {e}")
if retry_count < max_retries:
retry_count += 1
print(f"🔄 Retrying (attempt {retry_count}/{max_retries})")
await self.client.call_tool("wait", {"seconds": 2})
continue
else:
raise Exception(f"Failed after {max_retries} retries: {e}")
if time.time() - start > timeout:
raise TimeoutError("wait for download link timeout")
await self.client.call_tool("wait", {"seconds": 1})
# 下载文件
try:
result = await self.client.call_tool("download_file", {"session_id": session_id, "url": news_element.attributes.get("href")})
print("download_file:", result)
self.assertTrue(result.data.success)
print("download file:", result.data.file_path)
except Exception as e:
print(f"❌ Download failed: {e}")
raise e
finally:
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_upload_file(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
url = "https://proof.ovh.net/files/"
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": url})
await self.client.call_tool("wait", {"seconds": 3})
await self.client.call_tool("scroll_page", {"session_id": session_id, "direction": "down"})
await self.client.call_tool("wait", {"seconds": 3})
import time
timeout = 20
start = time.time()
while True:
await self.client.call_tool("wait", {"seconds": 3})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 3})
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
news_element = next((el for el in new_state.data.elements if "1 mio file" in el.text.lower() and el.tag == "a"), None)
if news_element:
print("download link:", news_element)
break
if time.time() - start > timeout:
raise TimeoutError("wait for download link timeout")
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("download_file", {"session_id": session_id, "url": news_element.attributes.get("href")})
print("download_file:", result)
self.assertTrue(result.data.success)
print("download file:", result.data.file_path)
file_path = result.data.file_path
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://pqina.nl/filepond/"})
await self.client.call_tool("wait", {"seconds": 4})
import time
timeout = 20
start = time.time()
while True:
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
news_element = next((el for el in new_state.data.elements if "file" in el.attributes.get("type", "").lower() and el.tag == "input"), None)
if news_element:
print("file input element:", news_element)
break
if time.time() - start > timeout:
raise TimeoutError("wait for file input element timeout")
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("upload_file", {"session_id": session_id, "index": news_element.index, "file_path": file_path})
await self.client.call_tool("wait", {"seconds": 10})
print("upload_file:", result)
self.assertEqual("Successfully uploaded file" in result.data, True)
import time
timeout = 10
start = time.time()
while True:
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
upload_btn = next((el for el in new_state.data.elements if "upload" in el.text.lower() and el.tag == "button"), None)
if upload_btn:
print("upload button element:", upload_btn)
break
if time.time() - start > timeout:
raise TimeoutError("wait for upload button timeout")
await self.client.call_tool("wait", {"seconds": 1})
print("upload button element:", upload_btn)
result = await self.client.call_tool("click_element", {"session_id": session_id, "index": upload_btn.index})
await self.client.call_tool("wait", {"seconds": 10})
print("click upload button result:", result)
timeout = 10
start = time.time()
while True:
new_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
complete_element = next((el for el in new_state.data.elements if "upload complete" in el.text.lower()), None)
if complete_element:
print("upload complete element:", complete_element)
break
if time.time() - start > timeout:
raise TimeoutError("wait for upload complete timeout")
await self.client.call_tool("wait", {"seconds": 1})
self.assertTrue(complete_element, "not found upload complete on page")
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_download_image(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
url = "https://httpbin.org/image/png"
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": url})
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("download_image", {"session_id": session_id, "image_url": url})
print("download_image:", result)
self.assertTrue(result and not result.is_error)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_download_images_from_bing(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
url = "https://www.bing.com/images"
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": url})
await self.client.call_tool("wait", {"seconds": 2})
await self.client.call_tool("scroll_page", {"session_id": session_id, "direction": "down"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("scroll_page", {"session_id": session_id, "direction": "down"})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
image_elements = [el for el in state.data.elements if el.tag.lower() == "a" and json.loads(el.attributes.get("m", "{}")).get("turl", "")]
self.assertTrue(len(image_elements) >= 3, f"Not enough images found on Bing Images page, found {len(image_elements)}")
for i, img_el in enumerate(image_elements[:3]):
img_url = json.loads(img_el.attributes.get("m", "{}")).get("turl", "")
result = await self.client.call_tool("download_image", {"session_id": session_id, "image_url": img_url})
print(f"download_image {i+1}:", result)
self.assertTrue(result and not result.is_error, f"Image {i+1} download failed")
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_take_screenshot(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
# 1. default parameters
result = await self.client.call_tool("take_screenshot", {"session_id": session_id})
print("default:", result)
self.assertTrue(result and not result.is_error)
# 2. specify target element
result = await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"target": "h1"
})
print("target h1:", result)
self.assertTrue(result and not result.is_error)
# 2.1 non-existent target
with self.assertRaises(Exception):
await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"target": "#not-exist"
})
# 3. custom viewport
result = await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"width": 800,
"height": 300,
"full_page": False
})
print("custom viewport:", result)
self.assertTrue(result and not result.is_error)
# 4. format=jpeg, quality=50
result = await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"format": "jpeg",
"quality": 50
})
print("jpeg quality 50:", result)
self.assertTrue(result and not result.is_error)
# 5. full_page=False
result = await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"full_page": False
})
print("full_page False:", result)
self.assertTrue(result and not result.is_error)
# 6. invalid format
with self.assertRaises(Exception):
await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"format": "bmp"
})
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_set_and_get_cookies(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/cookies"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("set_cookie", {"session_id": session_id, "name": "testcookie", "value": "cookieval", "domain": "httpbin.org"})
cookies = await self.client.call_tool("get_cookies", {"session_id": session_id})
print("get_cookies:", cookies)
self.assertTrue(any(c.get("name") == "testcookie" for c in cookies.data))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_wait(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("wait", {"seconds": 1})
print("wait:", result)
self.assertIn("Waited", str(result))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_search_bing(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
result = await self.client.call_tool("search_bing", {"session_id": session_id, "query": "python"})
await self.client.call_tool("wait", {"seconds": 2})
page_state = await self.client.call_tool("get_page_state", {"session_id": session_id})
print("search_bing:", result)
print("page_state:", page_state)
# assume page_state.data.html or page_state.data.text contains page content
self.assertIn("python", str(page_state).lower())
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_scroll_page(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("set_browser_config", {"session_id": session_id, "viewport_width": 150, "viewport_height": 1000})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("send_keys", {"session_id": session_id, "keys": "End"})
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.call_tool("scroll_page", {"session_id": session_id, "direction": "up"})
print("scroll_page:", result)
self.assertIn("Scrolled", str(result))
result = await self.client.call_tool("take_screenshot", {
"session_id": session_id,
"full_page": False,
"width": 150,
"height": 1000
})
print("scroll_page:", result)
self.assertTrue(result and not result.is_error)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_extract_content(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
result = await self.client.call_tool("extract_content", {"session_id": session_id, "query": "Herman"})
print("extract_content:", result)
self.assertIn("Content matching", str(result))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_get_dropdown_options(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://www.w3schools.com/tags/tryit.asp?filename=tryhtml_select"})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
select_idx = None
el = None
for el in state.data.elements:
if el.tag.lower() == "select":
select_idx = el.index
el = el
break
if select_idx is not None:
# get dropdown options (structured return)
result = await self.client.call_tool("get_dropdown_options", {"session_id": session_id, "index": select_idx})
print("get_dropdown_options:", result)
# assert structured return
self.assertTrue(hasattr(result.data, "options"))
options = result.data.options
self.assertTrue(len(options) >= 2)
# set select's value to second option
set_result = await self.client.call_tool("set_element_value", {"session_id": session_id, "index": select_idx, "value": options[1].value})
print("set_element_value:", set_result)
self.assertIn("Set value", str(set_result))
# click to test
click_result = await self.client.call_tool("click_element_by_xpath", {"session_id": session_id, "xpath": el.xpath})
print("click_element:", click_result)
self.assertIn("Clicked", str(click_result))
# check if element's value has changed to second_value
element_info = await self.client.call_tool("get_element_info", {"session_id": session_id, "index": select_idx})
print("element_info after set:", element_info)
self.assertEqual(element_info.data.value, options[1].value)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_browser_tips(self):
async with self.client:
result = await self.client.call_tool("browser_tips", {})
self.assertIn("Best Practices", str(result))
async def test_resource_status(self):
async with self.client:
result = await self.client.read_resource("browser://status")
print("resource_status:", result)
self.assertIn("Session-Based Browser Manager Status", str(result))
async def test_resource_instances(self):
async with self.client:
result = await self.client.read_resource("browser://instances")
print("resource_instances:", result)
self.assertTrue(
"Browser Sessions" in str(result) or "No browser sessions active" in str(result)
)
async def test_resource_help(self):
async with self.client:
result = await self.client.read_resource("browser://help")
print("resource_help:", result)
self.assertIn("API Reference", str(result))
async def test_prompt_web_testing(self):
async with self.client:
result = await self.client.get_prompt("web_testing", {"url": "https://httpbin.org/html", "test_scenario": "basic page load"})
print("prompt_web_testing:", result)
self.assertIn("test", str(result).lower())
async def test_prompt_data_extraction(self):
async with self.client:
result = await self.client.get_prompt("data_extraction", {"url": "https://httpbin.org/html", "data_type": "table"})
print("prompt_data_extraction:", result)
self.assertIn("extract", str(result).lower())
async def test_prompt_form_filling(self):
async with self.client:
result = await self.client.get_prompt("form_filling", {"url": "https://httpbin.org/forms/post", "form_data": "name=foo&email=bar"})
print("prompt_form_filling:", result)
self.assertTrue(result)
async def test_prompt_automation_troubleshooting(self):
async with self.client:
result = await self.client.get_prompt("automation_troubleshooting", {})
print("prompt_automation_troubleshooting:", result)
self.assertIn("troubleshooting", str(result).lower())
async def test_resource_instance_page(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.read_resource(f"browser://instance/{session_id}/page")
print("instance_page:", result)
self.assertIn("Current Page", str(result))
self.assertIn("httpbin.org/html", str(result))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_resource_instance_tabs(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://example.com/", "new_tab": True})
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.read_resource(f"browser://instance/{session_id}/tabs")
print("instance_tabs:", result)
self.assertIn("Open Tabs", str(result))
self.assertIn("example.com", str(result))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_resource_instance_screenshots(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
result = await self.client.read_resource(f"browser://instance/{session_id}/screenshots")
print("instance_screenshots:", result)
self.assertIn("Screenshots for Instance", str(result))
self.assertIn(".png", str(result))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_resource_instance_status(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
result = await self.client.read_resource(f"browser://instance/{session_id}/status")
print("instance_status:", result)
self.assertIn("session_id", str(result))
self.assertIn("current_url", str(result))
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_resource_instance_files_and_file(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
# take screenshot
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/html"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
# download a file
url = "https://proof.ovh.net/files/"
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": url})
await self.client.call_tool("wait", {"seconds": 3})
await self.client.call_tool("scroll_page", {"session_id": session_id, "direction": "down"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("take_screenshot", {"session_id": session_id})
await self.client.call_tool("wait", {"seconds": 1})
state = await self.client.call_tool("get_page_state", {"session_id": session_id})
download_link = next((el for el in state.data.elements if "1 mio file" in el.text.lower() and el.tag == "a"), None)
self.assertIsNotNone(download_link)
download_result = await self.client.call_tool("download_file", {"session_id": session_id, "url": download_link.attributes.get("href")})
self.assertTrue(download_result.data.success)
file_name = download_result.data.file_name
# get file list
files_result = await self.client.read_resource(f"browser://instance/{session_id}/files")
print("instance_files:", files_result)
# read single file content
file_content = await self.client.read_resource(f"browser://instance/{session_id}/file/{file_name}")
print("instance_file_content:", file_content)
self.assertTrue(file_content)
await self.client.call_tool("close_instance", {"session_id": session_id})
async def test_resource_instance_cookies(self):
async with self.client:
session = await self.client.call_tool("create_chrome_instance", {"headless": True})
session_id = session.data.session_id
await self.client.call_tool("navigate_to", {"session_id": session_id, "url": "https://httpbin.org/cookies"})
await self.client.call_tool("wait", {"seconds": 1})
await self.client.call_tool("set_cookie", {"session_id": session_id, "name": "testcookie", "value": "cookieval", "domain": "httpbin.org"})
cookies_result = await self.client.read_resource(f"browser://instance/{session_id}/cookies")
print("instance_cookies:", cookies_result)
self.assertTrue(any("testcookie" in str(c) for c in str(cookies_result).splitlines()))
await self.client.call_tool("close_instance", {"session_id": session_id})
if __name__ == "__main__":
unittest.main()