test: fix some tests

Author: standy66

.travis.yml

@@ -10,7 +10,7 @@ stages:
     if: tag IS present AND type != pull_request
 env:
   # We run pypy tests first because they run the slowest
-  - PYTHON_IMAGE="pypy:3 PURERPC_BACKEND="asyncio"
+  - PYTHON_IMAGE="pypy:3" PURERPC_BACKEND="asyncio"
   - PYTHON_IMAGE="pypy:3" PURERPC_BACKEND="curio"
   - PYTHON_IMAGE="pypy:3" PURERPC_BACKEND="trio"
   - PYTHON_IMAGE="standy/pypy:3.6-6.1.0" PURERPC_BACKEND="asyncio"

README.md

@@ -1,6 +1,7 @@
 # purerpc
 
 [![Build Status](https://travis-ci.org/standy66/purerpc.png?branch=master)](https://travis-ci.org/standy66/purerpc)
+[![Downloads](https://pepy.tech/badge/purerpc/month)](https://pepy.tech/project/purerpc)
 
 Asynchronous pure Python gRPC client and server implementation supporting
 [asyncio](https://docs.python.org/3/library/asyncio.html),
@@ -47,7 +48,7 @@ python -m grpc_tools.protoc --purerpc_out=. --python_out=. -I. greeter.proto
 
 NOTE: `greeter_grpc` module is generated by purerpc's `protoc-gen-purerpc` plugin.
 
-Below is the examples for Python 3.6 and above which introduced asynchronous generators as a language concept.
+Below are the examples for Python 3.6 and above which introduced asynchronous generators as a language concept.
 For Python 3.5, where native asynchronous generators are not supported, you can use [async_generator](https://github.com/python-trio/async_generator) library for this purpose.
 Just mark yielding coroutines with `@async_generator` decorator and use `await yield_(value)` and `await yield_from_(async_iterable)` instead of `yield`.
 
@@ -98,7 +99,7 @@ async def main():
 
 
 if __name__ == "__main__":
-    curio.run(main)  # Or trio.run(main)
+    curio.run(main)  # Or trio.run(main), or run in asyncio event loop
 ```
 
 You can mix server and client code, for example make a server that requests something using purerpc from another gRPC server, etc.

setup.py

@@ -2,6 +2,8 @@
 import os
 from setuptools import setup, find_packages
 
+exec(open("src/purerpc/_version.py", encoding="utf-8").read())
+
 
 def read(*names, **kwargs):
     with open(os.path.join(os.path.dirname(__file__), *names), "r") as fin:
@@ -12,7 +14,7 @@ def main():
     console_scripts = ['protoc-gen-purerpc=purerpc.protoc_plugin.plugin:main']
     setup(
         name="purerpc",
-        version="0.1.6",
+        version=__version__,
         license="Apache License Version 2.0",
         description="Asynchronous pure python gRPC server and client implementation using curio "
                     "and hyper-h2.",

src/purerpc/_version.py

@@ -0,0 +1,3 @@
+# This file is imported from __init__.py and exec'd from setup.py
+
+__version__ = "0.2.0"

src/purerpc/client.py

@@ -16,14 +16,12 @@ def __init__(self, host, port):
         self._host = host
         self._port = port
         self._grpc_socket = None
-        self._task_group = None
 
     async def __aenter__(self):
         await super().__aenter__()  # Does nothing
 
         background_task_group = await self.enter_async_context(anyio.create_task_group())
         self.push_async_callback(background_task_group.cancel_scope.cancel)
-        self._task_group = await self.enter_async_context(anyio.create_task_group())
 
         socket = await anyio.connect_tcp(self._host, self._port, autostart_tls=False, tls_standard_compatible=False)
         config = GRPCConfiguration(client_side=True)
@@ -57,10 +55,10 @@ def get_method_stub(self, method_name: str, signature: RPCSignature):
         stream_fn = functools.partial(self.rpc, method_name, signature.request_type,
                                       signature.response_type)
         if signature.cardinality == Cardinality.STREAM_STREAM:
-            return ClientStubStreamStream(stream_fn, self.channel._task_group)
+            return ClientStubStreamStream(stream_fn)
         elif signature.cardinality == Cardinality.UNARY_STREAM:
-            return ClientStubUnaryStream(stream_fn, self.channel._task_group)
+            return ClientStubUnaryStream(stream_fn)
         elif signature.cardinality == Cardinality.STREAM_UNARY:
-            return ClientStubStreamUnary(stream_fn, self.channel._task_group)
+            return ClientStubStreamUnary(stream_fn)
         else:
-            return ClientStubUnaryUnary(stream_fn, self.channel._task_group)
+            return ClientStubUnaryUnary(stream_fn)

src/purerpc/wrappers.py

@@ -1,4 +1,8 @@
-from async_generator import aclosing, async_generator, yield_, yield_from_
+import functools
+
+import anyio
+from async_generator import aclosing, async_generator, asynccontextmanager, yield_, yield_from_
+
 from .grpclib.exceptions import ProtocolError, raise_status
 from .grpclib.status import Status, StatusCode
 from purerpc.grpc_proto import GRPCProtoStream
@@ -29,9 +33,9 @@ async def stream_to_async_iterator(stream: GRPCProtoStream):
         await yield_(msg)
 
 
-async def send_multiple_messages_server(stream, agen):
-    async with aclosing(agen) as tmp:
-        async for message in tmp:
+async def send_multiple_messages_server(stream, aiter):
+    async with aclosing(aiter) as aiter:
+        async for message in aiter:
             await stream.send_message(message)
     await stream.close(Status(StatusCode.OK))
 
@@ -41,18 +45,20 @@ async def send_single_message_server(stream, message):
     await stream.close(Status(StatusCode.OK))
 
 
-async def send_multiple_messages_client(stream, agen):
+async def send_multiple_messages_client(stream, aiter):
     try:
-        async with aclosing(agen) as tmp:
-            async for message in tmp:
+        async with aclosing(aiter) as aiter:
+            async for message in aiter:
                 await stream.send_message(message)
     finally:
         await stream.close()
 
 
 async def send_single_message_client(stream, message):
-    await stream.send_message(message)
-    await stream.close()
+    try:
+        await stream.send_message(message)
+    finally:
+        await stream.close()
 
 
 async def call_server_unary_unary(func, stream):
@@ -76,9 +82,8 @@ async def call_server_stream_stream(func, stream):
 
 
 class ClientStub:
-    def __init__(self, stream_fn, parent_task_group):
+    def __init__(self, stream_fn):
         self._stream_fn = stream_fn
-        self._parent_task_group = parent_task_group
 
 
 class ClientStubUnaryUnary(ClientStub):
@@ -99,16 +104,17 @@ async def __call__(self, message, *, metadata=None):
 class ClientStubStreamUnary(ClientStub):
     async def __call__(self, message_aiter, *, metadata=None):
         stream = await self._stream_fn(metadata=metadata)
-        await self._parent_task_group.spawn(send_multiple_messages_client, stream, message_aiter)
-        return await extract_message_from_singleton_stream(stream)
+        async with anyio.create_task_group() as task_group:
+            await task_group.spawn(send_multiple_messages_client, stream, message_aiter)
+            return await extract_message_from_singleton_stream(stream)
 
 
 class ClientStubStreamStream(ClientStub):
     @async_generator
     async def call_aiter(self, message_aiter, metadata):
         stream = await self._stream_fn(metadata=metadata)
-        if message_aiter is not None:
-            await self._parent_task_group.spawn(send_multiple_messages_client, stream, message_aiter)
+        async with anyio.create_task_group() as task_group:
+            await task_group.spawn(send_multiple_messages_client, stream, message_aiter)
             await yield_from_(stream_to_async_iterator(stream))
 
     async def call_stream(self, metadata):

tests/test_client_server_errors.py

@@ -4,7 +4,7 @@
 import typing
 import time
 
-from async_generator import async_generator, yield_
+from async_generator import async_generator, aclosing, yield_
 
 from .greeter_pb2 import HelloReply, HelloRequest
 from .greeter_pb2_grpc import GreeterStub, GreeterServicer, add_GreeterServicer_to_server
@@ -60,7 +60,7 @@ def SayHelloToMany(self, messages, context):
 
             @async_generator
             async def generator():
-                for _ in range(10):
+                for _ in range(7):
                     await yield_(HelloRequest())
 
             GreeterStub = grpc_module.GreeterStub
@@ -69,11 +69,11 @@ async def worker(channel):
                 with self.assertRaisesRegex(purerpc.RpcFailedError, r"oops my bad"):
                     await stub.SayHello(HelloRequest(name="World"))
 
-                aiter = stub.SayHelloToMany(generator())
-                for _ in range(7):
-                    await aiter.__anext__()
-                with self.assertRaisesRegex(purerpc.RpcFailedError, r"Lucky 7"):
-                    await aiter.__anext__()
+                async with aclosing(stub.SayHelloToMany(generator())) as aiter:
+                    for _ in range(7):
+                        await aiter.__anext__()
+                    with self.assertRaisesRegex(purerpc.RpcFailedError, r"Lucky 7"):
+                        await aiter.__anext__()
 
             async def main():
                 async with purerpc.insecure_channel("localhost", port) as channel: