integration_test_logs.py

"""Session-scoped test log fixtures shared across integration tests."""

import logging
import pathlib

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,
)

logger = logging.getLogger(__name__)


@pytest.fixture(scope="session")
def hive_24hr(
    request: pytest.FixtureRequest,
    integration_test_path_config: IntegrationTestPathConfig,
) -> IntegrationTestLogs:
    """Provides shared `hive_24hr` test logs."""
    return _download_and_extract_gzip_dataset(
        request=request,
        integration_test_path_config=integration_test_path_config,
        name="hive-24hr",
        tarball_url="https://zenodo.org/records/7094921/files/hive-24hr.tar.gz?download=1",
    )


@pytest.fixture(scope="session")
def postgresql(
    request: pytest.FixtureRequest,
    integration_test_path_config: IntegrationTestPathConfig,
) -> IntegrationTestLogs:
    """Provides shared `postgresql` test logs."""
    return _download_and_extract_gzip_dataset(
        request=request,
        integration_test_path_config=integration_test_path_config,
        name="postgresql",
        tarball_url="https://zenodo.org/records/10516402/files/postgresql.tar.gz?download=1",
    )


@pytest.fixture(scope="session")
def simple_unstructured(
    request: pytest.FixtureRequest,
    integration_test_path_config: IntegrationTestPathConfig,
) -> IntegrationTestLogs:
    """Provides a simple unstructured test log."""
    name = "simple_unstructured"
    integration_test_logs = IntegrationTestLogs(
        name=name,
        tarball_url=f"{name}.tar.gz",
        integration_test_path_config=integration_test_path_config,
        num_log_events=11,
    )
    remove_path(integration_test_logs.extraction_dir)
    integration_test_logs.extraction_dir.mkdir(parents=True, exist_ok=False)

    with pathlib.Path.open(integration_test_logs.extraction_dir / f"{name}.log", "w") as f:
        f.write(
            "2015-03-23 05:48:30,122 TEST1\n"
            "2015-03-23 05:48:30,122Z TEST2\n"
            "2015-03-23 05:48:30,122 Z TEST3\n"
            "2015-03-23 05:48:30,122+00 TEST4\n"
            "2015-03-23 05:48:30,122+00Z TEST5\n"
            "2015-03-23 05:48:30,122 +00 TEST6\n"
            "2015-03-23 05:48:30,122 +00Z TEST7\n"
            "2015-03-23 05:48:30,122UTC+00 TEST8\n"
            "2015-03-23 05:48:30,122UTC+00Z TEST9\n"
            "2015-03-23 05:48:30,122 UTC+00 TEST10\n"
            "2015-03-23 05:48:30,122 UTC+00Z TEST11\n"
        )

    logger.info("Set up logs for dataset `%s`.", name)
    request.config.cache.set(name, True)
    return integration_test_logs


def _download_and_extract_gzip_dataset(
    request: pytest.FixtureRequest,
    integration_test_path_config: IntegrationTestPathConfig,
    name: str,
    tarball_url: str,
    keep_leading_dir: bool = False,
) -> IntegrationTestLogs:
    """
    Download and extract a gzip-compressed dataset tarball for setting up the `IntegrationTestLogs`
    fixture. Adjust its file permissions for test use.

    :param request: Provides access to the pytest cache.
    :param integration_test_path_config: See `IntegrationTestPathConfig`.
    :param name: Dataset name.
    :param tarball_url: Dataset tarball URL.
    :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.
    :raise pytest.fail: If `curl`, `tar`, or `chmod` returns a non-zero exit code.
    """
    integration_test_logs = IntegrationTestLogs(
        name=name,
        tarball_url=tarball_url,
        integration_test_path_config=integration_test_path_config,
    )
    if request.config.cache.get(name, False):
        logger.info("Test logs `%s` are up-to-date. Skipping download.", name)
        return integration_test_logs

    remove_path(integration_test_logs.tarball_path)
    remove_path(integration_test_logs.extraction_dir)
    integration_test_logs.extraction_dir.mkdir(parents=True, exist_ok=False)

    tarball_path_str = str(integration_test_logs.tarball_path)
    extract_path_str = str(integration_test_logs.extraction_dir)

    # fmt: off
    curl_cmd = [
        get_binary_path("curl"),
        "--fail",
        "--location",
        "--output", tarball_path_str,
        "--show-error",
        tarball_url,
    ]
    # fmt: on
    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 = [
        get_binary_path("tar"),
        "--extract",
        "--gzip",
        "--file", tarball_path_str,
        "--directory", extract_path_str,
    ]
    # fmt: on
    if not keep_leading_dir:
        extract_cmd.extend(["--strip-components", "1"])
    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")
    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)
    return integration_test_logs