feat: third batch of AsyncIO integration (#29)

* LRO client
* gRPC wrappers & helpers
* With unit tests & docs

4 files changed, 708 insertions, 0 deletions

diff --git a/tests/asyncio/gapic/test_method_async.py b/tests/asyncio/gapic/test_method_async.py
new file mode 100644
index 0000000..7318362
--- /dev/null
+++ b/tests/asyncio/gapic/test_method_async.py
@@ -0,0 +1,243 @@
+# Copyright 2017 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+
+from grpc.experimental import aio
+import mock
+import pytest
+
+from google.api_core import (exceptions, gapic_v1, grpc_helpers_async,
+                             retry_async, timeout)
+
+
+def _utcnow_monotonic():
+    current_time = datetime.datetime.min
+    delta = datetime.timedelta(seconds=0.5)
+    while True:
+        yield current_time
+        current_time += delta
+
+
+@pytest.mark.asyncio
+async def test_wrap_method_basic():
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+
+    wrapped_method = gapic_v1.method_async.wrap_method(method)
+
+    result = await wrapped_method(1, 2, meep="moop")
+
+    assert result == 42
+    method.assert_called_once_with(1, 2, meep="moop", metadata=mock.ANY)
+
+    # Check that the default client info was specified in the metadata.
+    metadata = method.call_args[1]["metadata"]
+    assert len(metadata) == 1
+    client_info = gapic_v1.client_info.DEFAULT_CLIENT_INFO
+    user_agent_metadata = client_info.to_grpc_metadata()
+    assert user_agent_metadata in metadata
+
+
+@pytest.mark.asyncio
+async def test_wrap_method_with_no_client_info():
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall()
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, client_info=None
+    )
+
+    await wrapped_method(1, 2, meep="moop")
+
+    method.assert_called_once_with(1, 2, meep="moop")
+
+
+@pytest.mark.asyncio
+async def test_wrap_method_with_custom_client_info():
+    client_info = gapic_v1.client_info.ClientInfo(
+        python_version=1,
+        grpc_version=2,
+        api_core_version=3,
+        gapic_version=4,
+        client_library_version=5,
+    )
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall()
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, client_info=client_info
+    )
+
+    await wrapped_method(1, 2, meep="moop")
+
+    method.assert_called_once_with(1, 2, meep="moop", metadata=mock.ANY)
+
+    # Check that the custom client info was specified in the metadata.
+    metadata = method.call_args[1]["metadata"]
+    assert client_info.to_grpc_metadata() in metadata
+
+
+@pytest.mark.asyncio
+async def test_invoke_wrapped_method_with_metadata():
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall()
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+
+    wrapped_method = gapic_v1.method_async.wrap_method(method)
+
+    await wrapped_method(mock.sentinel.request, metadata=[("a", "b")])
+
+    method.assert_called_once_with(mock.sentinel.request, metadata=mock.ANY)
+    metadata = method.call_args[1]["metadata"]
+    # Metadata should have two items: the client info metadata and our custom
+    # metadata.
+    assert len(metadata) == 2
+    assert ("a", "b") in metadata
+
+
+@pytest.mark.asyncio
+async def test_invoke_wrapped_method_with_metadata_as_none():
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall()
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+
+    wrapped_method = gapic_v1.method_async.wrap_method(method)
+
+    await wrapped_method(mock.sentinel.request, metadata=None)
+
+    method.assert_called_once_with(mock.sentinel.request, metadata=mock.ANY)
+    metadata = method.call_args[1]["metadata"]
+    # Metadata should have just one items: the client info metadata.
+    assert len(metadata) == 1
+
+
+@mock.patch("asyncio.sleep")
+@pytest.mark.asyncio
+async def test_wrap_method_with_default_retry_and_timeout(unused_sleep):
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, side_effect=[
+        exceptions.InternalServerError(None),
+        fake_call,
+    ])
+
+    default_retry = retry_async.AsyncRetry()
+    default_timeout = timeout.ConstantTimeout(60)
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, default_retry, default_timeout
+    )
+
+    result = await wrapped_method()
+
+    assert result == 42
+    assert method.call_count == 2
+    method.assert_called_with(timeout=60, metadata=mock.ANY)
+
+
+@mock.patch("asyncio.sleep")
+@pytest.mark.asyncio
+async def test_wrap_method_with_default_retry_and_timeout_using_sentinel(unused_sleep):
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, side_effect=[
+        exceptions.InternalServerError(None),
+        fake_call,
+    ])
+
+    default_retry = retry_async.AsyncRetry()
+    default_timeout = timeout.ConstantTimeout(60)
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, default_retry, default_timeout
+    )
+
+    result = await wrapped_method(
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout=gapic_v1.method_async.DEFAULT,
+    )
+
+    assert result == 42
+    assert method.call_count == 2
+    method.assert_called_with(timeout=60, metadata=mock.ANY)
+
+
+@mock.patch("asyncio.sleep")
+@pytest.mark.asyncio
+async def test_wrap_method_with_overriding_retry_and_timeout(unused_sleep):
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, side_effect=[
+        exceptions.NotFound(None),
+        fake_call,
+    ])
+
+    default_retry = retry_async.AsyncRetry()
+    default_timeout = timeout.ConstantTimeout(60)
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, default_retry, default_timeout
+    )
+
+    result = await wrapped_method(
+        retry=retry_async.AsyncRetry(retry_async.if_exception_type(exceptions.NotFound)),
+        timeout=timeout.ConstantTimeout(22),
+    )
+
+    assert result == 42
+    assert method.call_count == 2
+    method.assert_called_with(timeout=22, metadata=mock.ANY)
+
+
+@mock.patch("asyncio.sleep")
+@mock.patch(
+    "google.api_core.datetime_helpers.utcnow",
+    side_effect=_utcnow_monotonic(),
+    autospec=True,
+)
+@pytest.mark.asyncio
+async def test_wrap_method_with_overriding_retry_deadline(utcnow, unused_sleep):
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(
+        spec=aio.UnaryUnaryMultiCallable,
+        side_effect=([exceptions.InternalServerError(None)] * 4) + [fake_call])
+
+    default_retry = retry_async.AsyncRetry()
+    default_timeout = timeout.ExponentialTimeout(deadline=60)
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, default_retry, default_timeout
+    )
+
+    # Overriding only the retry's deadline should also override the timeout's
+    # deadline.
+    result = await wrapped_method(retry=default_retry.with_deadline(30))
+
+    assert result == 42
+    timeout_args = [call[1]["timeout"] for call in method.call_args_list]
+    assert timeout_args == [5.0, 10.0, 20.0, 26.0, 25.0]
+    assert utcnow.call_count == (
+        1
+        + 1  # Compute wait_for timeout in retry_async
+        + 5  # First to set the deadline.
+        + 5  # One for each min(timeout, maximum, (DEADLINE - NOW).seconds)
+    )
+
+
+@pytest.mark.asyncio
+async def test_wrap_method_with_overriding_timeout_as_a_number():
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(42)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+    default_retry = retry_async.AsyncRetry()
+    default_timeout = timeout.ConstantTimeout(60)
+    wrapped_method = gapic_v1.method_async.wrap_method(
+        method, default_retry, default_timeout
+    )
+
+    result = await wrapped_method(timeout=22)
+
+    assert result == 42
+    method.assert_called_once_with(timeout=22, metadata=mock.ANY)
diff --git a/tests/asyncio/operations_v1/__init__.py b/tests/asyncio/operations_v1/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/asyncio/operations_v1/__init__.py
diff --git a/tests/asyncio/operations_v1/test_operations_async_client.py b/tests/asyncio/operations_v1/test_operations_async_client.py
new file mode 100644
index 0000000..0f9363f
--- /dev/null
+++ b/tests/asyncio/operations_v1/test_operations_async_client.py
@@ -0,0 +1,93 @@
+# Copyright 2017 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from grpc.experimental import aio
+import mock
+import pytest
+
+from google.api_core import (grpc_helpers_async, operations_v1,
+                             page_iterator_async)
+from google.longrunning import operations_pb2
+from google.protobuf import empty_pb2
+
+
+def _mock_grpc_objects(response):
+    fake_call = grpc_helpers_async.FakeUnaryUnaryCall(response)
+    method = mock.Mock(spec=aio.UnaryUnaryMultiCallable, return_value=fake_call)
+    mocked_channel = mock.Mock()
+    mocked_channel.unary_unary = mock.Mock(return_value=method)
+    return mocked_channel, method, fake_call
+
+
+@pytest.mark.asyncio
+async def test_get_operation():
+    mocked_channel, method, fake_call = _mock_grpc_objects(
+        operations_pb2.Operation(name="meep"))
+    client = operations_v1.OperationsAsyncClient(mocked_channel)
+
+    response = await client.get_operation("name")
+    assert method.call_count == 1
+    assert tuple(method.call_args_list[0])[0][0].name == "name"
+    assert response == fake_call.response
+
+
+@pytest.mark.asyncio
+async def test_list_operations():
+    operations = [
+        operations_pb2.Operation(name="1"),
+        operations_pb2.Operation(name="2"),
+    ]
+    list_response = operations_pb2.ListOperationsResponse(operations=operations)
+
+    mocked_channel, method, fake_call = _mock_grpc_objects(list_response)
+    client = operations_v1.OperationsAsyncClient(mocked_channel)
+
+    pager = await client.list_operations("name", "filter")
+
+    assert isinstance(pager, page_iterator_async.AsyncIterator)
+    responses = []
+    async for response in pager:
+        responses.append(response)
+
+    assert responses == operations
+
+    assert method.call_count == 1
+    request = tuple(method.call_args_list[0])[0][0]
+    assert isinstance(request, operations_pb2.ListOperationsRequest)
+    assert request.name == "name"
+    assert request.filter == "filter"
+
+
+@pytest.mark.asyncio
+async def test_delete_operation():
+    mocked_channel, method, fake_call = _mock_grpc_objects(
+        empty_pb2.Empty())
+    client = operations_v1.OperationsAsyncClient(mocked_channel)
+
+    await client.delete_operation("name")
+
+    assert method.call_count == 1
+    assert tuple(method.call_args_list[0])[0][0].name == "name"
+
+
+@pytest.mark.asyncio
+async def test_cancel_operation():
+    mocked_channel, method, fake_call = _mock_grpc_objects(
+        empty_pb2.Empty())
+    client = operations_v1.OperationsAsyncClient(mocked_channel)
+
+    await client.cancel_operation("name")
+
+    assert method.call_count == 1
+    assert tuple(method.call_args_list[0])[0][0].name == "name"
diff --git a/tests/asyncio/test_grpc_helpers_async.py b/tests/asyncio/test_grpc_helpers_async.py
new file mode 100644
index 0000000..0053952
--- /dev/null
+++ b/tests/asyncio/test_grpc_helpers_async.py
@@ -0,0 +1,372 @@
+# Copyright 2017 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import grpc
+from grpc.experimental import aio
+import mock
+import pytest
+
+from google.api_core import exceptions
+from google.api_core import grpc_helpers_async
+import google.auth.credentials
+
+
+class RpcErrorImpl(grpc.RpcError, grpc.Call):
+    def __init__(self, code):
+        super(RpcErrorImpl, self).__init__()
+        self._code = code
+
+    def code(self):
+        return self._code
+
+    def details(self):
+        return None
+
+
+@pytest.mark.asyncio
+async def test_wrap_unary_errors():
+    grpc_error = RpcErrorImpl(grpc.StatusCode.INVALID_ARGUMENT)
+    callable_ = mock.AsyncMock(spec=["__call__"], side_effect=grpc_error)
+
+    wrapped_callable = grpc_helpers_async._wrap_unary_errors(callable_)
+
+    with pytest.raises(exceptions.InvalidArgument) as exc_info:
+        await wrapped_callable(1, 2, three="four")
+
+    callable_.assert_called_once_with(1, 2, three="four")
+    assert exc_info.value.response == grpc_error
+
+
+@pytest.mark.asyncio
+async def test_common_methods_in_wrapped_call():
+    mock_call = mock.Mock(aio.UnaryUnaryCall, autospec=True)
+    wrapped_call = grpc_helpers_async._WrappedUnaryUnaryCall().with_call(mock_call)
+
+    await wrapped_call.initial_metadata()
+    assert mock_call.initial_metadata.call_count == 1
+
+    await wrapped_call.trailing_metadata()
+    assert mock_call.trailing_metadata.call_count == 1
+
+    await wrapped_call.code()
+    assert mock_call.code.call_count == 1
+
+    await wrapped_call.details()
+    assert mock_call.details.call_count == 1
+
+    wrapped_call.cancelled()
+    assert mock_call.cancelled.call_count == 1
+
+    wrapped_call.done()
+    assert mock_call.done.call_count == 1
+
+    wrapped_call.time_remaining()
+    assert mock_call.time_remaining.call_count == 1
+
+    wrapped_call.cancel()
+    assert mock_call.cancel.call_count == 1
+
+    callback = mock.sentinel.callback
+    wrapped_call.add_done_callback(callback)
+    mock_call.add_done_callback.assert_called_once_with(callback)
+
+    await wrapped_call.wait_for_connection()
+    assert mock_call.wait_for_connection.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_unary_stream():
+    mock_call = mock.Mock(aio.UnaryStreamCall, autospec=True)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    await wrapped_callable(1, 2, three="four")
+    multicallable.assert_called_once_with(1, 2, three="four")
+    assert mock_call.wait_for_connection.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_stream_unary():
+    mock_call = mock.Mock(aio.StreamUnaryCall, autospec=True)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    await wrapped_callable(1, 2, three="four")
+    multicallable.assert_called_once_with(1, 2, three="four")
+    assert mock_call.wait_for_connection.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_stream_stream():
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    await wrapped_callable(1, 2, three="four")
+    multicallable.assert_called_once_with(1, 2, three="four")
+    assert mock_call.wait_for_connection.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_type_error():
+    mock_call = mock.Mock()
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    with pytest.raises(TypeError):
+        await wrapped_callable()
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_raised():
+    grpc_error = RpcErrorImpl(grpc.StatusCode.INVALID_ARGUMENT)
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    mock_call.wait_for_connection = mock.AsyncMock(side_effect=[grpc_error])
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    with pytest.raises(exceptions.InvalidArgument):
+        await wrapped_callable()
+    assert mock_call.wait_for_connection.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_read():
+    grpc_error = RpcErrorImpl(grpc.StatusCode.INVALID_ARGUMENT)
+
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    mock_call.read = mock.AsyncMock(side_effect=grpc_error)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    wrapped_call = await wrapped_callable(1, 2, three="four")
+    multicallable.assert_called_once_with(1, 2, three="four")
+    assert mock_call.wait_for_connection.call_count == 1
+
+    with pytest.raises(exceptions.InvalidArgument) as exc_info:
+        await wrapped_call.read()
+    assert exc_info.value.response == grpc_error
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_aiter():
+    grpc_error = RpcErrorImpl(grpc.StatusCode.INVALID_ARGUMENT)
+
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    mocked_aiter = mock.Mock(spec=['__anext__'])
+    mocked_aiter.__anext__ = mock.AsyncMock(side_effect=[mock.sentinel.response, grpc_error])
+    mock_call.__aiter__ = mock.Mock(return_value=mocked_aiter)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+    wrapped_call = await wrapped_callable()
+
+    with pytest.raises(exceptions.InvalidArgument) as exc_info:
+        async for response in wrapped_call:
+            assert response == mock.sentinel.response
+    assert exc_info.value.response == grpc_error
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_aiter_non_rpc_error():
+    non_grpc_error = TypeError('Not a gRPC error')
+
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    mocked_aiter = mock.Mock(spec=['__anext__'])
+    mocked_aiter.__anext__ = mock.AsyncMock(side_effect=[mock.sentinel.response, non_grpc_error])
+    mock_call.__aiter__ = mock.Mock(return_value=mocked_aiter)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+    wrapped_call = await wrapped_callable()
+
+    with pytest.raises(TypeError) as exc_info:
+        async for response in wrapped_call:
+            assert response == mock.sentinel.response
+    assert exc_info.value == non_grpc_error
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_aiter_called_multiple_times():
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+    wrapped_call = await wrapped_callable()
+
+    assert wrapped_call.__aiter__() == wrapped_call.__aiter__()
+
+
+@pytest.mark.asyncio
+async def test_wrap_stream_errors_write():
+    grpc_error = RpcErrorImpl(grpc.StatusCode.INVALID_ARGUMENT)
+
+    mock_call = mock.Mock(aio.StreamStreamCall, autospec=True)
+    mock_call.write = mock.AsyncMock(side_effect=[None, grpc_error])
+    mock_call.done_writing = mock.AsyncMock(side_effect=[None, grpc_error])
+    multicallable = mock.Mock(return_value=mock_call)
+
+    wrapped_callable = grpc_helpers_async._wrap_stream_errors(multicallable)
+
+    wrapped_call = await wrapped_callable()
+
+    await wrapped_call.write(mock.sentinel.request)
+    with pytest.raises(exceptions.InvalidArgument) as exc_info:
+        await wrapped_call.write(mock.sentinel.request)
+    assert mock_call.write.call_count == 2
+    assert exc_info.value.response == grpc_error
+
+    await wrapped_call.done_writing()
+    with pytest.raises(exceptions.InvalidArgument) as exc_info:
+        await wrapped_call.done_writing()
+    assert mock_call.done_writing.call_count == 2
+    assert exc_info.value.response == grpc_error
+
+
+@mock.patch("google.api_core.grpc_helpers_async._wrap_unary_errors")
+def test_wrap_errors_non_streaming(wrap_unary_errors):
+    callable_ = mock.create_autospec(aio.UnaryUnaryMultiCallable)
+
+    result = grpc_helpers_async.wrap_errors(callable_)
+
+    assert result == wrap_unary_errors.return_value
+    wrap_unary_errors.assert_called_once_with(callable_)
+
+
+@mock.patch("google.api_core.grpc_helpers_async._wrap_stream_errors")
+def test_wrap_errors_streaming(wrap_stream_errors):
+    callable_ = mock.create_autospec(aio.UnaryStreamMultiCallable)
+
+    result = grpc_helpers_async.wrap_errors(callable_)
+
+    assert result == wrap_stream_errors.return_value
+    wrap_stream_errors.assert_called_once_with(callable_)
+
+
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch(
+    "google.auth.default",
+    return_value=(mock.sentinel.credentials, mock.sentinel.projet),
+)
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_implicit(grpc_secure_channel, default, composite_creds_call):
+    target = "example.com:443"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers_async.create_channel(target)
+
+    assert channel is grpc_secure_channel.return_value
+    default.assert_called_once_with(scopes=None)
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch(
+    "google.auth.default",
+    return_value=(mock.sentinel.credentials, mock.sentinel.projet),
+)
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_implicit_with_ssl_creds(
+    grpc_secure_channel, default, composite_creds_call
+):
+    target = "example.com:443"
+
+    ssl_creds = grpc.ssl_channel_credentials()
+
+    grpc_helpers_async.create_channel(target, ssl_credentials=ssl_creds)
+
+    default.assert_called_once_with(scopes=None)
+    composite_creds_call.assert_called_once_with(ssl_creds, mock.ANY)
+    composite_creds = composite_creds_call.return_value
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch(
+    "google.auth.default",
+    return_value=(mock.sentinel.credentials, mock.sentinel.projet),
+)
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_implicit_with_scopes(
+    grpc_secure_channel, default, composite_creds_call
+):
+    target = "example.com:443"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers_async.create_channel(target, scopes=["one", "two"])
+
+    assert channel is grpc_secure_channel.return_value
+    default.assert_called_once_with(scopes=["one", "two"])
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch("google.auth.credentials.with_scopes_if_required")
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_explicit(grpc_secure_channel, auth_creds, composite_creds_call):
+    target = "example.com:443"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers_async.create_channel(target, credentials=mock.sentinel.credentials)
+
+    auth_creds.assert_called_once_with(mock.sentinel.credentials, None)
+    assert channel is grpc_secure_channel.return_value
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_explicit_scoped(grpc_secure_channel, composite_creds_call):
+    target = "example.com:443"
+    scopes = ["1", "2"]
+    composite_creds = composite_creds_call.return_value
+
+    credentials = mock.create_autospec(google.auth.credentials.Scoped, instance=True)
+    credentials.requires_scopes = True
+
+    channel = grpc_helpers_async.create_channel(
+        target, credentials=credentials, scopes=scopes
+    )
+
+    credentials.with_scopes.assert_called_once_with(scopes)
+    assert channel is grpc_secure_channel.return_value
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+@pytest.mark.skipif(grpc_helpers_async.HAS_GRPC_GCP, reason="grpc_gcp module not available")
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_without_grpc_gcp(grpc_secure_channel):
+    target = "example.com:443"
+    scopes = ["test_scope"]
+
+    credentials = mock.create_autospec(google.auth.credentials.Scoped, instance=True)
+    credentials.requires_scopes = True
+
+    grpc_helpers_async.create_channel(target, credentials=credentials, scopes=scopes)
+    grpc_secure_channel.assert_called()
+    credentials.with_scopes.assert_called_once_with(scopes)
+
+
+@pytest.mark.asyncio
+async def test_fake_stream_unary_call():
+    fake_call = grpc_helpers_async.FakeStreamUnaryCall()
+    await fake_call.wait_for_connection()
+    response = await fake_call
+    assert fake_call.response == response