diff --git a/elt-common/src/elt_common/cli.py b/elt-common/src/elt_common/cli.py index f91edcd7..8931bc58 100644 --- a/elt-common/src/elt_common/cli.py +++ b/elt-common/src/elt_common/cli.py @@ -7,7 +7,6 @@ from typing import Literal, Optional, cast import click -import requests from elt_common.pipeline import PipelinesProject from elt_common.pipeline_types import ELTIngestManifest @@ -16,11 +15,6 @@ LOGGER = logging.getLogger(__name__) -# Disable ipv6 for anything using requests, which includes pyiceberg -# This is done to solve a performance problem when using WSL -# https://github.com/ISISNeutronMuon/analytics-data-platform/issues/433 -requests.packages.urllib3.util.connection.HAS_IPV6 = False - @click.group(context_settings={"show_default": True}) @click.option( diff --git a/elt-common/src/elt_common/ingest.py b/elt-common/src/elt_common/ingest.py index 6d29c8a5..97e28558 100644 --- a/elt-common/src/elt_common/ingest.py +++ b/elt-common/src/elt_common/ingest.py @@ -10,6 +10,7 @@ import pyarrow as pa import pyarrow.compute as pc +import requests from dotenv import load_dotenv, find_dotenv from pyiceberg.exceptions import NoSuchTableError @@ -27,6 +28,12 @@ LOGGER = logging.getLogger(__name__) +# Disable ipv6 for anything using requests, which includes pyiceberg +# This is done to solve a performance problem when using WSL +# https://github.com/ISISNeutronMuon/analytics-data-platform/issues/433 +requests.packages.urllib3.util.connection.HAS_IPV6 = False + + def run_ingest(job: ELTIngestManifest) -> dict[str, int]: """Import the extract function, call it, and write results to Iceberg.""" diff --git a/elt-common/src/elt_common/pipeline.py b/elt-common/src/elt_common/pipeline.py index 6059fe71..7e662039 100644 --- a/elt-common/src/elt_common/pipeline.py +++ b/elt-common/src/elt_common/pipeline.py @@ -78,7 +78,7 @@ def _discover_jobs(warehouse_name: str, ingest_dir: Path): """ return [ - _create_ingest_manifest(warehouse_name, job_dir) + create_ingest_manifest(warehouse_name, job_dir) for domain_dir in ingest_dir.iterdir() if domain_dir.is_dir() for job_dir in domain_dir.iterdir() @@ -86,7 +86,7 @@ def _discover_jobs(warehouse_name: str, ingest_dir: Path): ] -def _create_ingest_manifest(warehouse_name: str, job_dir: Path) -> ELTIngestManifest: +def create_ingest_manifest(warehouse_name: str, job_dir: Path) -> ELTIngestManifest: return ELTIngestManifest( warehouse_name=warehouse_name, name=job_dir.name, diff --git a/elt-common/src/elt_common/testing/fixtures.py b/elt-common/src/elt_common/testing/fixtures.py index 0e742f39..2ae2c71b 100644 --- a/elt-common/src/elt_common/testing/fixtures.py +++ b/elt-common/src/elt_common/testing/fixtures.py @@ -2,18 +2,19 @@ import tempfile import time -from typing import Generator import urllib.parse -import shutil import warnings +from collections.abc import Generator +from pathlib import Path +from tempfile import TemporaryDirectory -from minio import Minio import pytest import tenacity +from minio import Minio from . import DEFAULT_RETRY_ARGS from .dlt import PyIcebergDestinationTestConfiguration -from .lakekeeper import Settings, Server +from .lakekeeper import Server, Settings from .sqlcatalog import SqlCatalogWarehouse @@ -28,10 +29,11 @@ def warehouse(settings: Settings) -> Generator: ) if settings.catalog_type == "sql": - warehouse = SqlCatalogWarehouse(settings.warehouse_name) + d = TemporaryDirectory() + warehouse = SqlCatalogWarehouse(settings.warehouse_name, Path(d.name)) def cleanup_func(): - shutil.rmtree(warehouse.workdir.name) + d.cleanup() else: server = Server(settings) storage_config = settings.storage_config() diff --git a/elt-common/src/elt_common/testing/pipelines.py b/elt-common/src/elt_common/testing/pipelines.py new file mode 100644 index 00000000..589c086f --- /dev/null +++ b/elt-common/src/elt_common/testing/pipelines.py @@ -0,0 +1,125 @@ +"""Pytest fixtures and utilities for e2e testing ingest pipelines. + +The `test_catalog` fixture creates a temporary file based iceberg warehouse, +loads its catalog, and provides methods which tests can use to make assertions +about the data which gets written to it. + +`run_test_ingest` exposes a method that runs the pipeline under test, optionally +with configuration values provided as kwargs. + +Both rely on the test file being named 'test_.py' and existing in the same +directory as the pipeline it tests, in the elt-common directory structure: + +/ +|-- ingest/ +| |-- / +| | |-- / +| | | |-- .py +| | | |-- test_.py +""" + +import logging +from pathlib import Path + +import pytest +from pyiceberg.catalog import Catalog + +from elt_common.ingest import run_ingest +from elt_common.pipeline import create_ingest_manifest +from elt_common.testing.sqlcatalog import SqlCatalogWarehouse + +LOGGER = logging.getLogger(__name__) + + +class AssertableCatalog: + """Wraps an iceberg catalog with convenience methods for making assertions + about the data in it""" + + def __init__(self, catalog: Catalog): + self._catalog = catalog + + def clean_catalog(self): + for ns in self._catalog.list_namespaces(): + tables = self._catalog.list_tables(ns) + for qualified_table_name in tables: + self._catalog.purge_table(qualified_table_name) + + self._catalog.drop_namespace(ns) + + @property + def catalog(self): + return self._catalog + + def assert_has_exact_tables(self, namespace: str, tables: list[str]): + actual = self._catalog.list_tables(namespace) + assert set(actual) == {(namespace, table) for table in tables} + + def assert_has_columns(self, table_id: tuple[str, ...], column_names): + assert self._catalog.table_exists(table_id), f"{table_id} doesn't exist" + t = self._catalog.load_table(table_id) + for c in column_names: + assert c in t.schema().column_names + + def assert_has_n_rows(self, table_id: tuple[str, ...], n: int): + assert self.get_num_rows(table_id) == n + + def get_num_rows(self, table_id: tuple[str, ...]): + assert self._catalog.table_exists(table_id) + t = self._catalog.load_table(table_id) + return t.scan().count() + + +@pytest.fixture(scope="session") +def sql_warehouses(request, tmp_path_factory): + test_dir = tmp_path_factory.mktemp("warehouses") + test_paths = (t.path for t in request.session.items) + test_warehouse_names = {_get_warehouse_name_from_test_filepath(tp) for tp in test_paths} + LOGGER.debug(f"Creating {test_warehouse_names} warehouses in {test_dir}") + + warehouses = { + warehouse_name: SqlCatalogWarehouse(warehouse_name, test_dir) + for warehouse_name in test_warehouse_names + } + + return warehouses + + +@pytest.fixture +def test_catalog(sql_warehouses, request, monkeypatch): + test_warehouse_name = _get_warehouse_name_from_test_filepath(request.path) + warehouse = sql_warehouses[test_warehouse_name] + + monkeypatch.setenv("PYICEBERG_CATALOG__DEFAULT__TYPE", "sql") + monkeypatch.setenv("PYICEBERG_CATALOG__DEFAULT__URI", warehouse.uri) + monkeypatch.setenv("PYICEBERG_CATALOG__DEFAULT__WAREHOUSE", test_warehouse_name) + + catalog = warehouse.connect() + ret = AssertableCatalog(catalog) + + try: + yield ret + finally: + try: + ret.clean_catalog() + finally: + catalog.close() + + +@pytest.fixture +def run_test_ingest(request, monkeypatch): + w = _get_warehouse_name_from_test_filepath(request.path) + manifest = create_ingest_manifest(w, request.path.parent) + + def run(**config_vars): + LOGGER.debug(f"Running ingest for manifest {manifest} with config vars {config_vars}") + for k, v in config_vars.items(): + monkeypatch.setenv(f"{manifest.name}__{k}", v) + + run_ingest(manifest) + + return run + + +def _get_warehouse_name_from_test_filepath(fp: Path): + """Relies on the test file being '/ingest///test_.py'""" + return fp.parent.parent.parent.parent.name diff --git a/elt-common/src/elt_common/testing/sqlcatalog.py b/elt-common/src/elt_common/testing/sqlcatalog.py index 13dbbc00..19a5475b 100644 --- a/elt-common/src/elt_common/testing/sqlcatalog.py +++ b/elt-common/src/elt_common/testing/sqlcatalog.py @@ -1,16 +1,17 @@ -from tempfile import TemporaryDirectory +from pathlib import Path -from elt_common.dlt_destinations.pyiceberg.configuration import PyIcebergSqlCatalogCredentials from pyiceberg.catalog import Catalog as PyIcebergCatalog from pyiceberg.catalog import load_catalog +from elt_common.dlt_destinations.pyiceberg.configuration import PyIcebergSqlCatalogCredentials + class SqlCatalogWarehouse: - def __init__(self, warehouse_name: str): + def __init__(self, warehouse_name: str, workdir: Path): self.name = warehouse_name - self.workdir = TemporaryDirectory() - self.uri = f"sqlite:///{self.workdir.name}/{warehouse_name}.db" - self.warehouse_path = f"file://{self.workdir.name}/{warehouse_name}" + self.workdir = workdir + self.uri = f"sqlite:///{self.workdir}/{warehouse_name}.db" + self.warehouse_path = f"file://{self.workdir}/{warehouse_name}" def connect(self) -> PyIcebergCatalog: """Connect to the warehouse in the catalog""" diff --git a/elt-pipelines/README.md b/elt-pipelines/README.md index 5f5fdd48..5265cbb0 100644 --- a/elt-pipelines/README.md +++ b/elt-pipelines/README.md @@ -95,6 +95,17 @@ class Extract(BaseExtract): There is functionality built in to `elt_common` for some common types of pipeline. These can be found in the [`sources` package](../elt-common/src/elt_common/sources). +### Testing + +To test a pipeline, create a `test_.py` file in the same directory as the pipeline script and use +the `test_catalog` and `run_test_ingest` fixtures defined in `elt_common.testing.pipelines`. + +Tests are invoked with `pytest --disable-plugin-autoload -p elt_common.testing.pipelines`. The plugin arguments +are needed temporarily, until the `elt_common.testing.fixtures` module (which provides fixtures for `dlt` based +testing) is removed. + +**Configuration values need to be set for any pipeline(s) under test, as per [above](#configuration).** + ## Directory structure The project uses the following directory structure: diff --git a/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py b/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py new file mode 100644 index 00000000..473153ee --- /dev/null +++ b/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py @@ -0,0 +1,42 @@ +from elt_common.testing.pipelines import AssertableCatalog + +_namespace = "accelerator_statusdisplay" +_table_name = "elt_cycles" +_table_id = (_namespace, _table_name) + + +def test_expected_columns_created(test_catalog: AssertableCatalog, run_test_ingest): + run_test_ingest() + test_catalog.assert_has_columns( + _table_id, + [ + "id", + "label", + "status", + "phases", + "phases.element", # phases is a list type + ], + ) + num_rows = test_catalog.get_num_rows(_table_id) + assert num_rows >= 153, f"Found {num_rows} cycles, expected at least 153" + + test_catalog.clean_catalog() + + assert not test_catalog.catalog.table_exists(_table_id), ( + "Table should have been cleaned up" + ) + + +def test_multiple_runs(test_catalog: AssertableCatalog, run_test_ingest): + run_test_ingest() + n = test_catalog.get_num_rows(_table_id) + run_test_ingest() + run_test_ingest() + + # Check runs are overwriting, not appending + test_catalog.assert_has_n_rows(_table_id, n) + + nss = test_catalog.catalog.list_namespaces() + assert len(nss) == 1, "There should be a single namespace" + tables = test_catalog.catalog.list_tables(_namespace) + assert len(tables) == 1, "There should be a single table" diff --git a/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py b/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py new file mode 100644 index 00000000..9d057ffb --- /dev/null +++ b/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py @@ -0,0 +1,31 @@ +import json +import pytest + +from elt_common.testing.pipelines import AssertableCatalog + +_ns = "fase_proposal" + + +def test_expected_columns_created(test_catalog: AssertableCatalog, run_test_ingest): + run_test_ingest(tables=json.dumps(["call"])) + expected_columns = [ + "call_id", + "call_short_code", + "start_call", + "end_call", + "start_review", + "end_review", + ] + test_catalog.assert_has_columns((_ns, "call"), expected_columns) + + +def test_multiple_tables(test_catalog: AssertableCatalog, run_test_ingest): + tables = ["call", "countries", "questions"] + run_test_ingest(tables=json.dumps(tables), row_limit="5") + test_catalog.assert_has_exact_tables(_ns, tables) + + +@pytest.mark.parametrize("row_limit", [1, 5, 20]) +def test_row_limit_applied(test_catalog: AssertableCatalog, run_test_ingest, row_limit): + run_test_ingest(row_limit=str(row_limit), tables=json.dumps(["call"])) + test_catalog.assert_has_n_rows((_ns, "call"), row_limit) diff --git a/elt-pipelines/pyproject.toml b/elt-pipelines/pyproject.toml index 5bdc8562..d78c076a 100644 --- a/elt-pipelines/pyproject.toml +++ b/elt-pipelines/pyproject.toml @@ -41,4 +41,5 @@ elt-common = { path = "../elt-common", editable = true } [dependency-groups] dev = [ "prek>=0.4.5", + "pytest>=9.1.1", ] diff --git a/elt-pipelines/uv.lock b/elt-pipelines/uv.lock index 82b942bf..6d106934 100644 --- a/elt-pipelines/uv.lock +++ b/elt-pipelines/uv.lock @@ -785,6 +785,7 @@ statusdisplay = [ [package.dev-dependencies] dev = [ { name = "prek" }, + { name = "pytest" }, ] [package.metadata] @@ -806,7 +807,10 @@ requires-dist = [ provides-extras = ["proposal", "statusdisplay", "sharepoint", "opralogweb", "moderator-performance", "jira"] [package.metadata.requires-dev] -dev = [{ name = "prek", specifier = ">=0.4.5" }] +dev = [ + { name = "prek", specifier = ">=0.4.5" }, + { name = "pytest", specifier = ">=9.1.1" }, +] [[package]] name = "et-xmlfile" @@ -1090,6 +1094,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/38/3d/2d244233ac4f76e38533cfcb2991c9eb4c7bf688ae0a036d30725b8faafe/importlib_metadata-9.0.0-py3-none-any.whl", hash = "sha256:2d21d1cc5a017bd0559e36150c21c830ab1dc304dedd1b7ea85d20f45ef3edd7", size = 27789, upload-time = "2026-03-20T06:42:55.665Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "isodate" version = "0.7.2" @@ -2227,6 +2240,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8a/e0/39afe4bddbed6276c54e35e310aa345fbeb00f8890e96e7f48cdc2be9c66/pyroaring-1.1.0-cp314-cp314t-win_arm64.whl", hash = "sha256:99c42fe1449acfbf130da65e66b4d5b2726aba4497be359bae7672e38a15fc62", size = 234615, upload-time = "2026-04-24T21:29:08.751Z" }, ] +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0"