test_providers.py 文件源码

def test_fetch_messages_with_botocoreerror(mock_boto_session_sqs, boto_client_sqs):
    with mock_boto_session_sqs:
        error = BotoCoreError()
        boto_client_sqs.receive_message.side_effect = error

        provider = SQSProvider('queue-name')
        with pytest.raises(ProviderError):
            await provider.fetch_messages()