Enable demux option in exec_run

`exec_run` now returns a tuple of bytes if demux is True the first
element being the stdout and the second the stderr of the exec_run call.

Implementation is courtesy of:
https://github.com/SatelliteQE/broker/blob/60a52941f2eb297ccbdf7a0fa0b932eb23ad926b/broker/binds/containers.py#L8-L48

Resolves: #322

Signed-off-by: Nicola Sella <[email protected]>

podman/api/output_utils.py

@@ -0,0 +1,49 @@
+"""Utility functions for dealing with stdout and stderr."""
+
+HEADER_SIZE = 8
+STDOUT = 1
+STDERR = 2
+
+
+# pylint: disable=line-too-long
+def demux_output(data_bytes):
+    """Demuxes the output of a container stream into stdout and stderr streams.
+
+    Stream data is expected to be in the following format:
+    - 1 byte: stream type (1=stdout, 2=stderr)
+    - 3 bytes: padding
+    - 4 bytes: payload size (big-endian)
+    - N bytes: payload data
+    ref: https://docs.podman.io/en/latest/_static/api.html?version=v5.0#tag/containers/operation/ContainerAttachLibpod
+
+    Args:
+        data_bytes: Bytes object containing the combined stream data.
+
+    Returns:
+        A tuple containing two bytes objects: (stdout, stderr).
+    """
+    stdout = b""
+    stderr = b""
+    while len(data_bytes) >= HEADER_SIZE:
+        # Extract header information
+        header, data_bytes = data_bytes[:HEADER_SIZE], data_bytes[HEADER_SIZE:]
+        stream_type = header[0]
+        payload_size = int.from_bytes(header[4:HEADER_SIZE], "big")
+        # Check if data is sufficient for payload
+        if len(data_bytes) < payload_size:
+            break  # Incomplete frame, wait for more data
+
+        # Extract and process payload
+        payload = data_bytes[:payload_size]
+        if stream_type == STDOUT:
+            stdout += payload
+        elif stream_type == STDERR:
+            stderr += payload
+        else:
+            # todo: Handle unexpected stream types
+            pass
+
+        # Update data for next frame
+        data_bytes = data_bytes[payload_size:]
+
+    return stdout, stderr

podman/domain/containers.py

@@ -9,6 +9,7 @@
 import requests
 
 from podman import api
+from podman.api.output_utils import demux_output
 from podman.domain.images import Image
 from podman.domain.images_manager import ImagesManager
 from podman.domain.manager import PodmanResource
@@ -164,8 +165,10 @@ def exec_run(
             demux: Return stdout and stderr separately
 
         Returns:
-            First item is the command response code
-            Second item is the requests response content
+            First item is the command response code.
+            Second item is the requests response content.
+            If demux is True, the second item is a tuple of
+            (stdout, stderr).
 
         Raises:
             NotImplementedError: method not implemented.
@@ -199,6 +202,9 @@ def exec_run(
         # get and return exec information
         response = self.client.get(f"/exec/{exec_id}/json")
         response.raise_for_status()
+        if demux:
+            stdout_data, stderr_data = demux_output(start_resp.content)
+            return response.json().get('ExitCode'), (stdout_data, stderr_data)
         return response.json().get('ExitCode'), start_resp.content
 
     def export(self, chunk_size: int = api.DEFAULT_CHUNK_SIZE) -> Iterator[bytes]:

podman/tests/integration/test_container_exec.py

@@ -0,0 +1,50 @@
+import unittest
+
+import podman.tests.integration.base as base
+from podman import PodmanClient
+
+# @unittest.skipIf(os.geteuid() != 0, 'Skipping, not running as root')
+
+class ContainersExecIntegrationTests(base.IntegrationTest):
+    """Containers integration tests for exec"""
+
+    def setUp(self):
+        super().setUp()
+
+        self.client = PodmanClient(base_url=self.socket_uri)
+        self.addCleanup(self.client.close)
+
+        self.alpine_image = self.client.images.pull("quay.io/libpod/alpine", tag="latest")
+        self.containers = []
+
+    def tearDown(self):
+        for container in self.containers:
+            container.remove(force=True)
+
+    def test_container_exec_run(self):
+        """Test any command that will return code 0 and no output"""
+        container = self.client.containers.create(self.alpine_image, command=["top"], detach=True)
+        container.start()
+        error_code, stdout = container.exec_run("echo hello")
+
+        self.assertEqual(error_code, 0)
+        self.assertEqual(stdout, b'\x01\x00\x00\x00\x00\x00\x00\x06hello\n')
+
+    def test_container_exec_run_errorcode(self):
+        """Test a failing command with stdout and stderr in a single bytestring"""
+        container = self.client.containers.create(self.alpine_image, command=["top"], detach=True)
+        container.start()
+        error_code, output = container.exec_run("ls nonexistent")
+
+        self.assertEqual(error_code, 1)
+        self.assertEqual(output, b"\x02\x00\x00\x00\x00\x00\x00+ls: nonexistent: No such file or directory\n")
+
+    def test_container_exec_run_demux(self):
+        """Test a failing command with stdout and stderr in a bytestring tuple"""
+        container = self.client.containers.create(self.alpine_image, command=["top"], detach=True)
+        container.start()
+        error_code, output = container.exec_run("ls nonexistent", demux=True)
+
+        self.assertEqual(error_code, 1)
+        self.assertEqual(output[0], b'')
+        self.assertEqual(output[1], b"ls: nonexistent: No such file or directory\n")