Merge pull request #20938 from gnossen/unary_stream_single_threaded_test

Add additional test runner for single-threaded unary-stream code path

src/python/grpcio/grpc/_channel.py

@@ -16,6 +16,7 @@
 import copy
 import functools
 import logging
+import os
 import sys
 import threading
 import time
@@ -33,6 +34,11 @@ _USER_AGENT = 'grpc-python/{}'.format(_grpcio_metadata.__version__)
 
 _EMPTY_FLAGS = 0
 
+# NOTE(rbellevi): No guarantees are given about the maintenance of this
+# environment variable.
+_DEFAULT_SINGLE_THREADED_UNARY_STREAM = os.getenv(
+    "GRPC_SINGLE_THREADED_UNARY_STREAM") is not None
+
 _UNARY_UNARY_INITIAL_DUE = (
     cygrpc.OperationType.send_initial_metadata,
     cygrpc.OperationType.send_message,
@@ -1334,7 +1340,7 @@ class Channel(grpc.Channel):
             used over the lifetime of the channel.
         """
         python_options, core_options = _separate_channel_options(options)
-        self._single_threaded_unary_stream = False
+        self._single_threaded_unary_stream = _DEFAULT_SINGLE_THREADED_UNARY_STREAM
         self._process_python_options(python_options)
         self._channel = cygrpc.Channel(
             _common.encode(target), _augment_options(core_options, compression),

src/python/grpcio_tests/tests/unit/_interceptor_test.py

@@ -18,6 +18,7 @@ import itertools
 import threading
 import unittest
 import logging
+import os
 from concurrent import futures
 
 import grpc
@@ -560,6 +561,10 @@ class InterceptorTest(unittest.TestCase):
             's1:intercept_service', 's2:intercept_service'
         ])
 
+    # NOTE: The single-threaded unary-stream path does not support the
+    # grpc.Future interface, so this test does not apply.
+    @unittest.skipIf(
+        os.getenv("GRPC_SINGLE_THREADED_UNARY_STREAM"), "Not supported.")
     def testInterceptedUnaryRequestStreamResponseWithError(self):
         request = _EXCEPTION_REQUEST
 

src/python/grpcio_tests/tests/unit/thread_pool.py

@@ -24,7 +24,7 @@ class RecordingThreadPool(futures.ThreadPoolExecutor):
         self._lock = threading.Lock()
         self._was_used = False
 
-    def submit(self, fn, *args, **kwargs):
+    def submit(self, fn, *args, **kwargs):  # pylint: disable=arguments-differ
         with self._lock:
             self._was_used = True
         self._tp_executor.submit(fn, *args, **kwargs)

tools/bazel.rc

@@ -86,3 +86,5 @@ build:basicprof --copt=-DNDEBUG
 build:basicprof --copt=-O2
 build:basicprof --copt=-DGRPC_BASIC_PROFILER
 build:basicprof --copt=-DGRPC_TIMERS_RDTSC
+
+build:python_single_threaded_unary_stream --test_env="GRPC_SINGLE_THREADED_UNARY_STREAM=true"

tools/internal_ci/linux/grpc_python_bazel_test_in_docker.sh

@@ -24,8 +24,11 @@ git clone /var/local/jenkins/grpc /var/local/git/grpc
 && git submodule update --init --reference /var/local/jenkins/grpc/${name} \
 ${name}')
 cd /var/local/git/grpc/test
-bazel test --spawn_strategy=standalone --genrule_strategy=standalone --test_output=errors //src/python/...
-bazel test --spawn_strategy=standalone --genrule_strategy=standalone --test_output=errors //examples/python/...
+TEST_TARGETS="//src/python/... //examples/python/..."
+BAZEL_FLAGS="--spawn_strategy=standalone --genrule_strategy=standalone --test_output=errors"
+bazel test ${BAZEL_FLAGS} ${TEST_TARGETS}
+bazel test --config=python_single_threaded_unary_stream ${BAZEL_FLAGS} ${TEST_TARGETS}
+
 
 # TODO(https://github.com/grpc/grpc/issues/19854): Move this to a new Kokoro
 # job.