mrtoth/pipe-io-server
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fix test_stop (which was broken during a refactor)

Browse Source
This commit is contained in:
Kristóf Tóth 2019-07-29 18:14:18 +02:00
parent 846aec9fce
commit ed951f7ce9
1 changed files with 18 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
pipe_io_server/test_pipe_io_server.py
View File

@ -29,8 +29,8 @@ def get_test_init_params():
return join(here, 'in_pipe_tests'), join(here, 'out_pipe_tests') return join(here, 'in_pipe_tests'), join(here, 'out_pipe_tests')
def raise_if_thread_blocks(thread_target_function, unblock_function): def raise_if_thread_blocks(*, target, unblock_function):
thread = Thread(target=thread_target_function) thread = Thread(target=target)
thread.start() thread.start()
unblock_function() unblock_function()
thread.join(timeout=1) thread.join(timeout=1)
@ -72,11 +72,22 @@ def test_run_creates_pipes(io_pipes):
def test_stop(): def test_stop():
pipe_io = EchoPipeIOServer(*get_test_init_params()) pipe_io = EchoPipeIOServer(*get_test_init_params())
def open_close_in_pipe(): pipe_io.start()
pipe_io.stop() raise_if_thread_blocks(target=pipe_io.wait, unblock_function=pipe_io.stop)
raise_if_thread_blocks(pipe_io.run, open_close_in_pipe)
for path in (pipe_io.in_pipe, pipe_io.out_pipe): pipe_io = EchoPipeIOServer(*get_test_init_params())
assert not exists(path) 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)
pipe_io = EchoPipeIOServer(*get_test_init_params())
pipe_io.start()
with IOPipes(pipe_io.in_pipe, pipe_io.out_pipe) as iopipes:
test_message = token_urlsafe(randint(128, 256))
iopipes.send_message(test_message.encode())
assert test_message == iopipes.recv().decode()
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(