Enable all AsyncIO tests for Bazel

src/python/grpcio_tests/commands.py

@@ -120,6 +120,8 @@ class TestAio(setuptools.Command):
         pass
 
     def run(self):
+        from grpc.experimental import aio
+        aio.init_grpc_aio()
         self._add_eggs_to_path()
 
         import tests

src/python/grpcio_tests/tests_aio/unit/BUILD.bazel

@@ -17,9 +17,20 @@ package(
     default_visibility = ["//visibility:public"],
 )
 
-GRPC_ASYNC_TESTS = [
-    "server_test.py",
-]
+GRPC_ASYNC_TESTS = glob(["*_test.py"])
+
+
+py_library(
+    name = "_test_server",
+    srcs_version = "PY3",
+    srcs = ["_test_server.py"],
+    deps = [
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/proto/grpc/testing:py_messages_proto",
+        "//src/proto/grpc/testing:test_py_pb2_grpc",
+        "//src/proto/grpc/testing:empty_py_pb2",
+    ]
+)
 
 [
     py_test(
@@ -29,10 +40,12 @@ GRPC_ASYNC_TESTS = [
         main=test_file_name,
         python_version="PY3",
         deps=[
+            ":_test_server",
             "//src/python/grpcio/grpc:grpcio",
             "//src/proto/grpc/testing:py_messages_proto",
             "//src/proto/grpc/testing:benchmark_service_py_pb2_grpc",
             "//src/proto/grpc/testing:benchmark_service_py_pb2",
+            "//src/python/grpcio_tests/tests/unit/framework/common",
             "//external:six"
         ],
         imports=["../../",],

src/python/grpcio_tests/tests_aio/unit/sync_server.py → src/python/grpcio_tests/tests_aio/unit/_test_server.py

@@ -18,38 +18,27 @@ from concurrent import futures
 from time import sleep
 
 import grpc
+from grpc.experimental import aio
 from src.proto.grpc.testing import messages_pb2
 from src.proto.grpc.testing import test_pb2_grpc
 from tests.unit.framework.common import test_constants
 
 
-# TODO (https://github.com/grpc/grpc/issues/19762)
-# Change for an asynchronous server version once it's implemented.
 class TestServiceServicer(test_pb2_grpc.TestServiceServicer):
 
-    def UnaryCall(self, request, context):
+    async def UnaryCall(self, request, context):
         return messages_pb2.SimpleResponse()
 
-    def EmptyCall(self, request, context):
+    async def EmptyCall(self, request, context):
         while True:
             sleep(test_constants.LONG_TIMEOUT)
 
 
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description='Synchronous gRPC server.')
-    parser.add_argument(
-        '--host_and_port',
-        required=True,
-        type=str,
-        nargs=1,
-        help='the host and port to listen.')
-    args = parser.parse_args()
-
-    server = grpc.server(
-        futures.ThreadPoolExecutor(max_workers=1),
-        options=(('grpc.so_reuseport', 1),))
+async def start_test_server():
+    server = aio.server(options=(('grpc.so_reuseport', 0),))
     test_pb2_grpc.add_TestServiceServicer_to_server(TestServiceServicer(),
                                                     server)
-    server.add_insecure_port(args.host_and_port[0])
-    server.start()
-    server.wait_for_termination()
+    port = server.add_insecure_port('[::]:0')
+    await server.start()
+    # NOTE(lidizheng) returning the server to prevent it from deallocation
+    return 'localhost:%d' % port, server

src/python/grpcio_tests/tests_aio/unit/channel_test.py

@@ -12,23 +12,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import logging
 import unittest
 
 import grpc
 
 from grpc.experimental import aio
-from tests_aio.unit import test_base
 from src.proto.grpc.testing import messages_pb2
 from tests.unit.framework.common import test_constants
+from tests_aio.unit._test_server import start_test_server
 
 
-class TestChannel(test_base.AioTestBase):
+class TestChannel(unittest.TestCase):
 
     def test_async_context(self):
 
         async def coro():
-            async with aio.insecure_channel(self.server_target) as channel:
+            server_target, unused_server = await start_test_server()
+
+            async with aio.insecure_channel(server_target) as channel:
                 hi = channel.unary_unary(
                     '/grpc.testing.TestService/UnaryCall',
                     request_serializer=messages_pb2.SimpleRequest.
@@ -37,12 +40,14 @@ class TestChannel(test_base.AioTestBase):
                 )
                 await hi(messages_pb2.SimpleRequest())
 
-        self.loop.run_until_complete(coro())
+        asyncio.get_event_loop().run_until_complete(coro())
 
     def test_unary_unary(self):
 
         async def coro():
-            channel = aio.insecure_channel(self.server_target)
+            server_target, unused_server = await start_test_server()
+
+            channel = aio.insecure_channel(server_target)
             hi = channel.unary_unary(
                 '/grpc.testing.TestService/UnaryCall',
                 request_serializer=messages_pb2.SimpleRequest.SerializeToString,
@@ -53,12 +58,14 @@ class TestChannel(test_base.AioTestBase):
 
             await channel.close()
 
-        self.loop.run_until_complete(coro())
+        asyncio.get_event_loop().run_until_complete(coro())
 
     def test_unary_call_times_out(self):
 
         async def coro():
-            async with aio.insecure_channel(self.server_target) as channel:
+            server_target, unused_server = await start_test_server()
+
+            async with aio.insecure_channel(server_target) as channel:
                 empty_call_with_sleep = channel.unary_unary(
                     "/grpc.testing.TestService/EmptyCall",
                     request_serializer=messages_pb2.SimpleRequest.
@@ -83,9 +90,10 @@ class TestChannel(test_base.AioTestBase):
                 self.assertIsNotNone(
                     exception_context.exception.trailing_metadata())
 
-        self.loop.run_until_complete(coro())
+        asyncio.get_event_loop().run_until_complete(coro())
 
 
 if __name__ == '__main__':
+    aio.init_grpc_aio()
     logging.basicConfig()
     unittest.main(verbosity=2)

src/python/grpcio_tests/tests_aio/unit/init_test.py

@@ -12,12 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import logging
 import unittest
 
 import grpc
 from grpc.experimental import aio
-from tests_aio.unit import test_base
+from tests_aio.unit._test_server import start_test_server
 
 
 class TestAioRpcError(unittest.TestCase):
@@ -59,17 +60,20 @@ class TestAioRpcError(unittest.TestCase):
                       second_aio_rpc_error.__class__)
 
 
-class TestInsecureChannel(test_base.AioTestBase):
+class TestInsecureChannel(unittest.TestCase):
 
     def test_insecure_channel(self):
 
         async def coro():
-            channel = aio.insecure_channel(self.server_target)
+            server_target, unused_server = await start_test_server()
+
+            channel = aio.insecure_channel(server_target)
             self.assertIsInstance(channel, aio.Channel)
 
-        self.loop.run_until_complete(coro())
+        asyncio.get_event_loop().run_until_complete(coro())
 
 
 if __name__ == '__main__':
+    aio.init_grpc_aio()
     logging.basicConfig()
     unittest.main(verbosity=2)

src/python/grpcio_tests/tests_aio/unit/test_base.py

@@ -1,101 +0,0 @@
-# Copyright 2019 The gRPC Authors.
-#
-# 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 os
-import sys
-import subprocess
-
-import asyncio
-import unittest
-import socket
-
-from grpc.experimental import aio
-from tests_aio.unit import sync_server
-
-
-def _get_free_loopback_tcp_port():
-    if socket.has_ipv6:
-        tcp_socket = socket.socket(socket.AF_INET6)
-        host = "::1"
-        host_target = "[::1]"
-    else:
-        tcp_socket = socket.socket(socket.AF_INET)
-        host = "127.0.0.1"
-        host_target = host
-    tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
-    tcp_socket.bind((host, 0))
-    address_tuple = tcp_socket.getsockname()
-    return tcp_socket, "%s:%s" % (host_target, address_tuple[1])
-
-
-class _Server:
-    """_Server is an wrapper for a sync-server subprocess.
-
-    The synchronous server is executed in another process which initializes
-    implicitly the grpc using the synchronous configuration. Both worlds
-    can not coexist within the same process.
-    """
-
-    def __init__(self, host_and_port):  # pylint: disable=W0621
-        self._host_and_port = host_and_port
-        self._handle = None
-
-    def start(self):
-        assert self._handle is None
-
-        try:
-            from google3.pyglib import resources
-            executable = resources.GetResourceFilename(
-                "google3/third_party/py/grpc/sync_server")
-            args = [executable, '--host_and_port', self._host_and_port]
-        except ImportError:
-            executable = sys.executable
-            directory, _ = os.path.split(os.path.abspath(__file__))
-            filename = directory + '/sync_server.py'
-            args = [
-                executable, filename, '--host_and_port', self._host_and_port
-            ]
-
-        self._handle = subprocess.Popen(args)
-
-    def terminate(self):
-        if not self._handle:
-            return
-
-        self._handle.terminate()
-        self._handle.wait()
-        self._handle = None
-
-
-class AioTestBase(unittest.TestCase):
-
-    def setUp(self):
-        self._socket, self._target = _get_free_loopback_tcp_port()
-        self._server = _Server(self._target)
-        self._server.start()
-        self._loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(self._loop)
-        aio.init_grpc_aio()
-
-    def tearDown(self):
-        self._server.terminate()
-        self._socket.close()
-
-    @property
-    def loop(self):
-        return self._loop
-
-    @property
-    def server_target(self):
-        return self._target