Add manual SDK tests (#4555)

tests/sdk/python_sdk/.gitignore

@@ -0,0 +1,2 @@
+.env
+*.png

tests/sdk/python_sdk/conftest.py

@@ -0,0 +1,99 @@
+import asyncio
+import socket
+import threading
+import time
+from http.server import HTTPServer, SimpleHTTPRequestHandler
+from pathlib import Path
+
+import pytest
+
+from skyvern import Skyvern
+
+
+@pytest.fixture(scope="session")
+def event_loop():
+    """Create a session-scoped event loop for async session fixtures."""
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    loop.close()
+
+
+class QuietHTTPRequestHandler(SimpleHTTPRequestHandler):
+    """HTTP request handler that suppresses log messages."""
+
+    def log_message(self, format, *args):
+        """Override to suppress HTTP request logging."""
+
+
+def _wait_for_server(host: str, port: int, timeout: float = 5.0) -> bool:
+    """Wait for the server to be ready by attempting to connect."""
+    start_time = time.time()
+    deadline = start_time + timeout
+
+    while time.time() < deadline:
+        try:
+            with socket.create_connection((host, port), timeout=0.1):
+                return True
+        except (ConnectionRefusedError, OSError):
+            time.sleep(0.05)
+
+    return False
+
+
+@pytest.fixture(scope="session")
+async def web_server():
+    """
+    Start a local HTTP server on port 9009 serving from the 'web' directory.
+
+    This is an async fixture that properly waits for the server to be ready.
+    """
+    # Get the directory where this conftest file is located
+    test_dir = Path(__file__).parent
+    web_dir = test_dir.parent / "web"
+
+    # Create web directory if it doesn't exist
+    web_dir.mkdir(exist_ok=True)
+
+    # Create a handler class that serves from the specific directory
+    def handler_factory(*args, **kwargs):
+        return QuietHTTPRequestHandler(*args, directory=str(web_dir), **kwargs)
+
+    # Create and start the HTTP server
+    server = HTTPServer(("localhost", 9009), handler_factory)
+    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+    server_thread.start()
+
+    # Wait for the server to be ready (async-friendly)
+    await asyncio.sleep(0.1)
+
+    # Verify server is responding
+    if not _wait_for_server("localhost", 9009):
+        # Don't wait forever - just close the socket and let daemon thread die
+        server.server_close()
+        raise RuntimeError("Server failed to start")
+
+    base_url = "http://localhost:9009"
+    yield base_url
+
+    # Cleanup: Don't call shutdown() as it can block forever waiting for active connections
+    # Instead, just close the socket and let the daemon thread die
+    server.server_close()
+
+
+@pytest.fixture(scope="session")
+async def skyvern_browser():
+    """
+    Launch a local browser once for the entire test session and reuse it across tests.
+
+    This ensures all tests use the same browser instance, avoiding connection issues.
+    """
+    skyvern = Skyvern.local()
+    browser = await skyvern.launch_local_browser(headless=False)
+
+    yield browser
+
+    # Cleanup: close the browser after all tests complete
+    try:
+        await browser.close()
+    except Exception:
+        pass  # Ignore cleanup errors

tests/sdk/python_sdk/test_sdk_run_tasks.py

@@ -0,0 +1,74 @@
+import asyncio
+
+import pytest
+
+from skyvern.forge.sdk.schemas.credentials import NonEmptyPasswordCredential, TotpType
+from skyvern.schemas.run_blocks import CredentialType
+
+
+@pytest.mark.asyncio
+async def test_login(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/login.html")
+
+    credentials = await skyvern_browser.skyvern.get_credentials()
+    credential = next((item for item in credentials if item.name == "test_login"), None)
+    if credential is None:
+        print("Credentials not found. Creating new one.")
+        credential = await skyvern_browser.skyvern.create_credential(
+            name="test_login",
+            credential_type="password",
+            credential=NonEmptyPasswordCredential(
+                username="testlogin",
+                password="testpassword",
+                totp=None,
+                totp_type=TotpType.NONE,
+            ),
+        )
+
+    await page.agent.login(
+        credential_type=CredentialType.skyvern,
+        credential_id=credential.credential_id,
+    )
+
+    await page.click("#accountBtn")
+    await asyncio.sleep(1)
+    await page.act("Click on 'Click Me' button")
+    assert await page.locator("#clickCounter").text_content() == "Button clicked 1 times"
+
+    await asyncio.sleep(1)
+    await page.screenshot(path="screenshot.png", full_page=True)
+
+
+@pytest.mark.asyncio
+async def test_test_finishes_login(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto("https://www.saucedemo.com/")
+    await page.fill("#user-name", "standard_user")
+    await page.fill("#password", "secret_sauce")
+
+    await page.agent.run_task("Click on login button", engine="skyvern-1.0")
+
+    assert await page.get_by_role("button", name="Add to cart").count() > 0
+
+
+@pytest.mark.asyncio
+async def test_download_file(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/download_file.html")
+
+    r = await page.agent.download_files(
+        prompt="Click the 'Download PDF Report' button to download the sample PDF file",
+        download_suffix="sample_report.pdf",
+    )
+    for downloaded_file in r.downloaded_files:
+        print(downloaded_file)
+    assert len(r.downloaded_files) == 1
+
+    await asyncio.sleep(2)
+    await page.screenshot(path="download_test.png", full_page=True)
+
+    assert len(r.downloaded_files) == 1

tests/sdk/python_sdk/test_sdk_simple_actions.py

@@ -0,0 +1,235 @@
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_clicks(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/click.html")
+
+    assert await page.locator("#counter").text_content() == "Button clicked 0 times"
+    await page.click("#button")
+    assert await page.locator("#counter").text_content() == "Button clicked 1 times"
+
+    await page.click(prompt="Click on the button")
+    assert await page.locator("#counter").text_content() == "Button clicked 2 times"
+
+    print("Fallback")
+    await page.click("#counterBroken", prompt="Click on the button")
+    assert await page.locator("#counter").text_content() == "Button clicked 3 times"
+
+    await page.click("#button")
+    assert await page.locator("#counter").text_content() == "Button clicked 4 times"
+    print("All tests passed")
+
+
+@pytest.mark.asyncio
+async def test_fill(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/input.html")
+
+    assert await page.locator("#output").text_content() == ""
+    await page.fill("#nameInput", "Test1")
+    await page.click("#submitBtn")
+    assert await page.locator("#output").text_content() == "Hello, Test1!"
+
+    await page.fill(prompt="Type 'Test2' in the name input")
+    await page.click("#submitBtn")
+    assert await page.locator("#output").text_content() == "Hello, Test2!"
+
+    await page.fill(prompt="Type the value in the name input", value="Test3")
+    await page.click("#submitBtn")
+    assert await page.locator("#output").text_content() == "Hello, Test3!"
+
+    # fallback
+    await page.fill("#nameInputBroken", "TestFallback", prompt="Fill the name input")
+    await page.click("#submitBtn")
+    assert await page.locator("#output").text_content() == "Hello, TestFallback!"
+
+
+@pytest.mark.asyncio
+async def test_select_option(web_server, skyvern_browser):
+    """Test using page.act() with natural language prompts on a combobox."""
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/combobox.html")
+
+    await page.select_option("#cars", "audi")
+    assert await page.locator("#cars").input_value() == "audi"
+
+    await page.select_option("#cars", value="opel")
+    assert await page.locator("#cars").input_value() == "opel"
+
+    await page.select_option("#cars", label="Saab")
+    assert await page.locator("#cars").input_value() == "saab"
+
+    await page.select_option(prompt="Select 'Audi' i the car combobox")
+    assert await page.locator("#cars").input_value() == "audi"
+
+    # fallback
+    await page.select_option("#cars-broken", "opel", prompt="Select 'Opel' i the car combobox")
+    assert await page.locator("#cars").input_value() == "opel"
+
+    await page.select_option("#cars-broken", label="Saab", prompt="Select 'Saab' i the car combobox")
+    assert await page.locator("#cars").input_value() == "saab"
+
+
+@pytest.mark.asyncio
+async def test_act_combobox(web_server, skyvern_browser):
+    """Test using page.act() with natural language prompts on a combobox."""
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/combobox.html")
+
+    await page.act("select 'Audi' from the combobox")
+
+    assert await page.locator("#cars").input_value() == "audi"
+
+
+@pytest.mark.asyncio
+async def test_act_input_and_click(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/input.html")
+
+    await page.act("type 'ActTest' into the input box")
+    await page.act("click on the button")
+
+    assert await page.locator("#output").text_content() == "Hello, ActTest!"
+
+
+@pytest.mark.asyncio
+async def test_upload(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+    image_url = "https://img.freepik.com/free-photo/portrait-beautiful-purebred-pussycat-with-shorthair-orange-collar-neck-sitting-floor-reacting-camera-flash-scared-looking-light-indoor_8353-12551.jpg?semt=ais_hybrid&w=740&q=80"
+
+    await page.goto(f"{web_server}/upload.html")
+
+    await page.reload()
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__() == ""
+    await page.upload_file("#imageUpload", image_url)
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__().startswith("data:image/jpeg")
+
+    await page.reload()
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__() == ""
+    await page.upload_file(prompt="Upload this", files=image_url)
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__().startswith("data:image/jpeg")
+
+    await page.reload()
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__() == ""
+    await page.upload_file(prompt=f"Upload this file {image_url}")
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__().startswith("data:image/jpeg")
+
+    print("fallback")
+    await page.reload()
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__() == ""
+    await page.upload_file("#imageUploadBroken", image_url, prompt="Upload this file")
+    assert (await page.locator("#uploadedImage").get_attribute("src")).__str__().startswith("data:image/jpeg")
+    print("all done")
+
+
+@pytest.mark.asyncio
+async def test_extract(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/click.html")
+
+    result = await page.extract("give one sentence description of this page")
+    print(result)
+    assert "click" in str(result)
+
+    result = await page.extract(
+        prompt="Describe this page",
+        schema={
+            "type": "object",
+            "properties": {
+                "short": {"type": "string", "description": "one sentence description of this page"},
+                "long": {"type": "string", "description": "two-three sentence description of this page"},
+            },
+            "required": ["short", "long"],
+        },
+    )
+    print(result)
+    assert "click" in str(result)
+    assert "short" in str(result)
+    assert "long" in str(result)
+
+
+@pytest.mark.asyncio
+async def test_prompt_locator(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/click.html")
+
+    assert await page.locator("#counter").text_content() == "Button clicked 0 times"
+    await page.locator(prompt="Find the 'click me' button").click()
+    assert await page.locator("#counter").text_content() == "Button clicked 1 times"
+
+    await page.locator(prompt="Find the 'click me' button").nth(0).click()
+    assert await page.locator("#counter").text_content() == "Button clicked 2 times"
+
+    await page.locator("#bad-selector", prompt="Find the 'click me' button").click()
+    assert await page.locator("#counter").text_content() == "Button clicked 3 times"
+
+    await page.locator("#bad-selector", prompt="Find the 'click me' button").nth(0).click()
+    assert await page.locator("#counter").text_content() == "Button clicked 4 times"
+
+    await page.locator("#button").click()
+    assert await page.locator("#counter").text_content() == "Button clicked 5 times"
+
+
+@pytest.mark.asyncio
+async def test_prompt_locator_chaining(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/combobox.html")
+
+    await page.locator("#cars").select_option("opel")
+    assert await page.locator("#cars").input_value() == "opel"
+
+    await page.locator(prompt="Find the 'cars' combobox").select_option("saab")
+    assert await page.locator("#cars").input_value() == "saab"
+
+    await page.locator("#bad-selector", prompt="Find the 'cars' combobox").select_option("audi")
+    assert await page.locator("#cars").input_value() == "audi"
+
+
+@pytest.mark.asyncio
+async def test_validate(web_server, skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    await page.goto(f"{web_server}/click.html")
+    await page.click("#button")
+
+    is_valid = await page.validate("if clicked time > 0")
+    assert is_valid
+
+    is_valid = await page.validate("if clicked time > 1")
+    assert not is_valid
+
+    # invalid prompt does not pass validation
+    is_valid = await page.validate("the input text is valid")
+    assert not is_valid
+
+
+@pytest.mark.asyncio
+async def test_prompt(skyvern_browser):
+    page = await skyvern_browser.get_working_page()
+
+    r = await page.prompt("1111+1111")
+    print(r)
+    assert "2222" in str(r)
+
+    r = await page.prompt(
+        "2+2",
+        schema={
+            "type": "object",
+            "properties": {
+                "result_number": {"type": "int"},
+                "confidence": {"type": "number", "minimum": 0, "maximum": 1},
+            },
+        },
+    )
+    print(r)
+    assert r["result_number"] == 4