Infura Node Testing

Simple load testing for Infura blockchain node endpoints with basic RPC calls

LoadForge can record your browser, graphically build tests, scan your site with a wizard and more. Sign up now to run your first test.

Sign up now


This guide shows how to load test Infura blockchain node endpoints with basic RPC calls. Perfect for testing your Infura project's performance and rate limits.

Use Cases

  • Test Infura endpoint response times
  • Validate RPC call reliability
  • Check rate limit behavior
  • Monitor blockchain data retrieval performance

Simple Implementation

from locust import task, HttpUser
import json
import random

class InfuraUser(HttpUser):
    def on_start(self):
        # Replace with your Infura project ID
        self.infura_project_id = "your-project-id-here"
        self.rpc_url = f"https://mainnet.infura.io/v3/{self.infura_project_id}"
        
        # Test addresses for balance checks
        self.test_addresses = [
            "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",  # Vitalik
            "0xA0b86a33E6441b8435b662303c0f479c7e1b5b8e",  # USDC
            "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"   # WETH
        ]

    @task(3)
    def get_latest_block(self):
        """Get the latest block number"""
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_blockNumber",
            "params": [],
            "id": 1
        }
        
        with self.client.post(
            self.rpc_url,
            json=payload,
            headers={"Content-Type": "application/json"},
            name="Get Latest Block"
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if "result" in data:
                    block_number = int(data["result"], 16)
                    print(f"Latest block: {block_number}")
                else:
                    response.failure("No result in response")

    @task(2)
    def get_gas_price(self):
        """Get current gas price"""
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_gasPrice",
            "params": [],
            "id": 1
        }
        
        with self.client.post(
            self.rpc_url,
            json=payload,
            headers={"Content-Type": "application/json"},
            name="Get Gas Price"
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if "result" in data:
                    gas_price = int(data["result"], 16) / 1e9  # Convert to Gwei
                    print(f"Gas price: {gas_price:.2f} Gwei")

    @task(2)
    def check_balance(self):
        """Check ETH balance for random address"""
        address = random.choice(self.test_addresses)
        
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_getBalance",
            "params": [address, "latest"],
            "id": 1
        }
        
        with self.client.post(
            self.rpc_url,
            json=payload,
            headers={"Content-Type": "application/json"},
            name="Check Balance"
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if "result" in data:
                    balance = int(data["result"], 16) / 1e18  # Convert to ETH
                    print(f"Balance for {address}: {balance:.4f} ETH")

    @task(1)
    def get_block_details(self):
        """Get details of the latest block"""
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_getBlockByNumber",
            "params": ["latest", False],
            "id": 1
        }
        
        with self.client.post(
            self.rpc_url,
            json=payload,
            headers={"Content-Type": "application/json"},
            name="Get Block Details"
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if "result" in data and data["result"]:
                    block = data["result"]
                    tx_count = len(block.get("transactions", []))
                    print(f"Block {block.get('number')} has {tx_count} transactions")

Setup Instructions

  1. Sign up for a free Infura account at infura.io
  2. Create a new project and get your Project ID
  3. Replace your-project-id-here with your actual Infura Project ID
  4. Run the test to monitor your Infura endpoint performance

What This Tests

  • RPC Response Times: Measures how fast Infura responds to different calls
  • Rate Limiting: Tests how Infura handles multiple concurrent requests
  • Data Accuracy: Validates that blockchain data is returned correctly
  • Endpoint Reliability: Checks for any connection issues or timeouts

Common Issues

  • Rate Limits: Free Infura accounts have daily request limits
  • Network Delays: Blockchain networks can have variable response times
  • Invalid Responses: Always check for "result" field in JSON responses

Ready to run your test?
Launch your locust test at scale.