From 9557364219acd67ad17f188daf19b6fc837636e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Fri, 2 Aug 2019 19:58:26 +0200 Subject: [PATCH] Write more of those juice stop test cases --- pipe_io_server/test_pipe_io_server.py | 29 ++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/pipe_io_server/test_pipe_io_server.py b/pipe_io_server/test_pipe_io_server.py index 6a08847..62a13df 100644 --- a/pipe_io_server/test_pipe_io_server.py +++ b/pipe_io_server/test_pipe_io_server.py @@ -83,16 +83,20 @@ def test_no_manage_pipes(): assert not pipes_exist(pipe_io.in_pipe, pipe_io.out_pipe) -def test_stop(): +def test_start_stop(): pipe_io = EchoPipeIOServer(*get_test_init_params()) pipe_io.start() raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop) + +def test_start_open_stop(): pipe_io = EchoPipeIOServer(*get_test_init_params()) pipe_io.start() with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes: raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop) + +def test_start_open_rw_stop(): pipe_io = EchoPipeIOServer(*get_test_init_params()) pipe_io.start() with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes: @@ -103,6 +107,29 @@ def test_stop(): raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop) +def test_start_open_brokenpipe_stop(): + pipe_io = EchoPipeIOServer(*get_test_init_params()) + pipe_io.start() + with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes: + iopipes.out_pipe.close() + test_message = token_urlsafe(randint(128, 256)) + iopipes.send_message(test_message.encode()) + with pytest.raises(ValueError): + iopipes.recv() + raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop) + + +def test_start_open_eof_stop(): + pipe_io = EchoPipeIOServer(*get_test_init_params()) + pipe_io.start() + with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes: + iopipes.in_pipe.close() + test_message = token_urlsafe(randint(128, 256)) + with pytest.raises(ValueError): + iopipes.send_message(test_message.encode()) + raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop) + + @pytest.mark.parametrize( 'test_data', [ 'Cats and cheese',