diff options
author | Sam Thursfield <sam.thursfield@codethink.co.uk> | 2015-06-24 16:21:31 +0100 |
---|---|---|
committer | Sam Thursfield <sam.thursfield@codethink.co.uk> | 2015-06-24 16:21:31 +0100 |
commit | ddadbea082bbe532c25405ce0cd0df6c77f88b2b (patch) | |
tree | 9f287f680452bf1691d33fc6856affe98b13613e | |
parent | f4ece423ebf135992fc7b85b120ee5f20de24bc2 (diff) | |
download | sandboxlib-sam/output-multiplexing.tar.gz |
Add extra (currently failing) test for previous commit.sam/output-multiplexing
-rw-r--r-- | tests/test_utils.py | 61 |
1 files changed, 50 insertions, 11 deletions
diff --git a/tests/test_utils.py b/tests/test_utils.py index aa93480..c726e20 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -23,23 +23,62 @@ import threading import sandboxlib -def test_duplicate_streams(): - read_fd, write_fd = os.pipe() +def duplicate_and_assert_data(read_fd, expected_data): + '''Helper to test the utils.duplicate_streams() method.''' + + with tempfile.TemporaryFile() as f_1: + with tempfile.TemporaryFile() as f_2: + sandboxlib.utils.duplicate_streams({read_fd: [f_1, f_2]}) + f_1.seek(0) + assert f_1.read() == expected_data + f_2.seek(0) + assert f_2.read() == expected_data + + +def write_data_in_thread(write_fd, data): + '''Helper to test the utils.duplicate_streams() method. + + Since the utils.duplicate_streams() function blocks the main loop until + all data is transferred, we need to write test data in a separate thread. + + ''' def write_data(write_fd): write_f = os.fdopen(write_fd, 'wb') - write_f.write('hello\n'.encode('utf-8')) + write_f.write(data) write_f.close() write_thread = threading.Thread(target=write_data, args=[write_fd]) write_thread.run() - #os.close(write_fd) - with tempfile.TemporaryFile() as f_1: - with tempfile.TemporaryFile() as f_2: - sandboxlib.utils.duplicate_streams({read_fd: [f_1, f_2]}) - f_1.seek(0) - assert f_1.read().decode('utf-8') == 'hello\n' - f_2.seek(0) - assert f_2.read().decode('utf-8') == 'hello\n' +class TestDuplicateStreams(object): + def test_basic(self): + '''Write data through a pipe into two files at once.''' + + data = b'Hello\n' + read_fd, write_fd = os.pipe() + write_data_in_thread(write_fd, data) + duplicate_and_assert_data(read_fd, expected_data=data) + os.close(read_fd) + + def test_binary(self): + '''Ensure that the code can handle arbitrary binary data. + + It is a common mistake to make to assume data being processed is valid + UTF-8. POSIX streams don't enforce any kind of character encoding. + + ''' + + data = bytearray(range(0,255)) + read_fd, write_fd = os.pipe() + write_data_in_thread(write_fd, data) + duplicate_and_assert_data(read_fd, expected_data=data) + os.close(read_fd) + + + +# Ideas for tests: +# non-utf8 +# pause without sending EOF. +# send two streams to one file (stdout and stderr) and ensure ordering is OK. |