Fix Python interop tests

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

@@ -37,9 +37,16 @@ cdef int _INTERRUPT_CHECK_PERIOD_MS = 200
 
 cdef class CompletionQueue:
 
-  def __cinit__(self):
+  def __cinit__(self, shutdown_cq=False):
+    cdef grpc_completion_queue_attributes c_attrs
     grpc_init()
-    with nogil:
+    if shutdown_cq:
+      c_attrs.version = 1
+      c_attrs.cq_completion_type = GRPC_CQ_NEXT
+      c_attrs.cq_polling_type = GRPC_CQ_NON_LISTENING
+      self.c_completion_queue = grpc_completion_queue_create(
+          grpc_completion_queue_factory_lookup(&c_attrs), &c_attrs, NULL);
+    else:
       self.c_completion_queue = grpc_completion_queue_create_for_next(NULL)
     self.is_shutting_down = False
     self.is_shutdown = False

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

@@ -217,6 +217,20 @@ cdef extern from "grpc/grpc.h":
     GRPC_CALL_ERROR_INVALID_FLAGS
     GRPC_CALL_ERROR_INVALID_METADATA
 
+  ctypedef enum grpc_cq_completion_type:
+    GRPC_CQ_NEXT
+    GRPC_CQ_PLUCK
+
+  ctypedef enum grpc_cq_polling_type:
+    GRPC_CQ_DEFAULT_POLLING
+    GRPC_CQ_NON_LISTENING
+    GRPC_CQ_NON_POLLING
+
+  ctypedef struct grpc_completion_queue_attributes:
+    int version
+    grpc_cq_completion_type cq_completion_type
+    grpc_cq_polling_type cq_polling_type
+
   ctypedef enum grpc_connectivity_state:
     GRPC_CHANNEL_IDLE
     GRPC_CHANNEL_CONNECTING
@@ -309,6 +323,14 @@ cdef extern from "grpc/grpc.h":
   void grpc_init() nogil
   void grpc_shutdown() nogil
 
+  ctypedef struct grpc_completion_queue_factory:
+    pass
+
+  grpc_completion_queue_factory *grpc_completion_queue_factory_lookup(
+      const grpc_completion_queue_attributes* attributes) nogil
+  grpc_completion_queue *grpc_completion_queue_create(
+    const grpc_completion_queue_factory* factory,
+    const grpc_completion_queue_attributes* attr, void* reserved) nogil
   grpc_completion_queue *grpc_completion_queue_create_for_next(void *reserved) nogil
 
   grpc_event grpc_completion_queue_next(grpc_completion_queue *cq,

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

@@ -85,7 +85,7 @@ cdef class Server:
   def start(self):
     if self.is_started:
       raise ValueError("the server has already started")
-    self.backup_shutdown_queue = CompletionQueue()
+    self.backup_shutdown_queue = CompletionQueue(shutdown_cq=True)
     self.register_completion_queue(self.backup_shutdown_queue)
     self.is_started = True
     with nogil: