Report test results to an external system¶

You’ve written a test action. Its pass/fail result lands in the pipeline’s event log — but that’s inside Databricks. When your team tracks test outcomes somewhere else — a Delta audit table, Azure DevOps Test Plans, a CI dashboard — you want each result published there automatically, on every run.

You could hand-write that: an @dp.on_event_hook that parses flow_progress events, extracts the data-quality metrics, correlates each one to the right test, waits for the pipeline’s terminal state, and calls your publishing code exactly once. Or you declare a test_reporting provider in lhp.yaml, tag each test with a test_id, and let Lakehouse Plumber generate the whole hook. That is the idea on every page: declare your ETL, don’t hand-write it.

Before you start¶

You need a project with at least one test action and a Python function that knows how to publish results to your system. This guide publishes to a Delta audit table; the same shape works for any sink.

Important

Use on_violation: warn on any test whose result you report. A fail aborts the flow before its metrics are recorded, so the reporting hook never sees them — warn records the outcome and lets the run continue to terminal state, where the hook publishes.

Write the provider function¶

The provider is an ordinary Python module exporting one function with a fixed signature: publish_results(results, config, context, spark). Lakehouse Plumber calls it once, at pipeline terminal state, with the accumulated results.

providers/audit_delta.py¶

"""Test-reporting provider: write every test result to a Delta audit table.

Lakehouse Plumber calls ``publish_results`` once, at pipeline terminal state,
with the accumulated test results. The signature is fixed:

    publish_results(results, config, context, spark) -> dict

- ``results``  list of per-test dicts (test_id, status, passed/failed_records, ...)
- ``config``   the parsed ``config_file`` mapping (here: the target table)
- ``context``  pipeline_id, update_id, pipeline_name, terminal_state
- ``spark``    an active SparkSession

Return a ``{"published": N, "failed": M}`` dict so LHP can log the outcome.
Swap the body for an Azure DevOps Test Plans API call, a webhook, or any sink.
"""

from datetime import datetime, timezone


def publish_results(results, config, context, spark):
    audit_table = config["audit_table"]

    rows = [
        {
            "pipeline_name": context.get("pipeline_name", ""),
            "update_id": context.get("update_id", ""),
            "test_id": r["test_id"],
            "status": r["status"],
            "failed_records": r["failed_records"],
            "reported_at": datetime.now(timezone.utc).isoformat(),
        }
        for r in results
    ]

    (
        spark.createDataFrame(rows)
        .write.format("delta")
        .mode("append")
        .saveAsTable(audit_table)
    )

    return {"published": len(rows), "failed": 0}

Swap the body for an Azure DevOps API call or a webhook — the contract is the same. Return a {"published": N, "failed": M} dict so LHP can log the outcome.

Declare the provider in `lhp.yaml`¶

Point test_reporting at the module, the function, and (optionally) a config file passed to the provider verbatim:

lhp.yaml¶

# A test_reporting provider forwards every test outcome to an external system.
# module_path points at a Python module exporting the provider function;
# function_name is the callable LHP invokes at pipeline terminal state;
# config_file (optional) is a YAML file passed to the provider verbatim.
test_reporting:
  module_path: providers/audit_delta.py
  function_name: publish_results
  config_file: providers/audit_config.yaml

providers/audit_config.yaml¶

# Passed verbatim to the provider function as its `config` argument.
# ${...} tokens resolve per environment, like everywhere else in LHP.
audit_table: "${catalog}.${bronze_schema}.test_audit_log"

Tag the test with a `test_id`¶

The test_id correlates a test action with the entry your provider publishes. Add it to every test you report:

pipelines/orders_dq.yaml¶

# A test flowgroup whose result is forwarded to an external system.
# The test_id correlates this test with the entry the provider publishes.
# on_violation: warn keeps the run alive so the metrics reach the hook
# (a hard fail aborts the flow before its result is recorded).
pipeline: bronze_dq
flowgroup: orders_dq

actions:
  - name: orders_id_unique
    type: test
    test_type: uniqueness
    source: "${catalog}.${bronze_schema}.orders"
    columns: [order_id]
    on_violation: warn
    test_id: ORDERS-UNIQUE-001
    description: "Every bronze order has a unique order_id"

Generate with tests included¶

Test actions are opt-in, so pass --include-tests:

$ lhp generate --env dev --include-tests
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_dq  1 file
✓ generate (0.35s)
✓ format (0.02s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s

Read what Lakehouse Plumber wrote¶

Alongside the test table, LHP wrote one _test_reporting_hook.py per pipeline and copied your provider into test_reporting_providers/:

generated/dev/bronze_dq/
├── orders_dq.py                        # the test table
├── _test_reporting_hook.py             # the generated event hook
└── test_reporting_providers/
    ├── __init__.py
    └── audit_delta.py                  # your provider, copied in

The hook imports your function, maps each test table to its test_id, embeds the provider config, and accumulates results until the pipeline terminates:

generated/dev/bronze_dq/_test_reporting_hook.py¶

# ──────────────────────────────────────────────────────────────────────────────
# Test Reporting Event Hook — bronze_dq
# Generated by Lakehouse Plumber — DO NOT EDIT
# ──────────────────────────────────────────────────────────────────────────────

from pyspark import pipelines as dp
from datetime import datetime, timezone

from test_reporting_providers.audit_delta import publish_results

# Mapping: unqualified table name → external test ID
_TEST_ID_MAP = {"tmp_test_orders_id_unique": "ORDERS-UNIQUE-001"}

# Provider configuration (from config_file or empty dict)
_PROVIDER_CONFIG = {"audit_table": "${catalog}.${bronze_schema}.test_audit_log"}

# Module-level state — safe because hooks run synchronously on the driver
_collected_results = []
_published = False

_TERMINAL_STATES = frozenset({"STOPPING", "FAILED", "CANCELED", "COMPLETED"})

You wrote a provider function and two lines of config. Lakehouse Plumber wrote the event plumbing — the flow_progress parsing, the terminal-state guard, the single-publish latch — that is tedious to get right by hand.

Note

${...} tokens in the config file (like audit_table above) are passed through to the provider unresolved — the hook receives the literal ${catalog}.${bronze_schema}.test_audit_log. Resolve them in your provider, or hard-code fully qualified names in the config file.

What’s next¶

Report from many pipelines — the test_reporting block is project-wide, so every pipeline with a tagged test gets its own hook. One provider, all pipelines.
Reference — every test_reporting field is in the test action reference.