tests various exec_create/exec_start combinations

Test the interation of the tty, demux and stream parameters

Signed-off-by: Corentin Henry <[email protected]>

Author: little-dude

tests/integration/api_exec_test.py

@@ -75,6 +75,75 @@ def test_exec_command_streaming(self):
             res += chunk
         assert res == b'hello\nworld\n'
 
+    def test_exec_command_demux(self):
+        container = self.client.create_container(
+            BUSYBOX, 'cat', detach=True, stdin_open=True)
+        id = container['Id']
+        self.client.start(id)
+        self.tmp_containers.append(id)
+
+        script = ' ; '.join([
+            # Write something on stdout
+            'echo hello out',
+            # Busybox's sleep does not handle sub-second times.
+            # This loops takes ~0.3 second to execute on my machine.
+            'for i in $(seq 1 50000); do echo $i>/dev/null; done',
+            # Write something on stderr
+            'echo hello err >&2'])
+        cmd = 'sh -c "{}"'.format(script)
+
+        # tty=False, stream=False, demux=False
+        res = self.client.exec_create(id, cmd)
+        exec_log = self.client.exec_start(res)
+        assert exec_log == b'hello out\nhello err\n'
+
+        # tty=False, stream=True, demux=False
+        res = self.client.exec_create(id, cmd)
+        exec_log = self.client.exec_start(res, stream=True)
+        assert next(exec_log) == b'hello out\n'
+        assert next(exec_log) == b'hello err\n'
+        with self.assertRaises(StopIteration):
+            next(exec_log)
+
+        # tty=False, stream=False, demux=True
+        res = self.client.exec_create(id, cmd)
+        exec_log = self.client.exec_start(res, demux=True)
+        assert exec_log == (b'hello out\n', b'hello err\n')
+
+        # tty=False, stream=True, demux=True
+        res = self.client.exec_create(id, cmd)
+        exec_log = self.client.exec_start(res, demux=True, stream=True)
+        assert next(exec_log) == (b'hello out\n', None)
+        assert next(exec_log) == (None, b'hello err\n')
+        with self.assertRaises(StopIteration):
+            next(exec_log)
+
+        # tty=True, stream=False, demux=False
+        res = self.client.exec_create(id, cmd, tty=True)
+        exec_log = self.client.exec_start(res)
+        assert exec_log == b'hello out\r\nhello err\r\n'
+
+        # tty=True, stream=True, demux=False
+        res = self.client.exec_create(id, cmd, tty=True)
+        exec_log = self.client.exec_start(res, stream=True)
+        assert next(exec_log) == b'hello out\r\n'
+        assert next(exec_log) == b'hello err\r\n'
+        with self.assertRaises(StopIteration):
+            next(exec_log)
+
+        # tty=True, stream=False, demux=True
+        res = self.client.exec_create(id, cmd, tty=True)
+        exec_log = self.client.exec_start(res, demux=True)
+        assert exec_log == (b'hello out\r\nhello err\r\n', b'')
+
+        # tty=True, stream=True, demux=True
+        res = self.client.exec_create(id, cmd, tty=True)
+        exec_log = self.client.exec_start(res, demux=True, stream=True)
+        assert next(exec_log) == (b'hello out\r\n', None)
+        assert next(exec_log) == (b'hello err\r\n', None)
+        with self.assertRaises(StopIteration):
+            next(exec_log)
+
     def test_exec_start_socket(self):
         container = self.client.create_container(BUSYBOX, 'cat',
                                                  detach=True, stdin_open=True)