From a026143dafbc85d84cabc610e90d85a2f303d3ef Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Wed, 2 Sep 2026 16:07:53 +0100 Subject: [PATCH 1/8] deps(elt-pipelines): Add pytest dev dependency --- elt-pipelines/pyproject.toml | 1 + elt-pipelines/uv.lock | 31 ++++++++++++++++++++++++++++++- 2 files changed, 31 insertions(+), 1 deletion(-) 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 3e604567..b273e7bf 100644 --- a/elt-pipelines/uv.lock +++ b/elt-pipelines/uv.lock @@ -783,6 +783,7 @@ statusdisplay = [ [package.dev-dependencies] dev = [ { name = "prek" }, + { name = "pytest" }, ] [package.metadata] @@ -804,7 +805,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" @@ -1098,6 +1102,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" @@ -2235,6 +2248,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" From 3313748c70b79e5780de2cd227eb9f932fbc446d Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Thu, 3 Sep 2026 12:26:13 +0100 Subject: [PATCH 2/8] Move IPv6 disabling ref #443 Planned test harness will directly call run_ingest, so we need to move this out of cli to ensure parity. I believe all uses of requests happen as part of run_ingest, so this doesn't effect behaviour. --- elt-common/src/elt_common/cli.py | 6 ------ elt-common/src/elt_common/ingest.py | 7 +++++++ 2 files changed, 7 insertions(+), 6 deletions(-) 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 99c021e7..d675610c 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 pyiceberg.exceptions import NoSuchTableError from elt_common.extract import ( @@ -26,6 +27,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.""" From 4e752c1a50c0b84bfb8d2564820894aea59fe2f5 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Thu, 3 Sep 2026 12:26:07 +0100 Subject: [PATCH 3/8] feat(elt-common): Pipeline testing harness and examples ref #443 --- elt-common/src/elt_common/pipeline.py | 4 +- .../src/elt_common/testing/pipelines.py | 124 ++++++++++++++++++ .../src/elt_common/testing/sqlcatalog.py | 12 +- .../statusdisplay/test_statusdisplay.py | 49 +++++++ .../ingest/fase/proposal/test_proposal.py | 34 +++++ 5 files changed, 216 insertions(+), 7 deletions(-) create mode 100644 elt-common/src/elt_common/testing/pipelines.py create mode 100644 elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py create mode 100644 elt-pipelines/fase/ingest/fase/proposal/test_proposal.py 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/pipelines.py b/elt-common/src/elt_common/testing/pipelines.py new file mode 100644 index 00000000..910900ad --- /dev/null +++ b/elt-common/src/elt_common/testing/pipelines.py @@ -0,0 +1,124 @@ +"""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 collections.abc import Callable +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) + + def do_something(self, something: Callable[[Catalog], None]): + something(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) + snapshot = t.current_snapshot() + assert snapshot is not None + assert snapshot.summary is not None + return int(snapshot.summary.additional_properties["total-records"]) + + +@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() + + try: + yield AssertableCatalog(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..08974c81 100644 --- a/elt-common/src/elt_common/testing/sqlcatalog.py +++ b/elt-common/src/elt_common/testing/sqlcatalog.py @@ -1,16 +1,18 @@ +from pathlib import Path from tempfile import TemporaryDirectory -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, directory: Path | None = None): 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 = directory if directory else TemporaryDirectory().name + 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/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..cc96623d --- /dev/null +++ b/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py @@ -0,0 +1,49 @@ +from pyiceberg.catalog import Catalog + +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 + ], + ) + test_catalog.assert_has_n_rows(_table_id, 153) + + test_catalog.clean_catalog() + + def check_table_cleaned(catalog): + assert not catalog.table_exists(_table_id), "Table should have been cleaned up" + + test_catalog.do_something(check_table_cleaned) + + +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) + + def check_single_ns_and_table(catalog: Catalog): + nss = catalog.list_namespaces() + assert len(nss) == 1, "There should be a single namespace" + catalog.namespace_exists(_namespace) + assert len(catalog.list_tables(_namespace)) == 1, ( + "There should be a single table" + ) + + test_catalog.do_something(check_single_ns_and_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..982fce8f --- /dev/null +++ b/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py @@ -0,0 +1,34 @@ +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) + test_catalog.clean_catalog() + + +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) + test_catalog.clean_catalog() + + +@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) + test_catalog.clean_catalog() From 92f13be97280d22de05de5c4514a56e0e78631e4 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Mon, 7 Sep 2026 16:09:53 +0100 Subject: [PATCH 4/8] docs(elt-pipelines): Add testing information --- elt-pipelines/README.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/elt-pipelines/README.md b/elt-pipelines/README.md index ddccbf5e..26b6dce5 100644 --- a/elt-pipelines/README.md +++ b/elt-pipelines/README.md @@ -93,6 +93,15 @@ 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. + ## Directory structure The project uses the following directory structure: From a8eba9a1b5f2b24b73482b07534c7fd47c7fdeb5 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Mon, 7 Sep 2026 16:33:44 +0100 Subject: [PATCH 5/8] fix(elt-common): Make existing e2e tests work with SqlCatalogWarehouse change --- elt-common/src/elt_common/testing/fixtures.py | 14 ++++++++------ elt-common/src/elt_common/testing/sqlcatalog.py | 5 ++--- 2 files changed, 10 insertions(+), 9 deletions(-) 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/sqlcatalog.py b/elt-common/src/elt_common/testing/sqlcatalog.py index 08974c81..19a5475b 100644 --- a/elt-common/src/elt_common/testing/sqlcatalog.py +++ b/elt-common/src/elt_common/testing/sqlcatalog.py @@ -1,5 +1,4 @@ from pathlib import Path -from tempfile import TemporaryDirectory from pyiceberg.catalog import Catalog as PyIcebergCatalog from pyiceberg.catalog import load_catalog @@ -8,9 +7,9 @@ class SqlCatalogWarehouse: - def __init__(self, warehouse_name: str, directory: Path | None = None): + def __init__(self, warehouse_name: str, workdir: Path): self.name = warehouse_name - self.workdir = directory if directory else TemporaryDirectory().name + self.workdir = workdir self.uri = f"sqlite:///{self.workdir}/{warehouse_name}.db" self.warehouse_path = f"file://{self.workdir}/{warehouse_name}" From 0e922f630853862507343ad524773a4d89314ef5 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Tue, 8 Sep 2026 11:55:40 +0100 Subject: [PATCH 6/8] feat(elt-common): Automatically clean catalog after tests --- elt-common/src/elt_common/testing/pipelines.py | 8 ++++++-- elt-pipelines/fase/ingest/fase/proposal/test_proposal.py | 3 --- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/elt-common/src/elt_common/testing/pipelines.py b/elt-common/src/elt_common/testing/pipelines.py index 910900ad..ee94ce91 100644 --- a/elt-common/src/elt_common/testing/pipelines.py +++ b/elt-common/src/elt_common/testing/pipelines.py @@ -97,11 +97,15 @@ def test_catalog(sql_warehouses, request, monkeypatch): monkeypatch.setenv("PYICEBERG_CATALOG__DEFAULT__WAREHOUSE", test_warehouse_name) catalog = warehouse.connect() + ret = AssertableCatalog(catalog) try: - yield AssertableCatalog(catalog) + yield ret finally: - catalog.close() + try: + ret.clean_catalog() + finally: + catalog.close() @pytest.fixture diff --git a/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py b/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py index 982fce8f..9d057ffb 100644 --- a/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py +++ b/elt-pipelines/fase/ingest/fase/proposal/test_proposal.py @@ -17,18 +17,15 @@ def test_expected_columns_created(test_catalog: AssertableCatalog, run_test_inge "end_review", ] test_catalog.assert_has_columns((_ns, "call"), expected_columns) - test_catalog.clean_catalog() 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) - test_catalog.clean_catalog() @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) - test_catalog.clean_catalog() From 49e78ee8cb7a0c3c1ea066779aa836d9e6723ac1 Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Tue, 8 Sep 2026 11:58:47 +0100 Subject: [PATCH 7/8] docs(elt-pipelines): Note about configuration for tests --- elt-pipelines/README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/elt-pipelines/README.md b/elt-pipelines/README.md index 26b6dce5..7493bcb0 100644 --- a/elt-pipelines/README.md +++ b/elt-pipelines/README.md @@ -102,6 +102,8 @@ Tests are invoked with `pytest --disable-plugin-autoload -p elt_common.testing.p 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: From a7eaa5efe17528304d0c771136c5c3ef0b348f9d Mon Sep 17 00:00:00 2001 From: Will Taylor Date: Tue, 8 Sep 2026 16:16:38 +0100 Subject: [PATCH 8/8] Cleanup AssertableCatalog API ref https://github.com/ISISNeutronMuon/analytics-data-platform/pull/464#discussion_r3958552273 Will likely be expanded over time, but this kind of pattern seems like a good place to start --- .../src/elt_common/testing/pipelines.py | 11 +++----- .../statusdisplay/test_statusdisplay.py | 25 +++++++------------ 2 files changed, 13 insertions(+), 23 deletions(-) diff --git a/elt-common/src/elt_common/testing/pipelines.py b/elt-common/src/elt_common/testing/pipelines.py index ee94ce91..589c086f 100644 --- a/elt-common/src/elt_common/testing/pipelines.py +++ b/elt-common/src/elt_common/testing/pipelines.py @@ -19,7 +19,6 @@ """ import logging -from collections.abc import Callable from pathlib import Path import pytest @@ -47,8 +46,9 @@ def clean_catalog(self): self._catalog.drop_namespace(ns) - def do_something(self, something: Callable[[Catalog], None]): - something(self._catalog) + @property + def catalog(self): + return self._catalog def assert_has_exact_tables(self, namespace: str, tables: list[str]): actual = self._catalog.list_tables(namespace) @@ -66,10 +66,7 @@ def assert_has_n_rows(self, table_id: tuple[str, ...], n: int): def get_num_rows(self, table_id: tuple[str, ...]): assert self._catalog.table_exists(table_id) t = self._catalog.load_table(table_id) - snapshot = t.current_snapshot() - assert snapshot is not None - assert snapshot.summary is not None - return int(snapshot.summary.additional_properties["total-records"]) + return t.scan().count() @pytest.fixture(scope="session") diff --git a/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py b/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py index cc96623d..473153ee 100644 --- a/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py +++ b/elt-pipelines/facility_ops/ingest/accelerator/statusdisplay/test_statusdisplay.py @@ -1,5 +1,3 @@ -from pyiceberg.catalog import Catalog - from elt_common.testing.pipelines import AssertableCatalog _namespace = "accelerator_statusdisplay" @@ -19,14 +17,14 @@ def test_expected_columns_created(test_catalog: AssertableCatalog, run_test_inge "phases.element", # phases is a list type ], ) - test_catalog.assert_has_n_rows(_table_id, 153) + 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() - def check_table_cleaned(catalog): - assert not catalog.table_exists(_table_id), "Table should have been cleaned up" - - test_catalog.do_something(check_table_cleaned) + 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): @@ -38,12 +36,7 @@ def test_multiple_runs(test_catalog: AssertableCatalog, run_test_ingest): # Check runs are overwriting, not appending test_catalog.assert_has_n_rows(_table_id, n) - def check_single_ns_and_table(catalog: Catalog): - nss = catalog.list_namespaces() - assert len(nss) == 1, "There should be a single namespace" - catalog.namespace_exists(_namespace) - assert len(catalog.list_tables(_namespace)) == 1, ( - "There should be a single table" - ) - - test_catalog.do_something(check_single_ns_and_table) + 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"