feat(integration-tests): Test compression with the CLP package.

integration-tests/.pytest.ini

@@ -4,6 +4,7 @@ addopts =
     --code-highlight=yes
     --color=yes
     -rA
+    -rA
     --strict-config
     --strict-markers
     --verbose
@@ -20,3 +21,5 @@ markers =
     clp_s: mark tests that use the CLP-S storage engine
     core: mark tests that test the CLP core binaries
     package: mark tests that run when the CLP package is active
+
+BASE_PORT = 55000

integration-tests/tests/conftest.py

@@ -1,9 +1,36 @@
 """Global pytest setup."""
 
+import pytest
+
+# Make the fixtures defined in `tests/fixtures/` globally available without imports.
 # Make the fixtures defined in `tests/fixtures/` globally available without imports.
 pytest_plugins = [
     "tests.fixtures.integration_test_logs",
     "tests.fixtures.path_configs",
     "tests.fixtures.package_instance",
     "tests.fixtures.package_config",
 ]
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """
+    Adds options for pytest.
+
+    :param parser:
+    """
+    parser.addini(
+        "BASE_PORT",
+        "Base port for CLP package integration tests.",
+        default="55000",
+    )
+    parser.addoption(
+        "--job-name",
+        dest="JOB_NAME",
+        help="Filter CLP jobs by substring of their job_name.",
+    )
+    parser.addoption(
+        "--no-jobs",
+        action="store_true",
+        dest="NO_JOBS",
+        help=("Only validate CLP package start and stop. Do not create or run any test jobs."),
+    )

integration-tests/tests/fixtures/package_config.py

@@ -1,14 +1,27 @@
 """Fixtures that create and remove temporary config files for CLP packages."""
 
+import shutil
 from collections.abc import Iterator
 
 import pytest
+from clp_py_utils.clp_config import (
+    CLP_DEFAULT_DATA_DIRECTORY_PATH,
+    CLP_DEFAULT_LOG_DIRECTORY_PATH,
+    CLP_DEFAULT_TMP_DIRECTORY_PATH,
+)
 
+from tests.utils.clp_job_utils import (
+    build_package_job_list,
+)
 from tests.utils.clp_mode_utils import (
     get_clp_config_from_mode,
     get_required_component_list,
 )
-from tests.utils.config import PackageConfig, PackagePathConfig
+from tests.utils.config import (
+    PackageConfig,
+    PackagePathConfig,
+)
+from tests.utils.port_utils import assign_ports_from_base
 
 
 @pytest.fixture
@@ -26,17 +39,42 @@ def fixt_package_config(
 
     clp_config_obj = get_clp_config_from_mode(mode_name)
 
+    # Assign ports based on BASE_PORT from ini.
+    base_port_string = request.config.getini("BASE_PORT")
+    try:
+        base_port = int(base_port_string)
+    except ValueError as err:
+        err_msg = (
+            f"Invalid BASE_PORT value '{base_port_string}' in pytest.ini; expected an integer."
+        )
+        raise ValueError(err_msg) from err
+
+    assign_ports_from_base(base_port, clp_config_obj)
+
     required_components = get_required_component_list(clp_config_obj)
 
+    # Build the job list for this mode and the current job filter.
+    no_jobs: bool = bool(request.config.option.NO_JOBS)
+    job_filter: str = request.config.option.JOB_NAME or ""
+    package_job_list = None if no_jobs else build_package_job_list(mode_name, job_filter)
+
     # Construct PackageConfig.
     package_config = PackageConfig(
         path_config=fixt_package_path_config,
         mode_name=mode_name,
         component_list=required_components,
         clp_config=clp_config_obj,
+        package_job_list=package_job_list,
     )
 
     try:
         yield package_config
     finally:
         package_config.temp_config_file_path.unlink(missing_ok=True)
+
+        # Clear data, tmp, and log from the package directory.
+        data_dir = package_config.path_config.clp_package_dir / CLP_DEFAULT_DATA_DIRECTORY_PATH
+        tmp_dir = package_config.path_config.clp_package_dir / CLP_DEFAULT_TMP_DIRECTORY_PATH
+        log_dir = package_config.path_config.clp_package_dir / CLP_DEFAULT_LOG_DIRECTORY_PATH
+        for directory_path in (data_dir, tmp_dir, log_dir):
+            shutil.rmtree(directory_path, ignore_errors=True)

integration-tests/tests/fixtures/package_instance.py

@@ -15,20 +15,34 @@
 
 
 @pytest.fixture
-def fixt_package_instance(fixt_package_config: PackageConfig) -> Iterator[PackageInstance]:
+def fixt_package_instance(
+    request: pytest.FixtureRequest,
+    fixt_package_config: PackageConfig,
+) -> Iterator[PackageInstance]:
     """
     Starts a CLP package instance for the given configuration and stops it during teardown.
 
     :param fixt_package_config:
     :return: Iterator that yields the running package instance.
     """
     mode_name = fixt_package_config.mode_name
+    no_jobs: bool = bool(request.config.option.NO_JOBS)
+    package_job_list = fixt_package_config.package_job_list
+
+    # Do not start this mode if there are no jobs and the '--no-jobs' flag wasn't specified by user.
+    if package_job_list is None and not no_jobs:
+        pytest.skip(f"No jobs to run for mode {mode_name} with current job filter.")
 
     try:
         start_clp_package(fixt_package_config)
         instance = PackageInstance(package_config=fixt_package_config)
         yield instance
     except RuntimeError:
-        pytest.fail(f"Failed to start the {mode_name} package.")
+        base_port_string = request.config.getini("BASE_PORT")
+        pytest.fail(
+            f"Failed to start the {mode_name} package. This could mean that one of the ports"
+            f" derived from BASE_PORT={base_port_string} was unavailable. Try changing BASE_PORT in"
+            " .pytest.ini."
+        )
     finally:
         stop_clp_package(fixt_package_config)

integration-tests/tests/test_package_start.py

@@ -4,8 +4,17 @@
 
 import pytest
 
+from tests.utils.asserting_utils import (
+    validate_package_running,
+    validate_running_mode_correct,
+)
+from tests.utils.clp_job_utils import (
+    dispatch_test_jobs,
+)
 from tests.utils.clp_mode_utils import CLP_MODE_CONFIGS
-from tests.utils.config import PackageInstance
+from tests.utils.config import (
+    PackageInstance,
+)
 
 TEST_MODES = CLP_MODE_CONFIGS.keys()
 
@@ -14,19 +23,21 @@
 
 @pytest.mark.package
 @pytest.mark.parametrize("fixt_package_config", TEST_MODES, indirect=True)
-def test_clp_package(fixt_package_instance: PackageInstance) -> None:
+def test_clp_package(
+    request: pytest.FixtureRequest, fixt_package_instance: PackageInstance
+) -> None:
     """
     Validate that the CLP package starts up successfully for the selected mode(s) of operation.
 
     :param fixt_package_instance:
     """
-    # TODO: write code that properly validates that the package is running. This is a placeholder.
-    mode_name = fixt_package_instance.package_config.mode_name
-    logger.info("The '%s' package has been spun up successfully.", mode_name)
-
-    component_list = fixt_package_instance.package_config.component_list
-    logger.info(
-        "The '%s' package runs with the following required components :\n%s",
-        mode_name,
-        "\n".join(component_list),
-    )
+    # Ensure that all package components are running.
+    validate_package_running(fixt_package_instance)
+
+    # Ensure that the package is running in the correct mode.
+    validate_running_mode_correct(fixt_package_instance)
+
+    # Run all jobs.
+    package_job_list = fixt_package_instance.package_config.package_job_list
+    if package_job_list is not None:
+        dispatch_test_jobs(request, fixt_package_instance)

integration-tests/tests/utils/asserting_utils.py

@@ -6,6 +6,15 @@
 from typing import Any
 
 import pytest
+from clp_py_utils.clp_config import ClpConfig
+from pydantic import ValidationError
+
+from tests.utils.clp_mode_utils import (
+    compute_mode_signature,
+)
+from tests.utils.config import PackageInstance
+from tests.utils.docker_utils import list_running_containers_with_prefix
+from tests.utils.utils import load_yaml_to_dict
 
 logger = logging.getLogger(__name__)
 
@@ -28,3 +37,47 @@ def run_and_assert(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess
     except subprocess.TimeoutExpired as e:
         pytest.fail(f"Command timed out: {' '.join(cmd)}: {e}")
     return proc
+
+
+def validate_package_running(package_instance: PackageInstance) -> None:
+    """
+    Validate that the given package instance is running. Each required component must have at least
+    one running container whose name matches the expected prefix. Calls pytest.fail on the first
+    missing component.
+
+    :param package_instance:
+    """
+    instance_id = package_instance.clp_instance_id
+    required_components = package_instance.package_config.component_list
+
+    for component in required_components:
+        prefix = f"clp-package-{instance_id}-{component}-"
+        running_matches = list_running_containers_with_prefix(prefix)
+        if len(running_matches) == 0:
+            pytest.fail(
+                f"No running container found for component '{component}' "
+                f"(expected name prefix '{prefix}')."
+            )
+
+
+def validate_running_mode_correct(package_instance: PackageInstance) -> None:
+    """
+    Validate that the mode described in the shared config of the instance matches the intended mode
+    defined by the instance configuration. Calls pytest.fail if the shared config fails validation
+    or if the running mode does not match the intended mode.
+
+    :param package_instance:
+    """
+    shared_config_dict = load_yaml_to_dict(package_instance.shared_config_file_path)
+    try:
+        running_config = ClpConfig.model_validate(shared_config_dict)
+    except ValidationError as err:
+        pytest.fail(f"Shared config failed validation: {err}")
+
+    intended_config = package_instance.package_config.clp_config
+
+    running_signature = compute_mode_signature(running_config)
+    intended_signature = compute_mode_signature(intended_config)
+
+    if running_signature != intended_signature:
+        pytest.fail("Mode mismatch: running configuration does not match intended configuration.")

integration-tests/tests/utils/clp_job_utils.py

@@ -0,0 +1,101 @@
+"""Provides utilities related to the test jobs for the CLP package."""
+
+import logging
+
+import pytest
+
+from tests.utils.config import (
+    PackageCompressJob,
+    PackageInstance,
+    PackageJobList,
+)
+from tests.utils.package_utils import compress_with_clp_package
+
+logger = logging.getLogger(__name__)
+
+
+PACKAGE_COMPRESS_JOBS: dict[str, PackageCompressJob] = {
+    "compress-postgresql": PackageCompressJob(
+        job_name="compress-postgresql",
+        log_fixture_name="postgresql",
+        mode="clp-json",
+        log_format="json",
+        unstructured=False,
+        dataset_name="postgresql",
+        timestamp_key="timestamp",
+    ),
+    "compress-hive-24hr": PackageCompressJob(
+        job_name="compress-hive-24hr",
+        log_fixture_name="hive_24hr",
+        mode="clp-text",
+        log_format="text",
+        unstructured=True,
+    ),
+    # Insert more compression jobs here as needed.
+}
+
+
+def _matches_keyword(job_name: str, keyword_filter: str) -> bool:
+    """Return True if this job should be included given the current -k filter."""
+    if not keyword_filter:
+        return True
+    return keyword_filter.lower() in job_name.lower()
+
+
+def build_package_job_list(mode_name: str, job_filter: str) -> PackageJobList | None:
+    """
+    Builds the list of package jobs for this test run.
+
+    :param mode_name:
+    :param job_filter:
+    :return: PackageJobList if there are jobs for this mode, None if not.
+    """
+    logger.debug("Creating job list for mode %s (job filter: %s)", mode_name, job_filter)
+
+    package_compress_jobs: list[PackageCompressJob] = []
+
+    for job_name, package_compress_job in PACKAGE_COMPRESS_JOBS.items():
+        if package_compress_job.mode == mode_name and _matches_keyword(job_name, job_filter):
+            package_compress_jobs.append(package_compress_job)
+
+    if not package_compress_jobs:
+        return None
+    return PackageJobList(
+        package_compress_jobs=package_compress_jobs,
+    )
+
+
+def _run_package_compress_jobs(
+    request: pytest.FixtureRequest,
+    package_instance: PackageInstance,
+) -> None:
+    """
+    Run all the package compress jobs for this test run.
+
+    :param package_instance:
+    :param request:
+    """
+    package_job_list = package_instance.package_config.package_job_list
+    if package_job_list is None:
+        err_msg = "Package job list is not configured for this package instance."
+        raise RuntimeError(err_msg)
+
+    compress_jobs = package_job_list.package_compress_jobs
+    for compress_job in compress_jobs:
+        compress_with_clp_package(request, compress_job, package_instance)
+
+
+def dispatch_test_jobs(request: pytest.FixtureRequest, package_instance: PackageInstance) -> None:
+    """
+    Dispatches all the package jobs in `job_list` for this package test run.
+
+    :param jobs_list:
+    """
+    logger.debug("Dispatching test jobs.")
+
+    jobs_list = package_instance.package_config.package_job_list
+    if jobs_list is None:
+        return
+
+    if jobs_list.package_compress_jobs:
+        _run_package_compress_jobs(request, package_instance)