Write more of those juice stop test cases

This commit is contained in:
Kristóf Tóth 2019-08-02 19:58:26 +02:00
parent c39527d99e
commit 9557364219

View File

@ -83,16 +83,20 @@ def test_no_manage_pipes():
assert not pipes_exist(pipe_io.in_pipe, pipe_io.out_pipe) 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 = EchoPipeIOServer(*get_test_init_params())
pipe_io.start() pipe_io.start()
raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop) 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 = EchoPipeIOServer(*get_test_init_params())
pipe_io.start() pipe_io.start()
with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes: 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) 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 = EchoPipeIOServer(*get_test_init_params())
pipe_io.start() pipe_io.start()
with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes: 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) 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( @pytest.mark.parametrize(
'test_data', [ 'test_data', [
'Cats and cheese', 'Cats and cheese',