aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/python/grpcio_tests/tests/unit/_exit_test.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/python/grpcio_tests/tests/unit/_exit_test.py')
-rw-r--r--src/python/grpcio_tests/tests/unit/_exit_test.py243
1 files changed, 131 insertions, 112 deletions
diff --git a/src/python/grpcio_tests/tests/unit/_exit_test.py b/src/python/grpcio_tests/tests/unit/_exit_test.py
index 5a4a32887c..b99605dcb8 100644
--- a/src/python/grpcio_tests/tests/unit/_exit_test.py
+++ b/src/python/grpcio_tests/tests/unit/_exit_test.py
@@ -26,7 +26,6 @@
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
"""Tests clean exit of server/client on Python Interpreter exit/sigint.
The tests in this module spawn a subprocess for each test case, the
@@ -45,15 +44,15 @@ import unittest
from tests.unit import _exit_scenarios
-SCENARIO_FILE = os.path.abspath(os.path.join(
- os.path.dirname(os.path.realpath(__file__)), '_exit_scenarios.py'))
+SCENARIO_FILE = os.path.abspath(
+ os.path.join(
+ os.path.dirname(os.path.realpath(__file__)), '_exit_scenarios.py'))
INTERPRETER = sys.executable
BASE_COMMAND = [INTERPRETER, SCENARIO_FILE]
BASE_SIGTERM_COMMAND = BASE_COMMAND + ['--wait_for_interrupt']
INIT_TIME = 1.0
-
processes = []
process_lock = threading.Lock()
@@ -61,126 +60,146 @@ process_lock = threading.Lock()
# Make sure we attempt to clean up any
# processes we may have left running
def cleanup_processes():
- with process_lock:
- for process in processes:
- try:
- process.kill()
- except Exception:
- pass
+ with process_lock:
+ for process in processes:
+ try:
+ process.kill()
+ except Exception:
+ pass
+
+
atexit.register(cleanup_processes)
def interrupt_and_wait(process):
- with process_lock:
- processes.append(process)
- time.sleep(INIT_TIME)
- os.kill(process.pid, signal.SIGINT)
- process.wait()
+ with process_lock:
+ processes.append(process)
+ time.sleep(INIT_TIME)
+ os.kill(process.pid, signal.SIGINT)
+ process.wait()
def wait(process):
- with process_lock:
- processes.append(process)
- process.wait()
+ with process_lock:
+ processes.append(process)
+ process.wait()
@unittest.skip('https://github.com/grpc/grpc/issues/7311')
class ExitTest(unittest.TestCase):
- def test_unstarted_server(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.UNSTARTED_SERVER],
- stdout=sys.stdout, stderr=sys.stderr)
- wait(process)
-
- def test_unstarted_server_terminate(self):
- process = subprocess.Popen(
- BASE_SIGTERM_COMMAND + [_exit_scenarios.UNSTARTED_SERVER],
- stdout=sys.stdout)
- interrupt_and_wait(process)
-
- def test_running_server(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.RUNNING_SERVER],
- stdout=sys.stdout, stderr=sys.stderr)
- wait(process)
-
- def test_running_server_terminate(self):
- process = subprocess.Popen(
- BASE_SIGTERM_COMMAND + [_exit_scenarios.RUNNING_SERVER],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- def test_poll_connectivity_no_server(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY_NO_SERVER],
- stdout=sys.stdout, stderr=sys.stderr)
- wait(process)
-
- def test_poll_connectivity_no_server_terminate(self):
- process = subprocess.Popen(
- BASE_SIGTERM_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY_NO_SERVER],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- def test_poll_connectivity(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY],
- stdout=sys.stdout, stderr=sys.stderr)
- wait(process)
-
- def test_poll_connectivity_terminate(self):
- process = subprocess.Popen(
- BASE_SIGTERM_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- def test_in_flight_unary_unary_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_UNARY_UNARY_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
- def test_in_flight_unary_stream_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_UNARY_STREAM_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- def test_in_flight_stream_unary_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_STREAM_UNARY_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
- def test_in_flight_stream_stream_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_STREAM_STREAM_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
- def test_in_flight_partial_unary_stream_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_PARTIAL_UNARY_STREAM_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- def test_in_flight_partial_stream_unary_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_PARTIAL_STREAM_UNARY_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
-
- @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
- def test_in_flight_partial_stream_stream_call(self):
- process = subprocess.Popen(
- BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_PARTIAL_STREAM_STREAM_CALL],
- stdout=sys.stdout, stderr=sys.stderr)
- interrupt_and_wait(process)
+ def test_unstarted_server(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.UNSTARTED_SERVER],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ wait(process)
+
+ def test_unstarted_server_terminate(self):
+ process = subprocess.Popen(
+ BASE_SIGTERM_COMMAND + [_exit_scenarios.UNSTARTED_SERVER],
+ stdout=sys.stdout)
+ interrupt_and_wait(process)
+
+ def test_running_server(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.RUNNING_SERVER],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ wait(process)
+
+ def test_running_server_terminate(self):
+ process = subprocess.Popen(
+ BASE_SIGTERM_COMMAND + [_exit_scenarios.RUNNING_SERVER],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ def test_poll_connectivity_no_server(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY_NO_SERVER],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ wait(process)
+
+ def test_poll_connectivity_no_server_terminate(self):
+ process = subprocess.Popen(
+ BASE_SIGTERM_COMMAND +
+ [_exit_scenarios.POLL_CONNECTIVITY_NO_SERVER],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ def test_poll_connectivity(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ wait(process)
+
+ def test_poll_connectivity_terminate(self):
+ process = subprocess.Popen(
+ BASE_SIGTERM_COMMAND + [_exit_scenarios.POLL_CONNECTIVITY],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ def test_in_flight_unary_unary_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_UNARY_UNARY_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
+ def test_in_flight_unary_stream_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_UNARY_STREAM_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ def test_in_flight_stream_unary_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_STREAM_UNARY_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
+ def test_in_flight_stream_stream_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND + [_exit_scenarios.IN_FLIGHT_STREAM_STREAM_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
+ def test_in_flight_partial_unary_stream_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND +
+ [_exit_scenarios.IN_FLIGHT_PARTIAL_UNARY_STREAM_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ def test_in_flight_partial_stream_unary_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND +
+ [_exit_scenarios.IN_FLIGHT_PARTIAL_STREAM_UNARY_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
+
+ @unittest.skipIf(six.PY2, 'https://github.com/grpc/grpc/issues/6999')
+ def test_in_flight_partial_stream_stream_call(self):
+ process = subprocess.Popen(
+ BASE_COMMAND +
+ [_exit_scenarios.IN_FLIGHT_PARTIAL_STREAM_STREAM_CALL],
+ stdout=sys.stdout,
+ stderr=sys.stderr)
+ interrupt_and_wait(process)
if __name__ == '__main__':
- unittest.main(verbosity=2)
+ unittest.main(verbosity=2)
generated by cgit v1.2.3 (git 2.39.1) at 2024-05-29 01:02:48 +0000