Boto3Plugin

Boto3Plugin

Boto3Plugin(verifier)

Bases: BasePlugin

boto3/botocore interception plugin.

Patches botocore.client.BaseClient._make_api_call at the class level. Uses reference counting so nested sandboxes work correctly.

Each service:operation pair has its own FIFO deque of Boto3MockConfig objects.

Source code in src/tripwire/plugins/boto3_plugin.py

def __init__(self, verifier: StrictVerifier) -> None:
    super().__init__(verifier)
    self._queues: dict[str, deque[Boto3MockConfig]] = {}
    self._registry_lock: threading.Lock = threading.Lock()

mock_call

mock_call(service, operation, *, returns, raises=None, required=True)

Register a mock for a single boto3 API call invocation.

Args: service: The AWS service name (e.g., "s3"). operation: The API operation name in PascalCase (e.g., "GetObject"). returns: Value to return when this mock is consumed. raises: If provided, this exception is raised instead of returning. required: If False, the mock is not reported as unused at teardown.

Source code in src/tripwire/plugins/boto3_plugin.py

def mock_call(
    self,
    service: str,
    operation: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a single boto3 API call invocation.

    Args:
        service: The AWS service name (e.g., "s3").
        operation: The API operation name in PascalCase (e.g., "GetObject").
        returns: Value to return when this mock is consumed.
        raises: If provided, this exception is raised instead of returning.
        required: If False, the mock is not reported as unused at teardown.
    """
    config = Boto3MockConfig(
        service=service,
        operation=operation,
        returns=returns,
        raises=raises,
        required=required,
    )
    queue_key = f"{service}:{operation}"
    with self._registry_lock:
        if queue_key not in self._queues:
            self._queues[queue_key] = deque()
        self._queues[queue_key].append(config)

sentinel

sentinel(service, operation)

Return a sentinel for a specific service:operation pair.

Source code in src/tripwire/plugins/boto3_plugin.py

def sentinel(self, service: str, operation: str) -> _Boto3Sentinel:
    """Return a sentinel for a specific service:operation pair."""
    return _Boto3Sentinel(service, operation)

install_patches

install_patches()

Install botocore._make_api_call patch and set dummy AWS credentials.

Setting dummy credentials prevents botocore's credential provider from hitting the EC2 metadata service (169.254.169.254), which would leak DNS and HTTP calls to other plugin interceptors.

Source code in src/tripwire/plugins/boto3_plugin.py

def install_patches(self) -> None:
    """Install botocore._make_api_call patch and set dummy AWS credentials.

    Setting dummy credentials prevents botocore's credential provider from
    hitting the EC2 metadata service (169.254.169.254), which would leak DNS
    and HTTP calls to other plugin interceptors.
    """
    if not _BOTO3_AVAILABLE:
        raise ImportError(
            "Install python-tripwire[boto3] to use Boto3Plugin: "
            "pip install python-tripwire[boto3]"
        )
    # Save current env values and inject dummy credentials
    for key, value in self._CREDENTIAL_ENV_VARS.items():
        Boto3Plugin._saved_env[key] = os.environ.get(key)
        os.environ[key] = value

    Boto3Plugin._original_make_api_call = botocore.client.BaseClient._make_api_call
    botocore.client.BaseClient._make_api_call = _patched_make_api_call

restore_patches

restore_patches()

Restore original botocore._make_api_call and AWS credential env vars.

Source code in src/tripwire/plugins/boto3_plugin.py

def restore_patches(self) -> None:
    """Restore original botocore._make_api_call and AWS credential env vars."""
    if Boto3Plugin._original_make_api_call is not None:
        botocore.client.BaseClient._make_api_call = Boto3Plugin._original_make_api_call
        Boto3Plugin._original_make_api_call = None

    # Restore original env values
    for key, original_value in Boto3Plugin._saved_env.items():
        if original_value is None:
            os.environ.pop(key, None)
        else:
            os.environ[key] = original_value
    Boto3Plugin._saved_env.clear()

assert_boto3_call

assert_boto3_call(service, operation, *, params)

Typed helper: assert the next boto3 API call interaction.

Wraps assert_interaction() for ergonomic use. All three fields (service, operation, params) are required.

Source code in src/tripwire/plugins/boto3_plugin.py

def assert_boto3_call(
    self,
    service: str,
    operation: str,
    *,
    params: dict[str, Any],
) -> None:
    """Typed helper: assert the next boto3 API call interaction.

    Wraps assert_interaction() for ergonomic use. All three fields
    (service, operation, params) are required.
    """
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    sentinel = _Boto3Sentinel(service, operation)
    _get_test_verifier_or_raise().assert_interaction(
        sentinel,
        service=service,
        operation=operation,
        params=params,
    )