downloads test

Author: kamath

Diff for: examples/e2e/test_playwright.py

@@ -9,12 +9,13 @@
     playwright_basic,
     playwright_captcha,
     playwright_contexts,
+    playwright_downloads,
 )
 
 bb = Browserbase(api_key=BROWSERBASE_API_KEY)
 load_dotenv()
 
-SKIP_CAPTCHA_SOLVING = os.getenv("SKIP_CAPTCHA_SOLVING", "false").lower() == "true"
+SKIP_CAPTCHA_SOLVING = os.getenv("SKIP_CAPTCHA_SOLVING", "true").lower() == "true"
 
 
 @pytest.fixture(scope="session")
@@ -35,3 +36,7 @@ def test_playwright_captcha(playwright: Playwright):
 
 def test_playwright_contexts(playwright: Playwright):
     playwright_contexts.run(playwright)
+
+
+def test_playwright_downloads(playwright: Playwright):
+    playwright_downloads.run(playwright)

Diff for: examples/playwright_downloads.py

@@ -0,0 +1,86 @@
+import re
+import zipfile
+import io
+from playwright.sync_api import Playwright, sync_playwright
+from examples import (
+    BROWSERBASE_API_KEY,
+    BROWSERBASE_PROJECT_ID,
+    BROWSERBASE_CONNECT_URL,
+    bb,
+)
+
+download_re = re.compile(r"sandstorm-(\d{13})+\.mp3")
+
+
+def get_download(session_id: str) -> bytes:
+    response = bb.sessions.downloads.list(id=session_id)
+    return response.read()
+
+
+def run(playwright: Playwright):
+    # Create a session on Browserbase
+    session = bb.sessions.create(project_id=BROWSERBASE_PROJECT_ID)
+    assert session.id is not None
+    assert session.status == "RUNNING", f"Session status is {session.status}"
+
+    # Connect to the remote session
+    connect_url = (
+        f"{BROWSERBASE_CONNECT_URL}?sessionId={session.id}&apiKey={BROWSERBASE_API_KEY}"
+    )
+    browser = playwright.chromium.connect_over_cdp(connect_url)
+    context = browser.contexts[0]
+    page = context.pages[0]
+
+    # Set up CDP session for download behavior
+    client = context.new_cdp_session(page)
+    client.send(  # pyright: ignore
+        "Browser.setDownloadBehavior",
+        {
+            "behavior": "allow",
+            "downloadPath": "downloads",
+            "eventsEnabled": True,
+        },
+    )
+
+    # Navigate to the download test page
+    page.goto("https://browser-tests-alpha.vercel.app/api/download-test")
+
+    # Start download and wait for it to complete
+    with page.expect_download() as download_info:
+        page.locator("#download").click()
+    download = download_info.value
+
+    # Check for download errors
+    download_error = download.failure()
+    if download_error:
+        raise Exception(f"Download for session {session.id} failed: {download_error}")
+
+    page.close()
+    browser.close()
+
+    # Verify the download
+    zip_buffer = get_download(session.id)
+    if len(zip_buffer) == 0:
+        raise Exception(f"Download buffer is empty for session {session.id}")
+
+    zip_file = zipfile.ZipFile(io.BytesIO(zip_buffer))
+    zip_entries = zip_file.namelist()
+    mp3_entry = next((entry for entry in zip_entries if download_re.match(entry)), None)
+
+    if not mp3_entry:
+        raise Exception(
+            f"Session {session.id} is missing a file matching '{download_re.pattern}' in its zip entries: {zip_entries}"
+        )
+
+    expected_file_size = 6137541
+    actual_file_size = zip_file.getinfo(mp3_entry).file_size
+    assert (
+        actual_file_size == expected_file_size
+    ), f"Expected file size {expected_file_size}, but got {actual_file_size}"
+
+    print("Download test passed successfully!")
+
+
+if __name__ == "__main__":
+    with sync_playwright() as playwright:
+        run(playwright)