natas_hw
/
grpc
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Revert "Allow asyncio API to be imported as grpc.aio"

This commit is contained in:
Lidi Zheng 2020-10-06 12:35:26 -07:00 committed by GitHub
parent 7c7b452f16
commit d938e0052e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 25 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/python/grpcio/grpc/__init__.py
View File

@ -2123,7 +2123,3 @@ try:
sys.modules.update({'grpc.reflection': grpc_reflection})
except ImportError:
pass
if sys.version_info >= (3, 6):
from grpc import aio # pylint: disable=ungrouped-imports
sys.modules.update({'grpc.aio': aio})

2
src/python/grpcio_tests/tests_aio/tests.json
View File

@ -28,7 +28,7 @@
"unit.connectivity_test.TestConnectivityState",
"unit.context_peer_test.TestContextPeer",
"unit.done_callback_test.TestDoneCallback",
"unit.init_test.TestInit",
"unit.init_test.TestChannel",
"unit.metadata_test.TestMetadata",
"unit.outside_init_test.TestOutsideInit",
"unit.secure_call_test.TestStreamStreamSecureCall",

36
src/python/grpcio_tests/tests_aio/unit/init_test.py
View File

@ -14,26 +14,38 @@
import logging
import unittest
import grpc
from grpc.experimental import aio
from tests_aio.unit._test_server import start_test_server
from tests_aio.unit._test_base import AioTestBase
from tests.unit import resources
class TestInit(AioTestBase):
_PRIVATE_KEY = resources.private_key()
_CERTIFICATE_CHAIN = resources.certificate_chain()
_TEST_ROOT_CERTIFICATES = resources.test_root_certificates()
async def test_grpc(self):
import grpc # pylint: disable=wrong-import-position
channel = grpc.aio.insecure_channel('dummy')
self.assertIsInstance(channel, grpc.aio.Channel)
async def test_grpc_dot_aio(self):
import grpc.aio # pylint: disable=wrong-import-position
channel = grpc.aio.insecure_channel('dummy')
self.assertIsInstance(channel, grpc.aio.Channel)
class TestChannel(AioTestBase):
async def test_aio_from_grpc(self):
from grpc import aio # pylint: disable=wrong-import-position
channel = aio.insecure_channel('dummy')
async def test_insecure_channel(self):
server_target, _ = await start_test_server() # pylint: disable=unused-variable
channel = aio.insecure_channel(server_target)
self.assertIsInstance(channel, aio.Channel)
async def test_secure_channel(self):
server_target, _ = await start_test_server(secure=True) # pylint: disable=unused-variable
credentials = grpc.ssl_channel_credentials(
root_certificates=_TEST_ROOT_CERTIFICATES,
private_key=_PRIVATE_KEY,
certificate_chain=_CERTIFICATE_CHAIN,
)
secure_channel = aio.secure_channel(server_target, credentials)
self.assertIsInstance(secure_channel, aio.Channel)
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)