Data Pipeline Testing

A pipeline can finish “successfully” and still put wrong data in the wrong place, or fail halfway and tell nobody. So we test four things:

Automated checks are written in pytest, the same as backend tests.

5.1 Map the pipeline first

Before writing test cases, list every task in order, with what it reads and where it writes. This table is the basis for every test case on this page.

Example: daily eTax report pipeline

Put the pipeline name and its tasks in Kiwi as Components, one per task.

5.2 Each task lands correctly

For every task, check where the data landed and what it contains:

• Right place: the exact container, path, table, and partition from the map, including the date or run ID in the path.
• Nothing in the wrong place: no files in other folders, no rows in other partitions.
• Right shape: columns, field names, and types as expected. No missing or extra columns.
• Right amount: the row count matches the input, minus any rows the task is supposed to drop. Every difference is explained.
• Right values: a sample of rows checked against the expected transformation. Required fields not empty; no duplicates on the key.

“The task finished” is never a pass on its own. The data is what we test.

5.3 How to check the data

Five checks, from cheapest to most detailed. Use row count and uniqueness on every task; add the others when the data matters more.

Row count

Compare source and target, and explain every difference:

-- source
SELECT COUNT(*) FROM staging.etax_clean WHERE run_date = '2026-09-01';
-- target
SELECT COUNT(*) FROM report.etax_daily  WHERE run_date = '2026-09-01';

Expected: target = source - rows the task is meant to drop. If the task filters nothing, the tolerance is 0. Count per partition or per day, not only the whole table, because a correct total can still hide a day loaded twice and a day missing.

Checksum

A row count only proves how many rows arrived, not that the values inside them are right. A checksum hashes the columns and compares the hashes.

Per row (when there’s a unique key). This shows you which rows differ:

-- run on both sides, then compare by key
SELECT invoice_id,
       CONVERT(VARCHAR(64), HASHBYTES('SHA2_256',
         CONCAT_WS('|', invoice_no, CAST(total AS VARCHAR(30)), CONVERT(CHAR(10), invoice_date, 23))
       ), 2) AS row_hash
FROM   report.etax_daily
WHERE  run_date = '2026-09-01';

-- rows that differ or are missing on one side
SELECT COALESCE(s.invoice_id, t.invoice_id) AS invoice_id,
       s.row_hash AS source_hash, t.row_hash AS target_hash
FROM   source_hashes s
FULL OUTER JOIN target_hashes t ON s.invoice_id = t.invoice_id
WHERE  s.row_hash IS NULL OR t.row_hash IS NULL OR s.row_hash <> t.row_hash;

Expected: 0 rows.

All rows at once (no key, or a quick check). Sum the row hashes, so the order doesn’t matter:

SELECT SUM(CAST(CHECKSUM(invoice_no, total, invoice_date) AS BIGINT)) AS table_checksum
FROM   report.etax_daily WHERE run_date = '2026-09-01';

Both sides must return the same number. This tells you that something differs, not which row, so use the per-row version to find it.

In pytest with pandas:

import pandas as pd
 
def row_hashes(df, key, columns):
    normalized = df[columns].astype(str).apply(lambda c: c.str.strip())
    return pd.Series(pd.util.hash_pandas_object(normalized, index=False).values, index=df[key])
 
diff = row_hashes(source_df, "invoice_id", COLS).compare(row_hashes(target_df, "invoice_id", COLS))
assert diff.empty, f"{len(diff)} rows differ:\n{diff.head(20)}"

Normalize before hashing, or you’ll chase differences that aren’t real: trim spaces, treat NULL and empty the same way, fix the decimal places (1500000 vs 1500000.00), and use one date format and time zone.

Lineage

Lineage answers: which run produced this row, and where did it come from? Without it, a wrong row in the report can’t be traced back to a pipeline run.

Check both levels:

1. On the rows: every target row carries the run ID (and ideally the source file or batch).

   SELECT COUNT(*) FROM report.etax_daily
   WHERE  run_date = '2026-09-01' AND (run_id IS NULL OR run_id <> '<run id>');

   Expected: 0.

2. In the orchestrator log: a row per task per run, with status, start and end time, and rows read and written.

   SELECT task_name, status, rows_read, rows_written
   FROM   fact_orchestrator_03 WHERE run_id = '<run id>' ORDER BY start_time;

   Expected: one row per task, statuses matching what really happened, and rows_written matching the row count you measured yourself in the table.

ASSUMPTION: I don’t know what fact_orchestrator_03 actually stores. Confirm its columns with the data team and update the query above. If the pipeline has no lineage at all, that’s a gap worth raising: ask for a run_id column on target tables, because without it none of the checks above can be tied to a specific run.

Uniqueness and deduplication

With a unique key (single column or composite):

SELECT invoice_id, COUNT(*) AS n
FROM   report.etax_daily WHERE run_date = '2026-09-01'
GROUP  BY invoice_id HAVING COUNT(*) > 1;

Expected: 0 rows. In pandas: assert not df.duplicated(subset=["invoice_id"]).any().

Run this check twice: once after the normal run, and again after re-running the same pipeline (section 5.5). A pipeline that produces duplicates only on the second run is the most common data bug of all.

If there’s no unique key, agree one with the developer (often a business key plus the run date). Without a key you can’t prove uniqueness, and you can’t do the per-row checksum or the Excel lookup either.

Excel lookup

When the expected result is a manual Excel file, or someone checks the report by hand, compare by the unique key instead of reading line by line.

1. Export the target table to CSV and open both files in one workbook: expected and actual.

2. In expected, next to each row (key in column A, amount in column C):

   =XLOOKUP(A2, actual!$A:$A, actual!$C:$C, "NOT FOUND")

   VLOOKUP works too if XLOOKUP isn’t available: =IFERROR(VLOOKUP(A2, actual!$A:$C, 3, FALSE), "NOT FOUND")

3. Compare, allowing for rounding on amounts:

   =IF(D2="NOT FOUND", "MISSING", IF(ROUND(C2,2)=ROUND(D2,2), "OK", "DIFF"))

4. Count the problems: =COUNTIF(E:E,"DIFF") and =COUNTIF(E:E,"MISSING"). Both must be 0.

5. Check the other direction too, looking up actual keys in expected, to catch extra rows the pipeline invented.

The same comparison in pytest, for when it becomes a repeated check:

merged = expected_df.merge(actual_df, on="invoice_id", how="outer", indicator=True)
assert (merged["_merge"] == "both").all(), "rows missing on one side"
assert (merged["total_x"].round(2) == merged["total_y"].round(2)).all(), "amounts differ"

5.4 Success and failure are both handled correctly

When a task succeeds:

• Its status is recorded as succeeded, with the run ID.
• The next task starts, using this task’s output.
• Temporary files are cleaned up, if the design says so.

When a task fails:

• Its status is recorded as failed, with the run ID.
• The error is logged with a reason someone can act on (not just “error”).
• It doesn’t leave half-written data behind. Partial output is removed, or clearly marked as incomplete.
• It retries only as designed (e.g. 3 times), then stops.

To test failures, make each task fail on purpose in the test environment: give it a bad input file, remove its input, or point it at a stopped dependency.

5.5 One task fails → the whole pipeline fails

• The tasks after the failed task don’t run.
• The pipeline’s final status is Failed. Not “Succeeded”, not “Partially succeeded”.
• Nothing reaches the final destination (no rows in the report table, no published report for that run).
• After fixing the cause, re-running the pipeline gives the correct result, without duplicates.

Write one test case per task for this. With 4 tasks, the test set looks like:

5.6 A failure sends an alert

In the test environment, route alerts to a test channel you can check (a test Teams channel, test mailbox, or webhook), never the real on-call channel.

• An alert arrives for every failed run, within the agreed time (default: 5 minutes).
• It says enough to act on: pipeline name, run ID, which task failed, the error, and the time.
• One alert per failure, not one per retry.
• No alert on a successful run.

A failure nobody is told about counts as a failed test, even if the pipeline itself behaved correctly.

5.7 How a test looks

@pytest.mark.kiwi_case(33)
def test_DATA_ETAXREPORT_003_transform_fails_whole_pipeline_fails(pipeline, alerts, report_db):
    run = pipeline.trigger(run_date="2026-09-01", break_task="transform")
    pipeline.wait_until_finished(run, timeout=900)
 
    # 5.5: the whole pipeline fails, later tasks don't run
    assert pipeline.status(run) == "Failed"
    assert pipeline.task_status(run, "transform") == "Failed"
    assert pipeline.task_status(run, "load") == "NotStarted"
    assert pipeline.task_status(run, "publish") == "NotStarted"
 
    # 5.5: nothing reached the destination
    assert report_db.count("report.etax_daily", partition="2026-09-01") == 0
 
    # 5.6: one alert, with the details needed to act on it
    alert = alerts.wait_for(run_id=run.id, timeout=300)
    assert "transform" in alert.text and run.id in alert.text
    assert alerts.count(run_id=run.id) == 1

The fixtures (pipeline, alerts, report_db) depend on what runs the pipeline (e.g. Azure Data Factory, Airflow, or Azure Functions) and are written once per project. The checks stay the same.

5.8 Test data

• Use a small, fake dataset that includes the tricky cases: empty fields, duplicates, odd dates, very large values.
• Keep the dataset in the test repo, so every run uses the same input.
• Never copy production data with real personal or tax information into test.