Skip to content

Commit 2ffdb7a

Browse files
committed
Add tests for S3 client creation in SQS Channel
Introduce two tests to verify S3 client creation behavior: one for insecure connections and another for custom endpoint usage. This ensures proper configuration of boto3 client initialization in these scenarios.
1 parent 066e192 commit 2ffdb7a

File tree

1 file changed

+45
-0
lines changed

1 file changed

+45
-0
lines changed

t/unit/transport/test_SQS.py

+45
Original file line numberDiff line numberDiff line change
@@ -367,6 +367,51 @@ def test_optional_b64_decode(self):
367367
assert self.channel._optional_b64_decode(raw) == raw
368368
assert self.channel._optional_b64_decode(b"test123") == b"test123"
369369

370+
@patch('boto3.session.Session')
371+
def test_new_s3_client_with_is_secure_false(self, mock_session):
372+
self.channel.is_secure = False
373+
self.channel.endpoint_url = None
374+
375+
self.channel.new_s3_client(
376+
region='us-west-2',
377+
access_key_id='test_access_key',
378+
secret_access_key='test_secret_key'
379+
)
380+
381+
# assert isinstance(client, boto3.client('s3').__class__)
382+
mock_session.assert_called_once_with(
383+
region_name='us-west-2',
384+
aws_access_key_id='test_access_key',
385+
aws_secret_access_key='test_secret_key',
386+
aws_session_token=None
387+
)
388+
mock_session().client.assert_called_once_with(
389+
's3', use_ssl=False
390+
)
391+
392+
@patch('boto3.session.Session')
393+
def test_new_s3_client_with_custom_endpoint(self, mock_session):
394+
mock_client = Mock()
395+
mock_session.return_value.client.return_value = mock_client
396+
397+
self.channel.is_secure = True
398+
self.channel.endpoint_url = 'https://custom-endpoint.com'
399+
400+
result = self.channel.new_s3_client('us-west-2', 'access_key', 'secret_key')
401+
402+
mock_session.assert_called_once_with(
403+
region_name='us-west-2',
404+
aws_access_key_id='access_key',
405+
aws_secret_access_key='secret_key',
406+
aws_session_token=None
407+
)
408+
mock_session.return_value.client.assert_called_once_with(
409+
's3',
410+
use_ssl=True,
411+
endpoint_url='https://custom-endpoint.com'
412+
)
413+
assert result == mock_client
414+
370415
def test_messages_to_python(self):
371416
from kombu.asynchronous.aws.sqs.message import Message
372417

0 commit comments

Comments
 (0)