Boto3Plugin Guide

Boto3Plugin intercepts botocore.client.BaseClient._make_api_call at the class level. Each AWS service:operation pair has its own independent FIFO queue, so you can mock multiple calls to different (or the same) API operations and they are consumed in registration order.

Installation

pip install python-tripwire[boto3]

This installs botocore.

Setup

In pytest, access Boto3Plugin through the tripwire.boto3 proxy. It auto-creates the plugin for the current test on first use:

import tripwire

def test_s3_get_object():
    tripwire.boto3.mock_call(
        "s3", "GetObject",
        returns={"Body": b"file-contents", "ContentLength": 13},
    )

    with tripwire:
        import boto3
        client = boto3.client("s3")
        response = client.get_object(Bucket="my-bucket", Key="data.csv")

    assert response["ContentLength"] == 13

    tripwire.boto3.assert_boto3_call(
        service="s3",
        operation="GetObject",
        params={"Bucket": "my-bucket", "Key": "data.csv"},
    )

For manual use outside pytest, construct Boto3Plugin explicitly:

from tripwire import StrictVerifier
from tripwire.plugins.boto3_plugin import Boto3Plugin

verifier = StrictVerifier()
boto3 = Boto3Plugin(verifier)

Each verifier may have at most one Boto3Plugin. A second Boto3Plugin(verifier) raises ValueError.

Registering mocks

Use tripwire.boto3.mock_call(service, operation, *, returns, ...) to register a mock before entering the sandbox:

tripwire.boto3.mock_call("sqs", "SendMessage", returns={"MessageId": "abc123"})
tripwire.boto3.mock_call("dynamodb", "PutItem", returns={})

Parameters

Parameter	Type	Default	Description
service	str	required	AWS service name (e.g., "s3", "sqs", "dynamodb")
operation	str	required	API operation name in PascalCase (e.g., "GetObject", "SendMessage")
returns	Any	required	Value to return when this mock is consumed
raises	BaseException \| None	None	Exception to raise instead of returning
required	bool	True	Whether an unused mock causes UnusedMocksError at teardown

FIFO queues

Each service:operation pair has its own independent FIFO queue. Multiple mock_call("s3", "GetObject", ...) calls are consumed in registration order:

def test_multiple_s3_gets():
    tripwire.boto3.mock_call(
        "s3", "GetObject",
        returns={"Body": b"first", "ContentLength": 5},
    )
    tripwire.boto3.mock_call(
        "s3", "GetObject",
        returns={"Body": b"second", "ContentLength": 6},
    )

    with tripwire:
        import boto3
        client = boto3.client("s3")
        r1 = client.get_object(Bucket="bucket", Key="a.txt")
        r2 = client.get_object(Bucket="bucket", Key="b.txt")

    assert r1["Body"] == b"first"
    assert r2["Body"] == b"second"

    tripwire.boto3.assert_boto3_call(
        service="s3", operation="GetObject",
        params={"Bucket": "bucket", "Key": "a.txt"},
    )
    tripwire.boto3.assert_boto3_call(
        service="s3", operation="GetObject",
        params={"Bucket": "bucket", "Key": "b.txt"},
    )

Asserting interactions

Use the assert_boto3_call helper on tripwire.boto3. All three fields (service, operation, params) are required:

assert_boto3_call(service, operation, *, params)

tripwire.boto3.assert_boto3_call(
    service="sqs",
    operation="SendMessage",
    params={"QueueUrl": "https://sqs.us-east-1.amazonaws.com/123/my-queue", "MessageBody": "hello"},
)

Parameter	Type	Default	Description
service	str	required	AWS service name
operation	str	required	API operation name in PascalCase
params	dict[str, Any]	required	The API parameters passed to the call

Simulating errors

Use the raises parameter to simulate AWS service errors:

from botocore.exceptions import ClientError
import tripwire

def test_s3_not_found():
    error_response = {"Error": {"Code": "NoSuchKey", "Message": "The specified key does not exist."}}
    tripwire.boto3.mock_call(
        "s3", "GetObject",
        returns=None,
        raises=ClientError(error_response, "GetObject"),
    )

    with tripwire:
        import boto3
        client = boto3.client("s3")
        with pytest.raises(ClientError) as exc_info:
            client.get_object(Bucket="my-bucket", Key="missing.csv")

    assert exc_info.value.response["Error"]["Code"] == "NoSuchKey"

    tripwire.boto3.assert_boto3_call(
        service="s3", operation="GetObject",
        params={"Bucket": "my-bucket", "Key": "missing.csv"},
    )

Full example

Production code (examples/boto3_service/app.py):

"""S3 upload with SQS notification."""

import boto3


def upload_and_notify(bucket, key, body, queue_url):
    """Upload a file to S3 and send a notification to SQS."""
    s3 = boto3.client("s3", region_name="us-east-1")
    sqs = boto3.client("sqs", region_name="us-east-1")
    s3.put_object(Bucket=bucket, Key=key, Body=body)
    sqs.send_message(QueueUrl=queue_url, MessageBody=f"Uploaded {key}")

Test (examples/boto3_service/test_app.py):

"""Test boto3 S3 upload with SQS notification using tripwire boto3_mock."""

import logging

import pytest

import tripwire

from .app import upload_and_notify


@pytest.fixture(autouse=True)
def _silence_botocore():
    """Suppress botocore DEBUG logs that would generate dozens of LoggingPlugin interactions."""
    for name in ("botocore", "boto3", "urllib3"):
        logging.getLogger(name).setLevel(logging.WARNING)


def test_upload_and_notify():
    tripwire.boto3.mock_call("s3", "PutObject", returns={})
    tripwire.boto3.mock_call("sqs", "SendMessage", returns={"MessageId": "msg-001"})

    with tripwire:
        upload_and_notify(
            "data-bucket", "reports/q1.csv", b"revenue,100",
            "https://sqs.us-east-1.amazonaws.com/123/notifications",
        )

    tripwire.boto3.assert_boto3_call(
        service="s3", operation="PutObject",
        params={"Bucket": "data-bucket", "Key": "reports/q1.csv", "Body": b"revenue,100"},
    )
    tripwire.boto3.assert_boto3_call(
        service="sqs", operation="SendMessage",
        params={
            "QueueUrl": "https://sqs.us-east-1.amazonaws.com/123/notifications",
            "MessageBody": "Uploaded reports/q1.csv",
        },
    )

Optional mocks

Mark a mock as optional with required=False:

tripwire.boto3.mock_call("cloudwatch", "PutMetricData", returns={}, required=False)

An optional mock that is never triggered does not cause UnusedMocksError at teardown.

UnmockedInteractionError

When code calls a boto3 API operation that has no remaining mocks in its queue, tripwire raises UnmockedInteractionError:

s3.GetObject(...) was called but no mock was registered.
Register a mock with:
    tripwire.boto3.mock_call('s3', 'GetObject', returns=...)