browser-automation

# Synchronous (preferred for simple scripts)
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(
        viewport={"width": 1280, "height": 720},
        user_agent="Mozilla/5.0 (compatible; AutomationBot/1.0)"
    )
    page = context.new_page()

# Async (required for pytest-playwright, larger suites)
import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()

asyncio.run(main())

# Basic navigation with timeout
try:
    page.goto("https://example.com", wait_until="domcontentloaded", timeout=30000)
except playwright._impl._api_types.TimeoutError:
    print("Navigation timed out — site may be down or slow")
    raise

# Wait for network idle (SPA-heavy pages)
page.goto("https://spa-app.example.com", wait_until="networkidle")

# Useful post-navigation waits
page.wait_for_load_state("domcontentloaded")   # HTML parsed
page.wait_for_load_state("load")                # all resources loaded
page.wait_for_load_state("networkidle")         # no network for 500ms

# 1. data-testid (most stable)
page.click("[data-testid='submit-button']")

# 2. Role-based (accessible)
page.get_by_role("button", name="Submit").click()

# 3. Text content (user-visible stable)
page.get_by_text("Sign In").click()

# 4. Label (forms)
page.get_by_label("Email address").fill("user@example.com")

# 5. Placeholder
page.get_by_placeholder("Search...").fill("query")

# 6. CSS/XPath (last resort)
page.locator("form#login input[name='email']").fill("user@example.com")

# Form filling
await page.get_by_label("Email").fill("user@example.com")
await page.get_by_label("Password").fill("s3cret")
await page.get_by_role("button", name="Log In").click()

# Dropdown / select
await page.select_option("select#country", value="DE")

# Checkbox / radio
await page.get_by_label("I agree to terms").check()

# File upload
await page.set_input_files("input[type='file']", "/path/to/file.pdf")

# Hover and nested interactions
await page.get_by_text("Products").hover()
await page.get_by_text("New Arrivals").click()

# Keyboard shortcuts
await page.keyboard.press("Enter")
await page.keyboard.press("Control+A")

# Never use time.sleep(). Use these instead:
await page.wait_for_selector("[data-testid='result']", state="visible", timeout=10000)
await page.wait_for_function("() => document.querySelector('.spinner') === null")
await page.wait_for_url("**/dashboard**")
await page.wait_for_load_state("networkidle")

# For dynamic content that appears/disappears
await expect(page.get_by_text("Loading...")).to_be_hidden(timeout=15000)
await expect(page.get_by_text("Results")).to_be_visible(timeout=15000)

# For network-triggered updates
async with page.expect_response(lambda r: "/api/results" in r.url):
    await page.click("[data-testid='search-button']")

from playwright.async_api import expect

# Page-level
await expect(page).to_have_title("Dashboard — My App")
await expect(page).to_have_url("https://app.example.com/dashboard")

# Element visibility
await expect(page.get_by_text("Welcome back")).to_be_visible()
await expect(page.locator(".error-banner")).to_be_hidden()

# Content
await expect(page.get_by_test_id("user-name")).to_have_text("John Doe")
await expect(page.get_by_test_id("item-count")).to_contain_text("5")

# Form state
await expect(page.get_by_label("Email")).to_have_value("user@example.com")
await expect(page.get_by_label("Agree")).to_be_checked()

# Screenshot-based verification
screenshot = await page.screenshot(full_page=True)
# For visual regression, combine with pixelmatch or Percy

# Custom assertions for scraping
items = await page.locator(".product-card").count()
assert items >= 10, f"Expected at least 10 products, found {items}"

# With context managers (recommended)
async with async_playwright() as p:
    async with await p.chromium.launch() as browser:
        async with await browser.new_page() as page:
            await page.goto("https://example.com")
            # ... work ...
# Everything auto-closes at block exit

# Manual cleanup (when not using context managers)
await page.close()
await context.close()
await browser.close()
await p.stop()  # playwright instance

import asyncio
from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeout

async def robust_navigation(url: str, retries: int = 2):
    """Navigate with retry logic for flaky networks."""
    for attempt in range(retries + 1):
        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=15000)
            return
        except PlaywrightTimeout:
            if attempt == retries:
                raise
            print(f"Navigation attempt {attempt + 1} failed, retrying...")
            await asyncio.sleep(2 ** attempt)  # exponential backoff

# Stale element recovery
try:
    await page.click("[data-testid='dynamic-button']")
except PlaywrightTimeout:
    # Element may have been removed and re-rendered
    await page.wait_for_selector("[data-testid='dynamic-button']", state="attached")
    await page.click("[data-testid='dynamic-button']")

# Network failure handling
try:
    await page.goto("https://flaky-service.example.com")
except Exception as e:
    if "net::ERR_" in str(e):
        raise RuntimeError(f"Network error accessing page: {e}")
    raise

# Modal/dialog handling (accept before interaction)
page.on("dialog", lambda dialog: dialog.accept())

import time
from typing import List

class RespectfulScraper:
    """Scraper with built-in politeness delays."""

    def __init__(self, min_delay: float = 1.0, max_delay: float = 3.0):
        self.min_delay = min_delay
        self.max_delay = max_delay
        self._last_request = 0.0
        self._jitter = __import__("random").uniform

    async def wait(self):
        """Enforce minimum delay between requests."""
        elapsed = time.monotonic() - self._last_request
        if elapsed < self.min_delay:
            await asyncio.sleep(self.min_delay - elapsed + self._jitter(0, self.max_delay - self.min_delay))
        self._last_request = time.monotonic()

    async def get(self, page, url: str) -> None:
        await self.wait()
        await page.goto(url)

# ✅ DO: Read from environment variables
EMAIL = os.environ["TEST_EMAIL"]
PASSWORD = os.environ["TEST_PASSWORD"]

# ✅ DO: Read from a .env file (gitignored)
from dotenv import load_dotenv
load_dotenv()

# ❌ DON'T: Hardcode credentials
# EMAIL = "admin@company.com"  # NEVER DO THIS

# ❌ DON'T: Log credentials
# print(f"Logging in as {EMAIL} with password {PASSWORD}")  # NEVER

import os
from playwright.sync_api import sync_playwright, expect

BASE_URL = os.environ.get("BASE_URL", "https://app.example.com")
EMAIL = os.environ["TEST_EMAIL"]
PASSWORD = os.environ["TEST_PASSWORD"]

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()

    # Navigate
    page.goto(f"{BASE_URL}/login", wait_until="networkidle")

    # Fill & submit
    page.get_by_label("Email").fill(EMAIL)
    page.get_by_label("Password").fill(PASSWORD)
    page.get_by_role("button", name="Sign In").click()

    # Assert redirect
    expect(page).to_have_url(f"{BASE_URL}/dashboard", timeout=10000)
    expect(page.get_by_text("Welcome")).to_be_visible()

    browser.close()
    print("✅ Login test passed")

import json
import sys
from playwright.sync_api import sync_playwright

URL = sys.argv[1] if len(sys.argv) > 1 else "https://books.toscrape.com"

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto(URL, wait_until="domcontentloaded")

    books = page.locator(".product_pod").evaluate_all("""
        els => els.map(el => ({
            title: el.querySelector('h3 a')?.getAttribute('title') || '',
            price: el.querySelector('.price_color')?.textContent || '',
            availability: el.querySelector('.availability')?.textContent?.trim() || ''
        }))
    """)

    print(json.dumps(books, indent=2))
    browser.close()

import os
import sys
from datetime import datetime, timezone
from playwright.sync_api import sync_playwright

URL = os.environ.get("MONITOR_URL", "https://example.com")

def check_flow() -> bool:
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()

        try:
            # Step 1: Homepage loads
            page.goto(URL, wait_until="domcontentloaded", timeout=15000)
            assert page.title(), "Page has no title"

            # Step 2: Search works
            page.get_by_placeholder("Search").fill("test")
            page.keyboard.press("Enter")
            page.wait_for_load_state("networkidle")

            # Step 3: Results appear
            assert page.locator(".search-results").is_visible(), "No results container"
            print(f"[{datetime.now(timezone.utc).isoformat()}] ✅ Flow healthy")
            return True

        except Exception as e:
            print(f"[{datetime.now(timezone.utc).isoformat()}] ❌ Flow failed: {e}")
            # Save diagnostic screenshot
            page.screenshot(path=f"error-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}.png")
            return False
        finally:
            browser.close()

if __name__ == "__main__":
    success = check_flow()
    sys.exit(0 if success else 1)

class LoginPage:
    def __init__(self, page):
        self.page = page
        self.email_input = page.get_by_label("Email")
        self.password_input = page.get_by_label("Password")
        self.submit_button = page.get_by_role("button", name="Sign In")
        self.error_message = page.locator(".alert-error")

    async def goto(self):
        await self.page.goto("/login", wait_until="networkidle")

    async def login(self, email: str, password: str):
        await self.email_input.fill(email)
        await self.password_input.fill(password)
        await self.submit_button.click()

    async def get_error(self) -> str:
        await self.error_message.wait_for(state="visible")
        return await self.error_message.text_content()

# Mock API responses for stable tests
await page.route("**/api/users/**", lambda route: route.fulfill(
    status=200,
    content_type="application/json",
    body='{"id": 1, "name": "Test User"}'
))

# Capture network requests for debugging
requests = []
page.on("request", lambda req: requests.append(f"{req.method} {req.url}"))
page.on("response", lambda res: print(f"{res.status} {res.url}"))

# Wait for specific API call to complete
async with page.expect_response(lambda r: "/api/submit" in r.url) as response_info:
    await page.click("[data-testid='submit']")
response = await response_info.value
assert response.status == 200

# Capture and compare screenshots
await page.screenshot(path="baseline.png", full_page=True)
# Use with pixelmatch, Percy, or Chromatic for automated diffing

# Element-level screenshot
await page.locator(".pricing-table").screenshot(path="pricing.png")

# Clip to a specific region (avoid dynamic content)
await page.screenshot(
    path="header.png",
    clip={"x": 0, "y": 0, "width": 1280, "height": 200}
)