Improve utils.execute_command, add a Pytest selftest for it, and run it in GHA

.editorconfig

@@ -10,6 +10,9 @@ indent_style = space
 insert_final_newline = true
 trim_trailing_whitespace = true
 
+[.github/workflows/*.{yml,yaml}]
+indent_size = 2
+
 [*.{py,robot}]
 indent_style = space
 indent_size = 4

.github/workflows/code_quality.yaml

@@ -114,6 +114,42 @@ jobs:
       - run: poetry check --lock
 
       - name: Install ruff
-        run: poetry install --only=dev --sync
+        run: poetry install --sync
 
       - run: poetry run ruff check ods_ci/
+
+  selftests:
+    name: selftests
+    runs-on: ubuntu-latest
+    env:
+        poetry_version: '1.7.1'
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Cache poetry in ~/.local
+        uses: actions/cache@v3
+        id: cached-home-local
+        with:
+          path: ~/.local
+          key: ${{ runner.os }}-local-${{ env.poetry_version }}
+
+      - name: Install poetry
+        if: steps.cached-home-local.outputs.cache-hit != 'true'
+        run: pip install poetry==${{ env.poetry_version }}
+
+      - name: Set up Python
+        id: setup-python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+          cache: 'poetry'
+
+      - name: Configure poetry
+        run: poetry env use "${{ steps.setup-python.outputs.python-path }}"
+
+      - name: Install deps
+        run: poetry install --sync
+
+      - run: poetry run pytest

ods_ci/selftests/utils/scripts/conftest.py

@@ -0,0 +1,2 @@
+# executable scripts in the scripts directory all manipulate `sys.path`
+import ods_ci.utils.scripts.ocm.ocm as _  # imported for side-effects

ods_ci/selftests/utils/scripts/test_util.py

@@ -0,0 +1,93 @@
+import contextlib
+import doctest
+import functools
+import io
+import logging
+import subprocess
+import sys
+import unittest
+import unittest.mock
+
+import pytest
+
+from ods_ci.utils.scripts import util
+
+
+@contextlib.contextmanager
+def capture_logging(level: int | None = None, formatter: logging.Formatter | None = None):
+    """Temporarily replaces all logging handlers with a StringIO logging handler.
+
+    Note: Pytest has the `caplog` fixture (`LogCaptureFixture` class) which functions similarly.
+
+    Usage example:
+
+    >>> with capture_logging(logging.DEBUG, formatter=logging.Formatter("%(message)s")) as string_io:
+    ...     logging.debug('debug message')
+    >>> print(string_io.getvalue(), end="")
+    debug message
+    """
+    string_io = io.StringIO()
+    ch = logging.StreamHandler(string_io)
+    if level:
+        ch.setLevel(level)
+
+    if formatter:
+        ch.setFormatter(formatter)
+
+    handlers = logging.root.handlers
+    logging.root.handlers = []
+    logging.root.addHandler(ch)
+    try:
+        yield string_io
+    finally:
+        logging.root.removeHandler(ch)
+        logging.root.handlers = handlers
+
+
+# https://stackoverflow.com/questions/5681330/using-doctests-from-within-unittests
+def load_tests(loader: unittest.TestLoader, tests: unittest.TestSuite, pattern: str) -> unittest.TestSuite:
+    tests.addTest(doctest.DocTestSuite())
+    return tests
+
+
+class TestExecuteCommand(unittest.TestCase):
+    def test_failed_to_run(self):
+        # without mocking, the test would be sensitive to `/bin/sh --version`
+        with unittest.mock.patch.object(
+            subprocess.Popen, "__init__", new=functools.partialmethod(subprocess.Popen.__init__, executable="/bin/bash")
+        ):
+            assert "No such file or directory" in util.execute_command("/this-file-does-not-exist", print_stdout=False)
+
+    def test_success(self):
+        assert util.execute_command("/bin/true") == ""
+
+    def test_fail(self):
+        assert util.execute_command("/bin/false") == ""
+
+    def test_stdout(self):
+        assert util.execute_command("echo stdout") == "stdout\n"
+
+    def test_stderr(self):
+        assert util.execute_command("echo stderr >&2") == "stderr\n"
+
+    def test_string_cmd(self):
+        assert util.execute_command("echo hello world", print_stdout=False) == "hello world\n"
+
+    def test_list_cmd_single_list_item(self):
+        assert util.execute_command(["echo hello world"], print_stdout=False) == "hello world\n"
+
+    def test_list_cmd_multiple_list_items(self):
+        # this is surprising, but it's what subprocess.Popen does
+        assert util.execute_command(["echo", "hello", "world"], print_stdout=False) == "\n"
+
+    def test_multiple_output_lines(self):
+        python = sys.executable
+        assert util.execute_command(f"""{python} -c 'print("a\\n"*13, end="")'""", print_stdout=False) == "a\n" * 13
+
+    @pytest.mark.slow
+    def test_many_long_output_lines(self):
+        python = sys.executable
+        assert (
+            util.execute_command(f"""{python} -c 'print(("a" * 40 + "\\n")*1_000_000, end="")'""", print_stdout=False)
+            == ("a" * 40 + "\n") * 1_000_000
+        )

ods_ci/utils/scripts/util.py

@@ -53,13 +53,12 @@ def read_yaml(filename):
         return None
 
 
-def execute_command(cmd):
+def execute_command(cmd, print_stdout=True):
     """
     Executes command in the local node, and print real-time output
     """
-    output = ""
+    log.info(f"CMD: {cmd}")
     try:
-        log.info(f"CMD: {cmd}")
         with subprocess.Popen(
             cmd,
             shell=True,
@@ -69,17 +68,16 @@ def execute_command(cmd):
             encoding="utf-8",
             errors="replace",
         ) as p:
-            while True:
-                line = p.stdout.readline()
-                if line != "":
-                    output += line + "\n"
-                elif p.poll() is not None:
-                    break
-            sys.stdout.flush()
-            log.info(f"OUTPUT: {output}")
-            return output
-    except:
-        return None
+            output = []
+            for line in p.stdout:
+                output.append(line)
+                if print_stdout:
+                    print(">:", line, end="")
+                    sys.stdout.flush()
+            return "".join(output)
+    except Exception as e:
+        log.exception(f"Starting the subprocess '{cmd}' failed", exc_info=e)
+    return None
 
 
 def oc_login(ocp_console_url, username, password, timeout=600):

pyproject.toml

@@ -157,3 +157,14 @@ skip-magic-trailing-comma = false
 
 docstring-code-format = true
 docstring-code-line-length = "dynamic"
+
+[tool.pytest.ini_options]
+addopts = "-rfEX -p doctest --doctest-modules --strict-markers --import-mode=importlib"
+python_files = ["test_*.py"]
+python_classes = ["Test"]
+python_functions = ["test"]
+testpaths = ["ods_ci/selftests/"]
+xfail_strict = true
+markers = [
+    "slow",
+]