feat(integration-tests): Add infrastructure for running integration tests; Add basic integration tests for clp & clp-s compression.

.gitignore

@@ -2,3 +2,4 @@
 .task/
 build/
 **/dist/
+**/__pycache__/

docs/src/dev-guide/building-package.md

@@ -15,6 +15,7 @@ prebuilt version instead, check out the [releases](https://github.com/y-scope/cl
 * python3-venv (for the version of Python installed)
 * [Task] >= 3.38.0 and < 3.43.0
   * We constrain the version due to unresolved [issues][clp-issue-872].
+* [uv] 0.7.19 or newer
 
 ## Setup
 
@@ -61,6 +62,14 @@ where `<flavour>` is `json` or `text`.
 The tar will be written to `build/clp-<flavour>-<os>-<arch>-v<version>.tar.gz`, with appropriate
 values for the fields in angle brackets.
 
+## Test
+
+To test the package, run:
+
+```shell
+task test-package
+```
+
 ## Cleanup
 
 To clean up all build artifacts, run:
@@ -71,3 +80,4 @@ task clean
 
 [clp-issue-872]: https://github.com/y-scope/clp/issues/872
 [Task]: https://taskfile.dev/
+[uv]: https://docs.astral.sh/uv/

integration-tests/.pytest.ini

@@ -0,0 +1,15 @@
+[pytest]
+addopts =
+    --strict-config
+    --strict-markers
+    --capture=no
+    --verbose
+    --color=yes
+    --code-highlight=yes
+env =
+    D:CLP_BUILD_DIR=../build
+    D:CLP_PACKAGE_DIR=../build/clp-package
+markers =
+    binaries: mark tests that directly call the binaries in the package bin
+    clp: mark tests that use the CLP storage engine
+    clp_s: mark tests that use the CLP-S storage engine

integration-tests/.python-version

@@ -0,0 +1 @@
+3.10

integration-tests/pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "integration-tests"
+version = "0.1.0"
+description = "Integration tests for the CLP project."
+readme = "README.md"
+authors = [
+    { name = "YScope Inc.", email = "dev@yscope.com" }
+]
+requires-python = ">=3.10"
+
+[project.scripts]
+integration-tests = "integration_tests:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[dependency-groups]
+dev = [
+    "pytest>=8.3.5",
+    "pytest-benchmark>=5.1.0",
+    "pytest-env>=1.1.5",
+]

integration-tests/src/integration_tests/__init__.py

@@ -0,0 +1,2 @@
+def main() -> None:
+    print("Hello from integration-tests!")

integration-tests/tests/conftest.py

@@ -0,0 +1,7 @@
+from tests.fixtures.base_config import base_config
+from tests.fixtures.dataset_logs import (
+    download_and_extract_dataset,
+    hive_24hr,
+    postgresql,
+    spark_event_logs,
+)

integration-tests/tests/fixtures/base_config.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+import pytest
+from tests.utils.config import BaseConfig
+from tests.utils.utils import get_env_var
+
+
+@pytest.fixture(scope="session")
+def base_config() -> BaseConfig:
+    clp_build_dir = Path(get_env_var("CLP_BUILD_DIR")).resolve()
+    clp_package_dir = Path(get_env_var("CLP_PACKAGE_DIR")).resolve()
+
+    base_config = BaseConfig(
+        clp_bin_dir=clp_package_dir / "bin",
+        clp_package_dir=clp_package_dir,
+        clp_sbin_dir=clp_package_dir / "sbin",
+        test_output_dir=clp_build_dir / "var" / "logs" / "pytest",
+        uncompressed_logs_dir=clp_build_dir / "var" / "data" / "pytest" / "downloads",
+    )
+    base_config.test_output_dir.mkdir(parents=True, exist_ok=True)
+    base_config.uncompressed_logs_dir.mkdir(parents=True, exist_ok=True)
+    return base_config

integration-tests/tests/fixtures/dataset_logs.py

@@ -0,0 +1,71 @@
+import shutil
+
+import pytest
+from tests.utils.config import (
+    BaseConfig,
+    DatasetLogs,
+)
+from tests.utils.utils import run_and_assert
+
+
+@pytest.fixture(scope="session")
+def hive_24hr() -> DatasetLogs:
+    return DatasetLogs(
+        name="hive-24hr",
+        tar_url="https://zenodo.org/records/7094921/files/hive-24hr.tar.gz?download=1",
+    )
+
+
+@pytest.fixture(scope="session")
+def elasticsearch() -> DatasetLogs:
+    return DatasetLogs(
+        name="elasticsearch",
+        tar_url="https://zenodo.org/records/10516227/files/elasticsearch.tar.gz?download=1",
+    )
+
+
+@pytest.fixture(scope="session")
+def spark_event_logs() -> DatasetLogs:
+    return DatasetLogs(
+        name="spark-event-logs",
+        tar_url="https://zenodo.org/records/10516346/files/spark-event-logs.tar.gz?download=1",
+    )
+
+
+@pytest.fixture(scope="session")
+def postgresql() -> DatasetLogs:
+    return DatasetLogs(
+        name="postgresql",
+        tar_url="https://zenodo.org/records/10516402/files/postgresql.tar.gz?download=1",
+    )
+
+
+@pytest.fixture(autouse=True)
+def download_and_extract_dataset(request, base_config: BaseConfig) -> DatasetLogs:
+    dataset_config = request.getfixturevalue(request.param)
+    dataset_name = dataset_config.name
+    if request.config.cache.get(dataset_name, False):
+        print(f"Uncompressed logs for dataset `{dataset_name}` is up-to-date.")
+        return dataset_config
+
+    download_path = str(base_config.uncompressed_logs_dir / f"{dataset_name}.tar.gz")
+    extract_path = str(base_config.uncompressed_logs_dir / dataset_name)
+    # fmt: off
+    cmds = [
+        "curl",
+        "--fail",
+        "--location",
+        "--output", str(download_path),
+        "--show-error",
+        dataset_config.tar_url,
+    ]
+    # fmt: on
+    run_and_assert(cmds)
+
+    try:
+        shutil.unpack_archive(download_path, extract_path)
+    except:
+        assert False, f"Tar extraction failed for downloaded dataset `{dataset_name}`."
+
+    request.config.cache.set(dataset_name, True)
+    return dataset_config

integration-tests/tests/test_identity_transformation.py

@@ -0,0 +1,122 @@
+import shutil
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from typing import IO
+
+import pytest
+from tests.utils.config import (
+    BaseConfig,
+    DatasetLogs,
+)
+from tests.utils.utils import (
+    diff_equal,
+    run_and_assert,
+)
+
+pytestmark = pytest.mark.binaries
+
+text_datasets = pytest.mark.parametrize(
+    "download_and_extract_dataset",
+    [
+        "hive_24hr",
+    ],
+    indirect=["download_and_extract_dataset"],
+)
+
+json_datasets = pytest.mark.parametrize(
+    "download_and_extract_dataset",
+    [
+        "spark_event_logs",
+        "postgresql",
+    ],
+    indirect=["download_and_extract_dataset"],
+)
+
+
+@pytest.mark.clp
+@text_datasets
+def test_clp_identity_transform(
+    request, base_config: BaseConfig, download_and_extract_dataset: DatasetLogs
+) -> None:
+    binary_path_str = str(base_config.clp_bin_dir / "clp")
+    dataset_name = download_and_extract_dataset.name
+    download_dir = base_config.uncompressed_logs_dir / dataset_name
+    archives_dir = base_config.test_output_dir / f"{dataset_name}-archives"
+    extract_dir = base_config.test_output_dir / f"{dataset_name}-logs"
+
+    shutil.rmtree(archives_dir, ignore_errors=True)
+    shutil.rmtree(extract_dir, ignore_errors=True)
+
+    # fmt: off
+    compression_cmd = [
+        binary_path_str,
+        "c",
+        "--progress",
+        "--remove-path-prefix", str(download_dir),
+        str(archives_dir),
+        str(download_dir),
+    ]
+    # fmt: on
+    run_and_assert(compression_cmd)
+    run_and_assert([binary_path_str, "x", str(archives_dir), str(extract_dir)])
+
+    diff_equal(download_dir, extract_dir)
+
+    shutil.rmtree(archives_dir, ignore_errors=True)
+    shutil.rmtree(extract_dir, ignore_errors=True)
+
+
+@pytest.mark.clp_s
+@json_datasets
+def test_clp_s_identity_transform(
+    request, base_config: BaseConfig, download_and_extract_dataset: DatasetLogs
+) -> None:
+    binary_path_str = str(base_config.clp_bin_dir / "clp-s")
+    dataset_name = download_and_extract_dataset.name
+    download_dir = base_config.uncompressed_logs_dir / dataset_name
+    archives_dir = base_config.test_output_dir / f"{dataset_name}-archives"
+    extract_dir = base_config.test_output_dir / f"{dataset_name}-logs"
+
+    shutil.rmtree(archives_dir, ignore_errors=True)
+    shutil.rmtree(extract_dir, ignore_errors=True)
+
+    run_and_assert([binary_path_str, "c", str(archives_dir), str(download_dir)])
+    run_and_assert([binary_path_str, "x", str(archives_dir), str(extract_dir)])
+
+    # Recompress the decompressed single-file output and decompress it again to verify consistency.
+    # TODO: Remove this check once we can directly compare decompressed logs (which would preserve
+    #       the directory structure and row/key order) with the original downloaded logs.
+    # See also: https://docs.yscope.com/clp/main/user-guide/core-clp-s.html#current-limitations
+    single_file_archives_dir = base_config.test_output_dir / f"{dataset_name}-single-file-archives"
+    single_file_extract_dir = base_config.test_output_dir / f"{dataset_name}-single-file-logs"
+
+    shutil.rmtree(single_file_archives_dir, ignore_errors=True)
+    shutil.rmtree(single_file_extract_dir, ignore_errors=True)
+
+    run_and_assert([binary_path_str, "c", single_file_archives_dir, extract_dir])
+    run_and_assert([binary_path_str, "x", single_file_archives_dir, single_file_extract_dir])
+
+    # Key and row orders are not preserved during `clp-s` operations, so sort before diffing.
+    with _sort_json_keys_and_rows(extract_dir / "original") as s1, _sort_json_keys_and_rows(
+        single_file_extract_dir / "original"
+    ) as s2:
+        diff_equal(s1.name, s2.name)
+
+    shutil.rmtree(archives_dir, ignore_errors=True)
+    shutil.rmtree(extract_dir, ignore_errors=True)
+    shutil.rmtree(single_file_archives_dir, ignore_errors=True)
+    shutil.rmtree(single_file_extract_dir, ignore_errors=True)
+
+
+def _sort_json_keys_and_rows(json_fp: Path) -> IO[bytes]:
+    with NamedTemporaryFile(mode="w+", delete=True) as keys_sorted, NamedTemporaryFile(
+        mode="w+", delete=True
+    ) as flattened:
+        keys_and_rows_sorted = NamedTemporaryFile(mode="w+", delete=True)
+        run_and_assert(["jq", "--sort-keys", ".", str(json_fp)], stdout=keys_sorted)
+        keys_sorted.flush()
+        run_and_assert(["jq", ".", keys_sorted.name], stdout=flattened)
+        flattened.flush()
+        run_and_assert(["sort", flattened.name], stdout=keys_and_rows_sorted)
+        keys_and_rows_sorted.flush()
+        return keys_and_rows_sorted

integration-tests/tests/utils/config.py

@@ -0,0 +1,17 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class BaseConfig:
+    clp_bin_dir: Path
+    clp_package_dir: Path
+    clp_sbin_dir: Path
+    test_output_dir: Path
+    uncompressed_logs_dir: Path
+
+
+@dataclass(frozen=True)
+class DatasetLogs:
+    name: str
+    tar_url: str

integration-tests/tests/utils/utils.py

@@ -0,0 +1,25 @@
+import os
+import subprocess
+from pathlib import Path
+from typing import List
+
+
+def diff_equal(path1: Path, path2: Path) -> None:
+    cmd = ["diff", "--brief", "--recursive", str(path1), str(path2)]
+    proc = subprocess.run(cmd, stdout=subprocess.PIPE)
+    if 0 != proc.returncode:
+        if 1 == proc.returncode:
+            assert False, "Files/Directories don't match."
+        assert False, f"Command failed {' '.join(cmd)}"
+
+
+def get_env_var(var_name: str) -> str:
+    value = os.environ.get(var_name)
+    assert value is not None, f"Environment variable {var_name} is not set."
+    return value
+
+
+def run_and_assert(cmd: List[str], **kwargs) -> subprocess.CompletedProcess:
+    proc = subprocess.run(cmd, **kwargs)
+    assert 0 == proc.returncode, f"Command failed: {' '.join(cmd)}"
+    return proc