Add extra (currently failing) test for previous commit.sam/output-multiplexing

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.