Skip to main content

Overview

The Run Python step lets you execute custom Python code within a test with full Playwright access. Use it for complex logic, validations, data processing, and API/UI cross‑checks.
Note
  • Do not use double braces {{ }} inside this step. Instead, env, vars, and random are provided directly as in‑memory objects.
  • env.NAME is read‑only, vars.name (alias testVars.name) contains values produced by earlier steps, and random.* provides generators.

Execution model

  • Wrap code in async def supatest_run_code(...):
  • Runs in the test’s browser context with Playwright available; always await async calls
  • Optional inputs come from selected variables; the function’s return value is saved to a named variable
  • Uses the step timeout; prefer explicit waits over fixed delays
  • print() output appears in Raw Logs; failures include stack traces when available

Capabilities and globals

Available during execution:
  • Playwright: page, expect
  • Networking: fetch (JSON/ArrayBuffer)
  • Supatest: env, update_env(name, value), vars (alias testVars), random
async def supatest_run_code():
    await page.goto(env.APP_URL)
    await expect(page).to_have_url(re.compile(r"dashboard"))

    res = await fetch(f"{env.API_BASE}/health")
    return {"api_status": res.status, "user": vars.current_user, "email": random.email()}

Environment variables

  • env.KEY is a read‑only view of the selected environment at the start of the run.
  • update_env(name, value) persists changes to existing variables in the selected environment (org‑scoped).
    • Missing or failed updates cause the test to fail.
    • Raw Logs show per‑call queued updates and one final summary.
async def supatest_run_code():
    update_env('CURRENT_PASSWORD', 'New Value')

When to use

  • Conditional validations, loops, data transforms
  • Compare UI state with API responses
  • Temporary workaround when no dedicated step exists

Usage (form + editor)

  • Title: short, descriptive
  • Code: write inside async def supatest_run_code(...): with Monaco editor syntax highlighting
  • Variables (optional): select earlier variables to become function parameters
  • Return variable (optional): name for return value to use in later steps
  • Timeout (optional): override default

AI code generation

  • Prompt in plain English; code respects Playwright best practices
  • Refine with follow‑ups; review and edit before running

Examples

Minimal examples you can adapt:

Basic page interaction

async def supatest_run_code():
    await page.goto('https://example.com')
    await expect(page).to_have_title(re.compile(r"Example"))
    return await page.title()

API and UI comparison

async def supatest_run_code(api_key):
    r = await fetch(f"https://api.example.com/data?key={api_key}")
    api = await r.json()
    ui = (await page.text_content('#data-display') or '').strip()
    if api['value'] != ui:
        raise Exception(f"Mismatch: API={api['value']}, UI={ui}")
    return api['value']

Using env, vars, and random (no {{ }} required)

async def supatest_run_code():
    base = env.API_BASE
    email = random.email()
    await page.goto(f"{base}/profile")
    await page.fill('#email', email)
    await page.click('button:has-text("Save")')
    await expect(page.locator('#email')).to_have_value(email)
    return {"saved_email": email, "token_used": bool(vars.session_token)}

Download file and parse CSV

async def supatest_run_code():
    import csv
    import io
    import os
    import tempfile
    import aiofiles

    async with page.expect_download() as dl_info:
        await page.get_by_text("Download").click()
    download = await dl_info.value

    path = await download.path()
    if not path:
        fd, tmp = tempfile.mkstemp(suffix=f"_{download.suggested_filename}")
        os.close(fd)
        await download.save_as(tmp)
        path = tmp

    async with aiofiles.open(path, mode="r", encoding="utf-8", newline="") as f:
        text = await f.read()

    rows = list(csv.DictReader(io.StringIO(text)))
    return {
        "fileName": download.suggested_filename,
        "rows": len(rows),
        "firstRow": rows[0] if rows else None,
    }

Best practices

  • Structure: single entry supatest_run_code, return only what later steps need
  • Selectors: prefer stable attributes; avoid brittle position‑based locators
  • Waits: use Playwright waits (visible/attached/navigation) instead of fixed timeouts
  • Assertions: prefer expect(...) over manual checks
  • Performance: keep code concise; minimize unnecessary DOM queries and network calls

Common issues and fixes

  • Not wrapped in supatest_run_code → wrap all logic in async def supatest_run_code(...):
  • Missing await → await Playwright and promise calls
  • Return value not saved → set a Return variable name and ensure you return it
  • Timeouts → add explicit waits or increase step timeout
  • Unreliable selectors → use more specific, stable attributes (e.g., data‑testid)

Third Party Libraries (Python)

  • Database (PostgreSQL): Use asyncpg for async Postgres access from your test code.
  • CSV parsing: Use built‑in csv with aiofiles for async I/O and streaming‑friendly parsing.

Example: Query PostgreSQL with asyncpg

async def supatest_run_code():
    import asyncpg

    conn = await asyncpg.connect(env.DATABASE_URL)
    try:
        rows = await conn.fetch("SELECT 1 AS ok")
        return {"db_ok": rows[0]["ok"] == 1}
    finally:
        await conn.close()

Example: Parse CSV from API using aiofiles + csv

async def supatest_run_code():
    import csv
    import io

    # Fetch CSV from an API endpoint
    res = await fetch(f"{env.API_BASE}/invoice.csv")
    if res.status != 200:
        raise Exception(f"Failed to fetch CSV: {res.status}")

    # Convert ArrayBuffer to text for csv.DictReader
    buffer = await res.arrayBuffer()
    text = buffer.decode("utf-8") if hasattr(buffer, "decode") else bytes(buffer).decode("utf-8")

    reader = csv.DictReader(io.StringIO(text))
    rows = list(reader)

    paid_total = sum(
        float((r.get("amount") or r.get("Amount") or 0) or 0)
        for r in rows
        if (r.get("status") or r.get("Status")) == "PAID"
    )
    has_invoice_123 = any(str(r.get("invoiceNumber") or r.get("Invoice #")) == "123" for r in rows)

    return {"rows": len(rows), "paidTotal": paid_total, "hasInvoice123": has_invoice_123}

Example: Stream parse a local CSV file using aiofiles

async def supatest_run_code():
    import csv
    import aiofiles

    # Assumes a file was prepared earlier in the test run (e.g., downloaded)
    path = vars.get("local_csv_path") or "sample.csv"

    async with aiofiles.open(path, mode="r", encoding="utf-8", newline="") as f:
        header_line = await f.readline()
        if not header_line:
            raise Exception("Empty CSV")
        headers = next(csv.reader([header_line]))
        index = {name: i for i, name in enumerate(headers)}

        required = ["id", "email", "created_at"]
        missing = [k for k in required if k not in index]
        if missing:
            raise Exception(f"Missing required columns: {missing}")

        errors = []
        line_no = 2
        async for line in f:
            for row in csv.reader([line]):
                try:
                    id_val = int(row[index["id"]])
                except Exception:
                    errors.append({"line": line_no, "error": "id must be positive int"})
                    line_no += 1
                    continue

                email_val = row[index["email"]].strip() if index["email"] < len(row) else ""
                created_val = row[index["created_at"]].strip() if index["created_at"] < len(row) else ""

                if id_val <= 0:
                    errors.append({"line": line_no, "error": "id must be positive int"})
                if "@" not in email_val:
                    errors.append({"line": line_no, "error": "email must be valid"})
                # Minimal ISO‑8601 check
                if "T" not in created_val:
                    errors.append({"line": line_no, "error": "created_at must be ISO‑8601"})
            line_no += 1

    if errors:
        raise Exception({"csv_errors": errors})
    return {"validated": True}
We will expand this list over time with additional approved Python libraries.
I