fergalmoran/picard
1
0
Fork 0
mirror of https://github.com/fergalmoran/picard.git synced 2025-12-30 13:19:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

suggestions included, minor refactor

Browse Source
This commit is contained in:
skelly37
2022-06-14 18:56:58 +02:00
parent c1d708cf5b
commit 6782ddf0c5
1 changed files with 12 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
test/test_util_pipe.py
View File

@@ -30,20 +30,22 @@ from picard.util import pipe
def pipe_listener(pipe_handler, end_of_sequence):
IGNORED_OUTPUT = {pipe.Pipe.MESSAGE_TO_IGNORE, pipe.Pipe.NO_RESPONSE_MESSAGE, "", end_of_sequence}
received = []
messages = []
while end_of_sequence not in messages:
messages = pipe_handler.read_from_pipe()
for message in messages:
if message not in (pipe.Pipe.MESSAGE_TO_IGNORE, pipe.Pipe.NO_RESPONSE_MESSAGE, "", end_of_sequence):
if message not in IGNORED_OUTPUT:
received.append(message)
return received
return tuple(received)
def pipe_writer(pipe_handler, to_send, end_of_sequence):
for m in to_send:
while not pipe_handler.send_to_pipe(m):
for message in to_send:
while not pipe_handler.send_to_pipe(message):
pass
while not pipe_handler.send_to_pipe(end_of_sequence):
pass
@@ -66,16 +68,15 @@ class TestPipe(PicardTestCase):
def test_pipe_protocol(self):
END_OF_SEQUENCE = "stop"
to_send = [["it", "tests", "picard", "pipe"],
["test", "number", "two"],
["my_music_file.mp3"]]
NUM_OF_TESTS = len(to_send)
to_send = (("it", "tests", "picard", "pipe"),
("test", "number", "two"),
("my_music_file.mp3",))
pipe_listener_handler = pipe.Pipe(PICARD_APP_NAME, PICARD_FANCY_VERSION_STR)
pipe_writer_handler = pipe.Pipe(PICARD_APP_NAME, PICARD_FANCY_VERSION_STR)
for i in range(NUM_OF_TESTS):
for messages in to_send:
__pool = concurrent.futures.ThreadPoolExecutor()
plistener = __pool.submit(pipe_listener, pipe_listener_handler, END_OF_SEQUENCE)
__pool.submit(pipe_writer, pipe_writer_handler, to_send[i], END_OF_SEQUENCE)
self.assertEqual(plistener.result(), to_send[i])
__pool.submit(pipe_writer, pipe_writer_handler, messages, END_OF_SEQUENCE)
self.assertEqual(plistener.result(), messages)