Repository: pydantic/pydantic-ai
File: tests/models/test_bedrock.py (Python)

Code before:
            PartEndEvent(
                index=0,
                part=ThinkingPart(
                    content='The user just says "Hi". We need to respond appropriately, friendly greeting. No special instructions. Should be short.'
                ),
                next_part_kind='text',
            ),
            PartStartEvent(index=1, part=TextPart(content='Hello! How can I help'), previous_part_kind='thinking'),
            FinalResultEvent(tool_name=None, tool_call_id=None),
            PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' you today?')),
            PartEndEvent(index=1, part=TextPart(content='Hello! How can I help you today?')),
        ]
    )


def test_bedrock_throttling_error(bedrock_provider: BedrockProvider):
    """Test that ThrottlingException converts to ModelHTTPError with 429 status."""
    expected_error = {'Code': 'ThrottlingException', 'Message': 'Rate exceeded'}
    expected_status = 429
    model_id = 'us.amazon.nova.micro-v1:0'

    # Mock the client to raise throttling exception
    error_response = {'Error': expected_error, 'ResponseMetadata': {'HTTPStatusCode': expected_status}}
    mock_client = Mock()
    mock_client.converse.side_effect = ClientError(error_response, 'Converse')  # type: ignore[reportArgumentType]
    mock_client.meta.endpoint_url = 'https://bedrock-runtime.us-east-1.amazonaws.com'

    # Create provider with mocked client
    from pydantic_ai.providers.bedrock import BedrockProvider

    provider = BedrockProvider(bedrock_client=mock_client)
    model = BedrockConverseModel(model_id, provider=provider)
    agent = Agent(model)

    with pytest.raises(ModelHTTPError) as exc_info:
        agent.run_sync('hello')

    assert exc_info.value.status_code == expected_status
    assert exc_info.value.model_name == model_id
    assert exc_info.value.body == expected_error


def test_bedrock_server_error():
    """Test that 5xx errors convert to ModelHTTPError."""
    expected_error = {'Code': 'InternalServerError', 'Message': 'Internal error'}
    expected_status = 500
    model_id = 'us.amazon.nova.micro-v1:0'

    # Mock the client to raise throttling exception
    error_response = {'Error': expected_error, 'ResponseMetadata': {'HTTPStatusCode': expected_status}}
    mock_client = Mock()

Review comment (bug):
Instead of mocking these requests, can we have the tests make real requests that will result in errors (like using a wrong model name), and then use VCR to capture the request/response in cassettes? See all of the tests with `allow_model_requests: None` as examples.

Code after:
                part=ThinkingPart(
                    content='The user just says "Hi". We need to respond appropriately, friendly greeting. No special instructions. Should be short.'
                ),
                next_part_kind='text',
            ),
            PartStartEvent(index=1, part=TextPart(content='Hello! How can I help'), previous_part_kind='thinking'),
            FinalResultEvent(tool_name=None, tool_call_id=None),
            PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' you today?')),
            PartEndEvent(index=1, part=TextPart(content='Hello! How can I help you today?')),
        ]
    )


@pytest.mark.vcr()
async def test_bedrock_error(allow_model_requests: None, bedrock_provider: BedrockProvider):
    """Test that errors convert to ModelHTTPError."""
    model_id = 'us.does-not-exist-model-v1:0'
    model = BedrockConverseModel(model_id, provider=bedrock_provider)
    agent = Agent(model)

    with pytest.raises(ModelHTTPError) as exc_info:
        await agent.run('hello')

    assert exc_info.value.status_code == 400
    assert exc_info.value.model_name == model_id
    assert exc_info.value.body.get('Error', {}).get('Message') == 'The provided model identifier is invalid.'  # type: ignore[union-attr]


@pytest.mark.vcr()
async def test_bedrock_streaming_error(allow_model_requests: None, bedrock_provider: BedrockProvider):
    """Test that errors during streaming convert to ModelHTTPError."""
    model_id = 'us.does-not-exist-model-v1:0'
    model = BedrockConverseModel(model_id, provider=bedrock_provider)
    agent = Agent(model)

    with pytest.raises(ModelHTTPError) as exc_info:
        async with agent.run_stream('hello'):
            pass

    assert exc_info.value.status_code == 400
    assert exc_info.value.model_name == model_id
    assert exc_info.value.body.get('Error', {}).get('Message') == 'The provided model identifier is invalid.'  # type: ignore[union-attr]