# Path: aiobotocore/httpsession.py
# class AIOHTTPSession:
#     def __init__(
#             self,
#             verify: bool = True,
#             proxies: Dict[str, str] = None,  # {scheme: url}
#             timeout: float = None,
#             max_pool_connections: int = MAX_POOL_CONNECTIONS,
#             socket_options=None,
#             client_cert=None,
#             proxies_config=None,
#             connector_args=None
#     ):
#     async def __aenter__(self):
#     async def __aexit__(self, exc_type, exc_val, exc_tb):
#     async def close(self):
#     def _get_ssl_context(self):
#     def _setup_proxy_ssl_context(self, proxies_settings):
#     async def send(self, request):


import asyncio
import pytest
import aioitertools
import aiobotocore.retries.adaptive
import botocore.retries.adaptive
from collections import defaultdict
from aiobotocore import httpsession
async def test_adaptive_retry(s3_client, config, create_object, bucket_name,
                              patch_attributes):
    await create_object('foobarbaz', body='body contents')

    # Check that our async implementations were correctly called.
    # We need to patch event listeners before the S3 client is created (see
    # documentation for `patch_attributes`), but as a result, other calls may be
    # performed during the setup of other fixtures. Thus, we can't rely on the total
    # number of calls, we just inspect the last one.
    assert len(patch_attributes[0].mock_calls) > 0  # on_sending_request
    _, _, call_args = patch_attributes[0].mock_calls[-1]
    assert call_args["event_name"] == "before-send.s3.PutObject"
    assert call_args["request"].url.endswith("foobarbaz")

    assert len(patch_attributes[1].mock_calls) > 0  # on_receiving_response
    _, _, call_args = patch_attributes[1].mock_calls[-1]
    assert call_args["event_name"] == "needs-retry.s3.PutObject"

    # Check that we did not call any blocking method.
    # Unfortunately can't directly patch threading.Lock.__enter__.
    patch_attributes[2].assert_not_called()
    patch_attributes[3].assert_not_called()


@pytest.mark.moto
@pytest.mark.asyncio
async def test_get_object_stream_wrapper(s3_client, create_object,
                                         bucket_name):
    await create_object('foobarbaz', body='body contents')
    response = await s3_client.get_object(Bucket=bucket_name, Key='foobarbaz')

# Path: aiobotocore/httpsession.py
# class AIOHTTPSession:
#     def __init__(
#             self,
#             verify: bool = True,
#             proxies: Dict[str, str] = None,  # {scheme: url}
#             timeout: float = None,
#             max_pool_connections: int = MAX_POOL_CONNECTIONS,
#             socket_options=None,
#             client_cert=None,
#             proxies_config=None,
#             connector_args=None
#     ):
#     async def __aenter__(self):
#     async def __aexit__(self, exc_type, exc_val, exc_tb):
#     async def close(self):
#     def _get_ssl_context(self):
#     def _setup_proxy_ssl_context(self, proxies_settings):
#     async def send(self, request):


# Path: tests/test_basic_s3.py
import asyncio
import pytest
import aioitertools
import aiobotocore.retries.adaptive
import botocore.retries.adaptive
from collections import defaultdict
from aiobotocore import httpsession

async def test_adaptive_retry(s3_client, config, create_object, bucket_name,
                              patch_attributes):
    await create_object('foobarbaz', body='body contents')

    # Check that our async implementations were correctly called.
    # We need to patch event listeners before the S3 client is created (see
    # documentation for `patch_attributes`), but as a result, other calls may be
    # performed during the setup of other fixtures. Thus, we can't rely on the total
    # number of calls, we just inspect the last one.
    assert len(patch_attributes[0].mock_calls) > 0  # on_sending_request
    _, _, call_args = patch_attributes[0].mock_calls[-1]
    assert call_args["event_name"] == "before-send.s3.PutObject"
    assert call_args["request"].url.endswith("foobarbaz")

    assert len(patch_attributes[1].mock_calls) > 0  # on_receiving_response
    _, _, call_args = patch_attributes[1].mock_calls[-1]
    assert call_args["event_name"] == "needs-retry.s3.PutObject"

    # Check that we did not call any blocking method.
    # Unfortunately can't directly patch threading.Lock.__enter__.
    patch_attributes[2].assert_not_called()
    patch_attributes[3].assert_not_called()


@pytest.mark.moto
@pytest.mark.asyncio
async def test_get_object_stream_wrapper(s3_client, create_object,
                                         bucket_name):
    await create_object('foobarbaz', body='body contents')
    response = await s3_client.get_object(Bucket=bucket_name, Key='foobarbaz')

    body = response['Body']