Resolve the conflict between PRs

src/python/grpcio/grpc/_cython/_cygrpc/aio/call.pyx.pxi

@@ -36,7 +36,6 @@ cdef class _AioCall(GrpcCallWrapper):
         self._loop = asyncio.get_event_loop()
         self._create_grpc_call(deadline, method, call_credentials)
         self._is_locally_cancelled = False
-        self._status_received = asyncio.Event(loop=self._loop)
 
     def __dealloc__(self):
         if self.call:
@@ -118,6 +117,7 @@ cdef class _AioCall(GrpcCallWrapper):
 
     async def unary_unary(self,
                           bytes request,
+                          tuple outbound_initial_metadata,
                           object initial_metadata_observer,
                           object status_observer):
         """Performs a unary unary RPC.
@@ -134,7 +134,7 @@ cdef class _AioCall(GrpcCallWrapper):
         cdef tuple ops
 
         cdef SendInitialMetadataOperation initial_metadata_op = SendInitialMetadataOperation(
-            self._initial_metadata,
+            outbound_initial_metadata,
             GRPC_INITIAL_METADATA_USED_MASK)
         cdef SendMessageOperation send_message_op = SendMessageOperation(request, _EMPTY_FLAGS)
         cdef SendCloseFromClientOperation send_close_op = SendCloseFromClientOperation(_EMPTY_FLAGS)
@@ -152,6 +152,9 @@ cdef class _AioCall(GrpcCallWrapper):
                             ops,
                             self._loop)
 
+        # Reports received initial metadata.
+        initial_metadata_observer(receive_initial_metadata_op.initial_metadata())
+
         status = AioRpcStatus(
             receive_status_on_client_op.code(),
             receive_status_on_client_op.details(),

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

@@ -40,7 +40,7 @@ cdef class RPCState:
         self.abort_exception = None
         self.metadata_sent = False
         self.status_sent = False
-        self.trailing_metadata = tuple()
+        self.trailing_metadata = _EMPTY_METADATA
 
     cdef bytes method(self):
         return _slice_bytes(self.details.method)
@@ -466,7 +466,7 @@ async def _handle_rpc(list generic_handlers, RPCState rpc_state, object loop):
         await _send_error_status_from_server(
             rpc_state,
             StatusCode.unimplemented,
-            b'Method not found!',
+            'Method not found!',
             _EMPTY_METADATA,
             rpc_state.metadata_sent,
             loop

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

@@ -33,17 +33,40 @@ _TEST_GENERIC_HANDLER = '/test/TestGenericHandler'
 _REQUEST = b'\x00\x00\x00'
 _RESPONSE = b'\x01\x01\x01'
 
-_INITIAL_METADATA_FROM_CLIENT_TO_SERVER = (('client-to-server', 'question'),)
-_INITIAL_METADATA_FROM_SERVER_TO_CLIENT = (('server-to-client', 'answer'),)
-_TRAILING_METADATA = (('a-trailing-metadata', 'stack-trace'),)
+_INITIAL_METADATA_FROM_CLIENT_TO_SERVER = (
+    ('client-to-server', 'question'),
+    ('client-to-server-bin', b'\x07\x07\x07'),
+)
+_INITIAL_METADATA_FROM_SERVER_TO_CLIENT = (
+    ('server-to-client', 'answer'),
+    ('server-to-client-bin', b'\x06\x06\x06'),
+)
+_TRAILING_METADATA = (('a-trailing-metadata', 'stack-trace'),
+                      ('a-trailing-metadata-bin', b'\x05\x05\x05'))
 _INITIAL_METADATA_FOR_GENERIC_HANDLER = (('a-must-have-key', 'secret'),)
 
+_INVALID_METADATA_TEST_CASES = (
+    (
+        TypeError,
+        ((42, 42),),
+    ),
+    (
+        TypeError,
+        (({}, {}),),
+    ),
+    (
+        TypeError,
+        (('normal', object()),),
+    ),
+)
+
 
 def _seen_metadata(expected, actual):
-    for key, value in actual:
-        if key == expected[0] and value == expected[1]:
-            return True
-    return False
+    metadata_dict = dict(actual)
+    for metadatum in expected:
+        if metadata_dict.get(metadatum[0]) != metadatum[1]:
+            return False
+    return True
 
 
 class _TestGenericHandlerForMethods(grpc.GenericRpcHandler):
@@ -83,19 +106,20 @@ class _TestGenericHandlerForMethods(grpc.GenericRpcHandler):
 
 class _TestGenericHandlerItself(grpc.GenericRpcHandler):
 
-    async def _method(self, request, unused_context):
+    @staticmethod
+    async def _method(request, unused_context):
         assert _REQUEST == request
         return _RESPONSE
 
     def service(self, handler_details):
         assert _seen_metadata(_INITIAL_METADATA_FOR_GENERIC_HANDLER,
-                              handler_details.invocation_metadata())
-        return
+                              handler_details.invocation_metadata)
+        return grpc.unary_unary_rpc_method_handler(self._method)
 
 
 async def _start_test_server():
     server = aio.server()
-    port = server.add_secure_port('[::]:0', grpc.local_server_credentials())
+    port = server.add_insecure_port('[::]:0')
     server.add_generic_rpc_handlers((
         _TestGenericHandlerForMethods(),
         _TestGenericHandlerItself(),
@@ -108,8 +132,7 @@ class TestMetadata(AioTestBase):
 
     async def setUp(self):
         address, self._server = await _start_test_server()
-        self._client = aio.secure_channel(address,
-                                          grpc.local_channel_credentials())
+        self._client = aio.insecure_channel(address)
 
     async def tearDown(self):
         await self._client.close()
@@ -126,22 +149,23 @@ class TestMetadata(AioTestBase):
         multicallable = self._client.unary_unary(_TEST_SERVER_TO_CLIENT)
         call = multicallable(_REQUEST)
         self.assertEqual(_INITIAL_METADATA_FROM_SERVER_TO_CLIENT, await
-                         call.initial_metadata)
+                         call.initial_metadata())
         self.assertEqual(_RESPONSE, await call)
         self.assertEqual(grpc.StatusCode.OK, await call.code())
 
     async def test_trailing_metadata(self):
-        multicallable = self._client.unary_unary(_TEST_SERVER_TO_CLIENT)
+        multicallable = self._client.unary_unary(_TEST_TRAILING_METADATA)
         call = multicallable(_REQUEST)
-        self.assertEqual(_TEST_TRAILING_METADATA, await call.trailing_metadata)
+        self.assertEqual(_TRAILING_METADATA, await call.trailing_metadata())
         self.assertEqual(_RESPONSE, await call)
         self.assertEqual(grpc.StatusCode.OK, await call.code())
 
-    async def test_binary_metadata(self):
-        pass
-
     async def test_invalid_metadata(self):
-        pass
+        multicallable = self._client.unary_unary(_TEST_CLIENT_TO_SERVER)
+        for exception_type, metadata in _INVALID_METADATA_TEST_CASES:
+            call = multicallable(_REQUEST, metadata=metadata)
+            with self.assertRaises(exception_type):
+                await call
 
     async def test_generic_handler(self):
         multicallable = self._client.unary_unary(_TEST_GENERIC_HANDLER)

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

@@ -36,6 +36,7 @@ _STREAM_UNARY_EVILLY_MIXED = '/test/StreamUnaryEvillyMixed'
 _STREAM_STREAM_ASYNC_GEN = '/test/StreamStreamAsyncGen'
 _STREAM_STREAM_READER_WRITER = '/test/StreamStreamReaderWriter'
 _STREAM_STREAM_EVILLY_MIXED = '/test/StreamStreamEvillyMixed'
+_UNIMPLEMENTED_METHOD = '/test/UnimplementedMethod'
 
 _REQUEST = b'\x00\x00\x00'
 _RESPONSE = b'\x01\x01\x01'
@@ -393,6 +394,14 @@ class TestServer(AioTestBase):
         async with aio.insecure_channel('localhost:%d' % port) as channel:
             await channel.unary_unary(_SIMPLE_UNARY_UNARY)(_REQUEST)
 
+    async def test_unimplemented(self):
+        async with aio.insecure_channel(self._server_target) as channel:
+            call = channel.unary_unary(_UNIMPLEMENTED_METHOD)
+            with self.assertRaises(aio.AioRpcError) as exception_context:
+                await call(_REQUEST)
+            rpc_error = exception_context.exception
+            self.assertEqual(grpc.StatusCode.UNIMPLEMENTED, rpc_error.code())
+
 
 if __name__ == '__main__':
     logging.basicConfig(level=logging.DEBUG)