Add prepend_send_initial_metadata_op function

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

@@ -120,6 +120,15 @@ async def execute_batch(GrpcCallWrapper grpc_call_wrapper,
     batch_operation_tag.event(c_event)
 
 
+cdef prepend_send_initial_metadata_op(tuple ops, tuple metadata):
+    # Eventually, this function should be the only function that produces
+    # SendInitialMetadataOperation. So we have more control over the flag.
+    return (SendInitialMetadataOperation(
+        metadata,
+        _EMPTY_FLAG
+    ),) + ops
+
+
 async def _receive_message(GrpcCallWrapper grpc_call_wrapper,
                            object loop):
     """Retrives parsed messages from Core.
@@ -147,15 +156,9 @@ async def _send_message(GrpcCallWrapper grpc_call_wrapper,
                         bint metadata_sent,
                         object loop):
     cdef SendMessageOperation op = SendMessageOperation(message, _EMPTY_FLAG)
-    cdef tuple ops
-    if metadata_sent:
-        ops = (op,)
-    else:
-        ops = (
-            # Initial metadata must be sent before first outbound message.
-            SendInitialMetadataOperation(None, _EMPTY_FLAG),
-            op,
-        )
+    cdef tuple ops = (op,)
+    if not metadata_sent:
+        ops = prepend_send_initial_metadata_op(ops, None)
     await execute_batch(grpc_call_wrapper, ops, loop)
 
 
@@ -189,9 +192,7 @@ async def _send_error_status_from_server(GrpcCallWrapper grpc_call_wrapper,
         details,
         _EMPTY_FLAGS,
     )
-    cdef tuple ops
-    if metadata_sent:
-        ops = (op,)
-    else:
-        ops = (op, SendInitialMetadataOperation(None, _EMPTY_FLAG))
+    cdef tuple ops = (op,)
+    if not metadata_sent:
+        ops = prepend_send_initial_metadata_op(ops, None)
     await execute_batch(grpc_call_wrapper, ops, loop)

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

@@ -197,24 +197,22 @@ async def _finish_handler_with_unary_response(RPCState rpc_state,
     )
 
     # Assembles the batch operations
-    cdef Operation send_status_op = SendStatusFromServerOperation(
-        rpc_state.trailing_metadata,
-        StatusCode.ok,
-        b'',
-        _EMPTY_FLAGS,
-    )
     cdef tuple finish_ops
+    finish_ops = (
+        SendMessageOperation(response_raw, _EMPTY_FLAGS),
+        SendStatusFromServerOperation(
+            rpc_state.trailing_metadata,
+            StatusCode.ok,
+            b'',
+            _EMPTY_FLAGS,
+        ),
+    )
     if not rpc_state.metadata_sent:
-        finish_ops = (
-            send_status_op,
-            SendInitialMetadataOperation(None, _EMPTY_FLAGS),
-            SendMessageOperation(response_raw, _EMPTY_FLAGS),
-        )
-    else:
-        finish_ops = (
-            send_status_op,
-            SendMessageOperation(response_raw, _EMPTY_FLAGS),
+        finish_ops = prepend_send_initial_metadata_op(
+            finish_ops,
+            None
         )
+    rpc_state.metadata_sent = True
     rpc_state.status_sent = True
     await execute_batch(rpc_state, finish_ops, loop)
 
@@ -271,7 +269,11 @@ async def _finish_handler_with_stream_responses(RPCState rpc_state,
 
     cdef tuple finish_ops = (op,)
     if not rpc_state.metadata_sent:
-        finish_ops = (op, SendInitialMetadataOperation(None, _EMPTY_FLAGS))
+        finish_ops = prepend_send_initial_metadata_op(
+            finish_ops,
+            None
+        )
+    rpc_state.metadata_sent = True
     rpc_state.status_sent = True
     await execute_batch(rpc_state, finish_ops, loop)
 

src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi

@@ -41,7 +41,7 @@ cdef void _store_c_metadata(
       for index, (key, value) in enumerate(metadata):
         encoded_key = _encode(key)
         encoded_value = value if encoded_key[-4:] == b'-bin' else _encode(value)
-        if type(encoded_value) != bytes:
+        if not isinstance(encoded_value, bytes):
           raise TypeError('Binary metadata key="%s" expected bytes, got %s' % (
             key,
             type(encoded_value)

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

@@ -24,7 +24,6 @@ from tests_aio.unit import _common
 from tests_aio.unit._test_base import AioTestBase
 from src.proto.grpc.testing import messages_pb2, test_pb2_grpc
 
-
 _LOCAL_CANCEL_DETAILS_EXPECTATION = 'Locally cancelled by application!'
 _INITIAL_METADATA_TO_INJECT = (
     (_INITIAL_METADATA_KEY, 'extra info'),

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

@@ -76,7 +76,7 @@ class _TestGenericHandlerForMethods(grpc.GenericRpcHandler):
     async def _test_client_to_server(request, context):
         assert _REQUEST == request
         assert _common.seen_metadata(_INITIAL_METADATA_FROM_CLIENT_TO_SERVER,
-                             context.invocation_metadata())
+                                     context.invocation_metadata())
         return _RESPONSE
 
     @staticmethod
@@ -114,7 +114,7 @@ class _TestGenericHandlerItself(grpc.GenericRpcHandler):
 
     def service(self, handler_details):
         assert _common.seen_metadata(_INITIAL_METADATA_FOR_GENERIC_HANDLER,
-                             handler_details.invocation_metadata)
+                                     handler_details.invocation_metadata)
         return grpc.unary_unary_rpc_method_handler(self._method)