refactor(integration-tests): Use ExternalAction flows throughout the testing suite; deprecate log_subprocess_output_to_file().

integration-tests/tests/fixtures/integration_test_logs.py

@@ -2,14 +2,15 @@
 
 import logging
 import pathlib
-import subprocess
 
 import pytest
 
+from tests.utils.classes import ExternalAction
 from tests.utils.config import (
     IntegrationTestLogs,
     IntegrationTestPathConfig,
 )
+from tests.utils.logging_utils import format_action_failure_msg
 from tests.utils.utils import (
     get_binary_path,
     remove_path,
@@ -100,7 +101,7 @@ def _download_and_extract_gzip_dataset(
     :param keep_leading_dir: Whether to preserve the top-level directory during tarball extraction.
         Defaults to False to avoid an unnecessary extra directory level.
     :return: An IntegrationTestLogs instance providing metadata for the downloaded logs.
-    :raises subprocess.CalledProcessError: If `curl`, `tar`, or `chmod` fails.
+    :raise pytest.fail: If `curl`, `tar`, or `chmod` returns a non-zero exit code.
     """
     integration_test_logs = IntegrationTestLogs(
         name=name,
@@ -128,7 +129,14 @@ def _download_and_extract_gzip_dataset(
         tarball_url,
     ]
     # fmt: on
-    subprocess.run(curl_cmd, check=True)
+    curl_action = ExternalAction.from_cmd(curl_cmd)
+    if curl_action.completed_proc.returncode != 0:
+        pytest.fail(
+            format_action_failure_msg(
+                f"`curl` failed when downloading `{tarball_url}`.",
+                curl_action,
+            )
+        )
 
     # fmt: off
     extract_cmd = [
@@ -141,13 +149,34 @@ def _download_and_extract_gzip_dataset(
     # fmt: on
     if not keep_leading_dir:
         extract_cmd.extend(["--strip-components", "1"])
-    subprocess.run(extract_cmd, check=True)
+    extract_action = ExternalAction.from_cmd(extract_cmd)
+    if extract_action.completed_proc.returncode != 0:
+        pytest.fail(
+            format_action_failure_msg(
+                f"`tar` failed when extracting `{tarball_path_str}`.",
+                extract_action,
+            )
+        )
 
     # Allow the downloaded and extracted contents to be deletable or overwritable by adding write
     # permissions for both the user and the group.
     chmod_bin = get_binary_path("chmod")
-    subprocess.run([chmod_bin, "gu+w", tarball_path_str], check=True)
-    subprocess.run([chmod_bin, "-R", "gu+w", extract_path_str], check=True)
+    chmod_tarball_action = ExternalAction.from_cmd([chmod_bin, "gu+w", tarball_path_str])
+    if chmod_tarball_action.completed_proc.returncode != 0:
+        pytest.fail(
+            format_action_failure_msg(
+                f"`chmod` failed for `{tarball_path_str}`.",
+                chmod_tarball_action,
+            )
+        )
+    chmod_extract_action = ExternalAction.from_cmd([chmod_bin, "-R", "gu+w", extract_path_str])
+    if chmod_extract_action.completed_proc.returncode != 0:
+        pytest.fail(
+            format_action_failure_msg(
+                f"`chmod` failed for `{extract_path_str}`.",
+                chmod_extract_action,
+            )
+        )
 
     logger.info("Downloaded and extracted uncompressed logs for dataset `%s`.", name)
     request.config.cache.set(name, True)

integration-tests/tests/test_identity_transformation.py

@@ -5,17 +5,18 @@
 
 import pytest
 
+from tests.utils.classes import ExternalAction
 from tests.utils.config import (
     ClpCorePathConfig,
     CompressionTestPathConfig,
     IntegrationTestLogs,
     IntegrationTestPathConfig,
 )
-from tests.utils.subprocess_utils import run_and_log_subprocess
-from tests.utils.utils import (
+from tests.utils.fs_validation import (
     is_dir_tree_content_equal,
     is_json_file_structurally_equal,
 )
+from tests.utils.logging_utils import format_action_failure_msg
 
 pytestmark = pytest.mark.core
 
@@ -73,10 +74,14 @@ def test_clp_identity_transform(
         src_path,
     ]
     # fmt: on
-    run_and_log_subprocess(compression_cmd)
+    compression_action = ExternalAction.from_cmd(compression_cmd)
+    if compression_action.completed_proc.returncode != 0:
+        pytest.fail(format_action_failure_msg("`clp` compression failed.", compression_action))
 
     decompression_cmd = [bin_path, "x", compression_path, decompression_path]
-    run_and_log_subprocess(decompression_cmd)
+    decompression_action = ExternalAction.from_cmd(decompression_cmd)
+    if decompression_action.completed_proc.returncode != 0:
+        pytest.fail(format_action_failure_msg("`clp` decompression failed.", decompression_action))
 
     input_path = test_paths.logs_source_dir
     output_path = test_paths.decompression_dir
@@ -148,5 +153,14 @@ def _clp_s_compress_and_decompress(
     src_path = str(test_paths.logs_source_dir)
     compression_path = str(test_paths.compression_dir)
     decompression_path = str(test_paths.decompression_dir)
-    run_and_log_subprocess([bin_path, "c", compression_path, src_path])
-    run_and_log_subprocess([bin_path, "x", compression_path, decompression_path])
+    compression_action = ExternalAction.from_cmd([bin_path, "c", compression_path, src_path])
+    if compression_action.completed_proc.returncode != 0:
+        pytest.fail(format_action_failure_msg("`clp-s` compression failed.", compression_action))
+
+    decompression_action = ExternalAction.from_cmd(
+        [bin_path, "x", compression_path, decompression_path]
+    )
+    if decompression_action.completed_proc.returncode != 0:
+        pytest.fail(
+            format_action_failure_msg("`clp-s` decompression failed.", decompression_action)
+        )

integration-tests/tests/test_log_converter.py

@@ -7,13 +7,14 @@
 
 import pytest
 
+from tests.utils.classes import ExternalAction
 from tests.utils.config import (
     ClpCorePathConfig,
     ConversionTestPathConfig,
     IntegrationTestLogs,
     IntegrationTestPathConfig,
 )
-from tests.utils.subprocess_utils import run_and_log_subprocess
+from tests.utils.logging_utils import format_action_failure_msg
 
 # Matching `LogSerializer::cTimestampKey`.
 LOG_CONVERTER_OUTPUT_TIMESTAMP_KEY = "timestamp"
@@ -69,8 +70,13 @@ def _convert_and_compress(
     src_path = str(test_paths.logs_source_dir)
     conversion_path = str(test_paths.conversion_dir)
     compression_path = str(test_paths.compression_dir)
-    run_and_log_subprocess([log_converter_bin_path, src_path, "--output-dir", conversion_path])
-    run_and_log_subprocess(
+    conversion_action = ExternalAction.from_cmd(
+        [log_converter_bin_path, src_path, "--output-dir", conversion_path]
+    )
+    if conversion_action.completed_proc.returncode != 0:
+        pytest.fail(format_action_failure_msg("`log-converter` failed.", conversion_action))
+
+    compression_action = ExternalAction.from_cmd(
         [
             clp_s_bin_path,
             "c",
@@ -80,12 +86,18 @@ def _convert_and_compress(
             LOG_CONVERTER_OUTPUT_TIMESTAMP_KEY,
         ]
     )
+    if compression_action.completed_proc.returncode != 0:
+        pytest.fail(format_action_failure_msg("`clp-s` compression failed.", compression_action))
 
     if test_paths.num_log_events is None:
         return
 
-    output = run_and_log_subprocess([clp_s_bin_path, "s", compression_path, "timestamp > 0"])
-    lines = output.stdout.splitlines() if output.stdout else []
+    search_action = ExternalAction.from_cmd(
+        [clp_s_bin_path, "s", compression_path, "timestamp > 0"]
+    )
+    if search_action.completed_proc.returncode != 0:
+        pytest.fail(format_action_failure_msg("`clp-s` search failed.", search_action))
+    lines = search_action.completed_proc.stdout.splitlines()
     if len(lines) != test_paths.num_log_events:
         pytest.fail(
             f"Expected {test_paths.num_log_events} log events after conversion, "

integration-tests/tests/utils/asserting_utils.py

@@ -8,13 +8,14 @@
 from clp_py_utils.clp_config import ClpConfig
 from pydantic import ValidationError
 
+from tests.utils.classes import ExternalAction
 from tests.utils.clp_mode_utils import compare_mode_signatures
 from tests.utils.config import PackageInstance, PackageTestConfig
 from tests.utils.docker_utils import list_running_services_in_compose_project
-from tests.utils.subprocess_utils import run_and_log_subprocess
+from tests.utils.fs_validation import is_dir_tree_content_equal
+from tests.utils.logging_utils import format_action_failure_msg
 from tests.utils.utils import (
     clear_directory,
-    is_dir_tree_content_equal,
     load_yaml_to_dict,
 )
 
@@ -133,7 +134,14 @@ def verify_package_compression(
         ]
 
         # Run decompression command and assert that it succeeds.
-        run_and_log_subprocess(decompress_cmd)
+        decompress_action = ExternalAction.from_cmd(decompress_cmd)
+        if decompress_action.completed_proc.returncode != 0:
+            pytest.fail(
+                format_action_failure_msg(
+                    f"Decompression script `{decompress_script_path.name}` failed.",
+                    decompress_action,
+                )
+            )
 
         # Verify content equality.
         output_path = decompression_dir / path_to_original_dataset.relative_to(

integration-tests/tests/utils/classes.py

@@ -9,6 +9,7 @@
 
 import pytest
 from pydantic import BaseModel
+from typing_extensions import Self
 
 from tests.conftest import get_test_log_dir
 from tests.utils.utils import validate_dir_exists, validate_file_exists
@@ -131,26 +132,58 @@ def logs_path(self) -> Path:
         return self.dataset_root_dir / self.metadata.logs_subdir
 
 
+class CmdArgs(BaseModel, ABC):
+    """Abstract base class for all CLP command argument models."""
+
+    @abstractmethod
+    def to_cmd(self) -> list[str]:
+        """:return: list of command arguments constructed from this instance's data members."""
+
+
 @dataclass
 class ExternalAction:
-    """Metadata for an external action executed during an integration test."""
+    """
+    Metadata for an external action executed during an integration test. Instances should be
+    constructed via the `from_args` or `from_cmd` class methods.
+    """
 
     #: Command to pass to `subprocess.run()`.
     cmd: list[str]
 
+    #: Optional structured arguments for verification purposes. Not used by `ExternalAction` itself.
+    args: CmdArgs | None = None
+
     #: The completed process returned from `subprocess.run()`.
     completed_proc: subprocess.CompletedProcess[str] = field(init=False)
 
     #: Path to the file where this action's subprocess output was logged.
     log_file_path: Path = field(init=False)
 
+    @classmethod
+    def from_args(cls, args: CmdArgs) -> Self:
+        """:return: An `ExternalAction` whose `cmd` is derived from `args.to_cmd()`."""
+        return cls(cmd=args.to_cmd(), args=args)
+
+    @classmethod
+    def from_cmd(cls, cmd: list[str]) -> Self:
+        """:return: An `ExternalAction` for the given raw `cmd`, with no associated `args`."""
+        return cls(cmd=cmd, args=None)
+
     def __post_init__(self) -> None:
         """Execute the external action and log output."""
         if not self.cmd:
             pytest.fail("Cannot create `ExternalAction` object: `cmd` list is empty.")
+        if self.args is not None and self.cmd != self.args.to_cmd():
+            pytest.fail(
+                "Cannot create `ExternalAction` object: `cmd` does not match `args.to_cmd()`."
+            )
         self.completed_proc = self._run_subprocess()
         self._log_action_summary_to_file()
 
+    def get_output(self) -> str:
+        """:return: The combined stdout and stderr from the completed subprocess."""
+        return self.completed_proc.stdout + self.completed_proc.stderr
+
     def _run_subprocess(self) -> subprocess.CompletedProcess[str]:
         """
         Passes `self.cmd` to `subprocess.run()` with preset parameters:
@@ -220,11 +253,3 @@ def _log_action_summary_to_file(self) -> None:
             f"Subprocess returned. stdout and stderr written to log file: '{self.log_file_path}'"
         )
         logger.info(log_msg)
-
-
-class CmdArgs(BaseModel, ABC):
-    """Abstract base class for all CLP command argument models."""
-
-    @abstractmethod
-    def to_cmd(self) -> list[str]:
-        """:return: list of command arguments constructed from this instance's data members."""

integration-tests/tests/utils/docker_utils.py

@@ -1,7 +1,7 @@
 """Provide utility functions related to the use of Docker during integration tests."""
 
-import subprocess
-
+from tests.utils.classes import ExternalAction
+from tests.utils.logging_utils import format_action_failure_msg
 from tests.utils.utils import get_binary_path
 
 
@@ -11,6 +11,7 @@ def list_running_services_in_compose_project(project_name: str) -> list[str]:
 
     :param project_name:
     :return: List of the running services that belong to the specified Docker Compose project.
+    :raise RuntimeError: if `docker compose ps` returns a non-zero exit code.
     """
     docker_bin = get_binary_path("docker")
 
@@ -24,10 +25,17 @@ def list_running_services_in_compose_project(project_name: str) -> list[str]:
     ]
     # fmt: on
 
-    compose_ps_proc = subprocess.run(compose_ps_cmd, stdout=subprocess.PIPE, text=True, check=True)
+    compose_ps_action = ExternalAction.from_cmd(compose_ps_cmd)
+    if compose_ps_action.completed_proc.returncode != 0:
+        err_msg = format_action_failure_msg(
+            "`docker compose ps` failed with exit code"
+            f" {compose_ps_action.completed_proc.returncode} for project `{project_name}`.",
+            compose_ps_action,
+        )
+        raise RuntimeError(err_msg)
 
     service_names: list[str] = []
-    for line in (compose_ps_proc.stdout or "").splitlines():
+    for line in compose_ps_action.completed_proc.stdout.splitlines():
         service_name_candidate = line.strip()
         if service_name_candidate:
             service_names.append(service_name_candidate)