Run Python

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()}

Special Supatest functions

AI-powered functions available in the runtime:

ai_assertion(user_instruction): Performs AI-powered visual assertions on the current page. Returns True if assertion passes, raises Exception if it fails.
ai_condition(condition_prompt): Evaluates a condition using AI to determine if it’s true or false. Returns bool.
ai_browser_action(user_prompt): Executes browser automation tasks using AI agent. Returns a dict with success, is_done, execution_time, etc.

async def supatest_run_code():
    # AI assertion
    await ai_assertion("Verify the user sees a success message after login")
    
    # AI condition check
    is_error_visible = await ai_condition("Is there an error message on the page?")
    if is_error_visible:
        raise Exception("Error detected on page")
    
    # AI browser action
    result = await ai_browser_action("Fill the registration form with test data")
    if not result.get('success'):
        raise Exception(f"Action failed: {result.get('error')}")

Utility functions

One-line helper functions available in global scope: Date & Time:

now(): Current datetime
today(): Current date
timestamp(): Unix timestamp
timedelta(days=...): Create time deltas
format_date(date_obj, fmt="%d/%m/%Y"): Format dates

String, Encoding & Crypto:

uuid(): Generate random UUID
md5(text): MD5 hash
sha256(text): SHA256 hash
hmac_sha256(key, msg): HMAC signature
b64encode(text) / b64decode(text): Base64 operations
urlencode(text) / urldecode(text): URL encoding/decoding
slugify(text): Convert text to slug

Math & Random:

randint(min, max): Random integer
choice(list): Pick random item
sample(list, k): Pick k random items

async def supatest_run_code():
    # Date utilities
    current_time = now()
    formatted = format_date(today(), "%Y-%m-%d")
    
    # Encoding
    encoded = b64encode("secret data")
    hashed = sha256("password123")
    
    # Random
    random_id = randint(1, 1000)
    item = choice(["option1", "option2", "option3"])

Available libraries

Standard Library (always available): asyncio, json, re, csv, io, os, sys, tempfile, time, datetime, pathlib, urllib.parse, base64, hashlib, random, string, collections, itertools, functools, typing Third-party (pre-imported):

playwright.async_api: Full Playwright async API
faker: Faker class for generating test data
pytest: Test framework utilities

Available for import:

aiofiles: Async file I/O operations
aiohttp: HTTP requests (use ClientSession for async operations)
ssl, certifi: SSL/TLS certificate handling
asyncpg: PostgreSQL database connections
motor: MongoDB database connections (import as motor.motor_asyncio)
beautifulsoup4: HTML parsing (import as bs4)
pydantic: Data validation and parsing

async def supatest_run_code():
    import aiohttp
    import asyncpg
    import motor.motor_asyncio
    import aiofiles
    
    # HTTP requests
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            data = await response.json()
    
    # PostgreSQL queries
    conn = await asyncpg.connect(env.DATABASE_URL)
    try:
        rows = await conn.fetch("SELECT * FROM users")
    finally:
        await conn.close()
    
    # MongoDB queries
    client = motor.motor_asyncio.AsyncIOMotorClient(env.MONGODB_URL)
    try:
        db = client[env.MONGODB_DATABASE]
        collection = db['users']
        users = await collection.find({}).to_list(length=100)
    finally:
        client.close()
    
    # File operations
    async with aiofiles.open('file.txt', 'r') as f:
        content = await f.read()

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)}

AI-powered assertions and actions

async def supatest_run_code():
    # Use AI to verify complex UI states
    await ai_assertion("Verify the checkout page shows order summary with correct total")
    
    # Conditional logic based on AI condition evaluation
    if await ai_condition("Is the payment form visible and ready for input?"):
        result = await ai_browser_action("Fill payment form with test card details")
        if result.get('success'):
            print("Payment form filled successfully")
    else:
        raise Exception("Payment form not ready")

Using utility functions

async def supatest_run_code():
    # Generate unique identifiers
    user_id = uuid()
    session_token = b64encode(f"{user_id}:{timestamp()}")
    
    # Date manipulation
    tomorrow = today() + timedelta(days=1)
    formatted_date = format_date(tomorrow, "%Y-%m-%d")
    
    # Random selection
    status = choice(["active", "inactive", "pending"])
    sample_ids = sample(range(1, 100), 5)
    
    return {
        "token": session_token,
        "expires": formatted_date,
        "status": status,
        "ids": sample_ids
    }

Response Format

When you specify a Return variable name (e.g., pythonResult), the value returned by your function is stored and can be accessed in subsequent steps.

Simple Values

async def supatest_run_code():
    return "success"

Stored value: "success" Access in later steps: ${pythonResult} → "success"

Numbers

async def supatest_run_code():
    return 42

Stored value: 42 Access in later steps: ${pythonResult} → 42

Objects (Dictionaries)

async def supatest_run_code():
    return {
        "userId": 123,
        "email": "[email protected]",
        "active": True
    }

Stored value:

{
  "userId": 123,
  "email": "[email protected]",
  "active": true
}

Access in later steps:

User ID: ${pythonResult.userId} → 123
Email: ${pythonResult.email} → "[email protected]"
Active status: ${pythonResult.active} → true

Arrays (Lists)

async def supatest_run_code():
    return ["apple", "banana", "cherry"]

Stored value:

["apple", "banana", "cherry"]

Access in later steps:

First item: ${pythonResult[0]} → "apple"
Array length: ${pythonResult.length} → 3

Complex Nested Data

async def supatest_run_code():
    res = await fetch(f"{env.API_BASE}/users/123")
    api_data = await res.json()

    ui_name = await page.text_content('.user-name')

    return {
        "api": api_data,
        "ui": {"displayName": ui_name},
        "match": api_data['name'] == ui_name
    }

Example stored value:

{
  "api": {
    "id": 123,
    "name": "John Doe",
    "email": "[email protected]"
  },
  "ui": {
    "displayName": "John Doe"
  },
  "match": true
}

Access in later steps:

API user name: ${pythonResult.api.name}
UI display name: ${pythonResult.ui.displayName}
Verification result: ${pythonResult.match}

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,
    }

Note File downloads will not work when running tests in the interactive editor due to limitations with remote browsers. However, they will work when running tests in headless mode from the test details page or in plans.

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

The runtime includes several third-party libraries for common tasks. See the Available libraries section above for the complete list.

Setting up environment variables for database connections

Before using database connections in your Run Python steps, you need to configure the necessary credentials:

Add environment variables: Go to your Supatest environment settings and add the required connection strings and credentials (e.g., DATABASE_URL for PostgreSQL, MONGODB_URL and MONGODB_DATABASE for MongoDB).
Select the environment: Make sure the correct environment is selected before running your test, as the code will access these variables using env.VARIABLE_NAME or env['VARIABLE_NAME'] syntax.
Access in code: Use env.DATABASE_URL, env.MONGODB_URL, etc. to access your configured credentials within the Run Python step.

Example: Query PostgreSQL with asyncpg

async def supatest_run_code():
    import asyncpg

    # Connect to PostgreSQL database
    conn = await asyncpg.connect(env.DATABASE_URL)
    try:
        # Test connection
        await conn.fetchval("SELECT 1")
        print("✅ Connected to PostgreSQL")
        
        # Example query - write your own application-specific query here
        rows = await conn.fetch("SELECT id, email, status FROM users WHERE status = $1", "active")
        print(f"📊 Found {len(rows)} active users")        
        user_emails = [row['email'] for row in rows]
        
        return {
            "db_ok": True,
            "active_users_count": len(rows),
            "user_emails": user_emails
        }
    finally:
        await conn.close()

Example: Query MongoDB with motor

async def supatest_run_code():
    import motor.motor_asyncio

    # Connect to MongoDB database
    client = motor.motor_asyncio.AsyncIOMotorClient(env.MONGODB_URL)
    try:
        # Get database and collection
        db = client[env.MONGODB_DATABASE]
        collection = db['movies']
        
        # Test connection
        await client.admin.command('ping')
        print("✅ Connected to MongoDB")
        
        # Example query - write your own application-specific query here
        cursor = collection.find({"year": {"$gte": 2000}}).limit(10)
        movies = await cursor.to_list(length=10)
        print(f"📊 Found {len(movies)} movies from year 2000 onwards")
        movie_titles = [movie.get('title') for movie in movies]
        
        return {
            "db_ok": True,
            "movies_count": len(movies),
            "sample_titles": movie_titles
        }
    finally:
        client.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.

Interactions

Navigation

Verification

Utilities

Technical

Overview

Execution model

Capabilities and globals

Special Supatest functions

Utility functions

Available libraries

Environment variables

When to use

Usage (form + editor)

AI code generation

Examples

Basic page interaction

API and UI comparison

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

AI-powered assertions and actions

Using utility functions

Response Format

Simple Values

Numbers

Objects (Dictionaries)

Arrays (Lists)

Complex Nested Data

Download file and parse CSV

Best practices

Common issues and fixes

Third Party Libraries

Setting up environment variables for database connections

Example: Query PostgreSQL with asyncpg

Example: Query MongoDB with motor

Example: Parse CSV from API using aiofiles + csv

Example: Stream parse a local CSV file using aiofiles

Interactions

Navigation

Verification

Utilities

Technical

​Overview

​Execution model

​Capabilities and globals

​Special Supatest functions

​Utility functions

​Available libraries

​Environment variables

​When to use

​Usage (form + editor)

​AI code generation

​Examples

​Basic page interaction

​API and UI comparison

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

​AI-powered assertions and actions

​Using utility functions

​Response Format

​Simple Values

​Numbers

​Objects (Dictionaries)

​Arrays (Lists)

​Complex Nested Data

​Download file and parse CSV

​Best practices

​Common issues and fixes

​Third Party Libraries

​Setting up environment variables for database connections

​Example: Query PostgreSQL with asyncpg

​Example: Query MongoDB with motor

​Example: Parse CSV from API using aiofiles + csv

​Example: Stream parse a local CSV file using aiofiles

​Related steps

Overview

Execution model

Capabilities and globals

Special Supatest functions

Utility functions

Available libraries

Environment variables

When to use

Usage (form + editor)

AI code generation

Examples

Basic page interaction

API and UI comparison

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

AI-powered assertions and actions

Using utility functions

Response Format

Simple Values

Numbers

Objects (Dictionaries)

Arrays (Lists)

Complex Nested Data

Download file and parse CSV

Best practices

Common issues and fixes

Third Party Libraries

Setting up environment variables for database connections

Example: Query PostgreSQL with asyncpg

Example: Query MongoDB with motor

Example: Parse CSV from API using aiofiles + csv

Example: Stream parse a local CSV file using aiofiles

Related steps