diff options
Diffstat (limited to 'tests/test_utils.py')
-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. |