Back to Repositories

Testing Connection Retry Mechanism in AutoGPT

This test suite validates the connection retry functionality in AutoGPT’s backend, covering both synchronous and asynchronous function scenarios. The tests ensure proper retry behavior with configurable attempts and wait times while handling errors appropriately.

Test Coverage Overview

The test suite provides comprehensive coverage of the conn_retry decorator functionality:

Synchronous function retry behavior testing
Asynchronous function retry behavior with asyncio
Maximum retry limit validation
Error handling and propagation
Configurable wait time parameters

Implementation Analysis

The testing approach utilizes pytest’s powerful fixtures and async support:

Decorator pattern testing with both sync and async functions
Local state management using nonlocal variables
Pytest’s asyncio marker for async test cases
Controlled error simulation for retry scenarios

Technical Details

Testing infrastructure includes:

pytest framework with asyncio plugin
Custom conn_retry decorator implementation
Configurable retry parameters (max_retry, max_wait, min_wait)
Error assertion using pytest.raises context manager

Best Practices Demonstrated

The test suite exemplifies several testing best practices:

Isolated test cases for sync and async scenarios
Proper error condition validation
Controlled environment with deterministic outcomes
Clear test case organization and naming
Effective use of pytest features and assertions

significant-gravitas/autogpt

autogpt_platform/backend/test/util/test_retry.py

            
import asyncio

import pytest

from backend.util.retry import conn_retry


def test_conn_retry_sync_function():
    retry_count = 0

    @conn_retry("Test", "Test function", max_retry=2, max_wait=0.1, min_wait=0.1)
    def test_function():
        nonlocal retry_count
        retry_count -= 1
        if retry_count > 0:
            raise ValueError("Test error")
        return "Success"

    retry_count = 2
    res = test_function()
    assert res == "Success"

    retry_count = 100
    with pytest.raises(ValueError) as e:
        test_function()
        assert str(e.value) == "Test error"


@pytest.mark.asyncio
async def test_conn_retry_async_function():
    retry_count = 0

    @conn_retry("Test", "Test function", max_retry=2, max_wait=0.1, min_wait=0.1)
    async def test_function():
        nonlocal retry_count
        await asyncio.sleep(1)
        retry_count -= 1
        if retry_count > 0:
            raise ValueError("Test error")
        return "Success"

    retry_count = 2
    res = await test_function()
    assert res == "Success"

    retry_count = 100
    with pytest.raises(ValueError) as e:
        await test_function()
        assert str(e.value) == "Test error"