16 Testing pipelines in CI
Unit and integration tests for a small S3-to-Redshift pipeline, and a GitHub Actions workflow that runs them against oblako.
16.1 Before you start
oblako up s3
oblako up redshiftThe chapter’s code is a small project with its own pyproject.toml, a tests folder, and a workflow under .github/workflows. Create its environment and trust oblako’s Redshift certificate in it, as in Chapter 2:
uv sync
oblako trust --python .venv/bin/pythonThe tests create everything they use and remove it when they finish.
16.2 The pipeline
The pipeline under test reads one day of order events from S3, totals each customer’s orders and revenue, and loads the totals into a Redshift table. The events are JSON lines under a prefix per day, raw/orders/date=2026-10-01/, written by some other system in one or more files. A refund subtracts its amount from the customer’s revenue and does not count as an order.
import io
import json
from collections import defaultdict
from decimal import Decimal
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
import redshift_connector
def summarise(events):
"""Total each customer's orders; refunds subtract."""
totals = defaultdict(lambda: [0, Decimal("0")])
for event in events:
amount = Decimal(event["amount"])
if event["type"] == "refund":
amount = -amount
else:
totals[event["customer"]][0] += 1
totals[event["customer"]][1] += amount
return [
{"customer": c, "orders": n, "revenue": r}
for c, (n, r) in sorted(totals.items())
]
def read_events(bucket, day):
s3 = boto3.client("s3")
pages = s3.get_paginator("list_objects_v2").paginate(
Bucket=bucket, Prefix=f"raw/orders/date={day}/"
)
for page in pages:
for obj in page.get("Contents", []):
body = s3.get_object(Bucket=bucket, Key=obj["Key"])[
"Body"
]
for line in body.iter_lines():
yield json.loads(line)
def load(conn, bucket, day, rows, schema, role):
table = pa.Table.from_pylist(
[{"day": day, **row} for row in rows],
schema=pa.schema(
[
("day", pa.string()),
("customer", pa.string()),
("orders", pa.int64()),
("revenue", pa.decimal128(12, 2)),
]
),
)
buffer = io.BytesIO()
pq.write_table(table, buffer)
key = f"staging/daily_totals/date={day}/part-0.parquet"
boto3.client("s3").put_object(
Bucket=bucket, Key=key, Body=buffer.getvalue()
)
cur = conn.cursor()
cur.execute(
f"DELETE FROM {schema}.daily_totals WHERE day = %s",
(day,),
)
cur.execute(
f"COPY {schema}.daily_totals FROM 's3://{bucket}/{key}'"
f" IAM_ROLE '{role}' FORMAT AS PARQUET"
)
conn.commit()
def run(day, *, bucket, schema, role, connect):
rows = summarise(read_events(bucket, day))
with connect() as conn:
load(conn, bucket, day, rows, schema, role)
return len(rows)
def connect_redshift(host, port, database, user, password):
return lambda: redshift_connector.connect(
host=host,
port=port,
database=database,
user=user,
password=password,
ssl=True,
sslmode="verify-ca",
)The work is split so that each part can be tested on its own. summarise is a plain function of a list of events. read_events and load talk to S3 and Redshift. run joins them, and takes the bucket, the schema, the IAM role and a connection factory as arguments, so a test can point it at resources it created. load writes the totals to S3 as Parquet and loads them with COPY, as Chapter 7 did.
A daily job is often run twice for the same day, after a failure or when late events arrive, so load deletes the day’s rows before copying the new ones. The DELETE and the COPY are committed together: if the COPY fails, the connection closes without a commit and the day’s earlier totals remain.
The table is created once, outside the pipeline:
CREATE TABLE {schema}.daily_totals (
day varchar(10),
customer varchar(32),
orders bigint,
revenue decimal(12, 2)
) DISTKEY(customer) SORTKEY(day);16.3 Two kinds of tests
A unit test checks a function on its own, with its inputs written in the test. It needs no services and runs in milliseconds, so it can cover many cases: here, the arithmetic of orders and refunds. An integration test runs the code against the services it uses in production. It catches what a unit test cannot: a wrong S3 prefix, a column type that COPY rejects, SQL that Redshift does not accept, a load that is not safe to repeat. Without a local stack, integration tests either run against a shared AWS account, with its costs and the risk of tests interfering with each other, or replace the services with mocks that accept whatever the code sends. With oblako they run against real engines on the machine that runs the tests.
The project marks integration tests so that the two kinds can be run separately:
[project]
name = "daily-totals"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["boto3", "pyarrow", "redshift-connector"]
[dependency-groups]
dev = ["pytest"]
[tool.pytest.ini_options]
markers = ["integration: needs oblako (or AWS) to be running"]
# redshift_connector builds its SSL context in a deprecated way
filterwarnings = ["ignore::DeprecationWarning:redshift_connector"]16.4 Unit tests
from decimal import Decimal as D
from pipeline import summarise
def event(customer, type_, amount):
return {"customer": customer, "type": type_, "amount": amount}
def test_orders_are_counted_and_summed():
events = [
event("c1", "order", "10.50"),
event("c1", "order", "4.50"),
event("c2", "order", "7.00"),
]
assert summarise(events) == [
{"customer": "c1", "orders": 2, "revenue": D("15.00")},
{"customer": "c2", "orders": 1, "revenue": D("7.00")},
]
def test_refunds_are_subtracted():
events = [
event("c1", "order", "20.00"),
event("c1", "refund", "5.00"),
event("c3", "refund", "3.00"),
]
assert summarise(events) == [
{"customer": "c1", "orders": 1, "revenue": D("15.00")},
{"customer": "c3", "orders": 0, "revenue": D("-3.00")},
]
def test_no_events_no_rows():
assert summarise([]) == []uv run pytest -m "not integration" -vtests/test_summarise.py::test_orders_are_counted_and_summed PASSED
tests/test_summarise.py::test_refunds_are_subtracted PASSED
tests/test_summarise.py::test_no_events_no_rows PASSED
================ 3 passed, 2 deselected in 0.01s =================16.5 Integration tests
Integration tests need a bucket and a Redshift table, and each test run should have its own, so that two runs on the same machine, or a run that failed halfway, never see each other’s data. pytest fixtures provide them:
import os
import uuid
from pathlib import Path
import boto3
import pytest
from pipeline import connect_redshift
REDSHIFT = {
"host": os.environ.get("REDSHIFT_HOST", "localhost"),
"port": int(os.environ.get("REDSHIFT_PORT", "5439")),
"database": os.environ.get("REDSHIFT_DATABASE", "oblako"),
"user": os.environ.get("REDSHIFT_USER", "oblako"),
"password": os.environ.get("REDSHIFT_PASSWORD", "oblako"),
}
ROLE = os.environ.get(
"REDSHIFT_COPY_ROLE",
"arn:aws:iam::123456789012:role/redshift-copy",
)
RUN = uuid.uuid4().hex[:8] # one id per test session
@pytest.fixture(scope="session")
def connect():
return connect_redshift(**REDSHIFT)
@pytest.fixture(scope="session")
def bucket():
s3 = boto3.client("s3")
name = f"daily-totals-test-{RUN}"
s3.create_bucket(Bucket=name)
yield name
for page in s3.get_paginator("list_objects_v2").paginate(
Bucket=name
):
for obj in page.get("Contents", []):
s3.delete_object(Bucket=name, Key=obj["Key"])
s3.delete_bucket(Bucket=name)
@pytest.fixture(scope="session")
def schema(connect):
name = f"test_{RUN}"
ddl = Path(__file__).parent.parent / "schema.sql"
with connect() as conn:
cur = conn.cursor()
cur.execute(f"CREATE SCHEMA {name}")
cur.execute(ddl.read_text().format(schema=name))
conn.commit()
yield name
with connect() as conn:
conn.cursor().execute(f"DROP SCHEMA {name} CASCADE")
conn.commit()Each test session draws a random identifier and creates a bucket and a schema named after it, with the table inside the schema. The fixtures are session-scoped, so they are created once for all the integration tests, and the code after each yield removes them when the session ends, whether the tests passed or failed. The connection settings come from environment variables with oblako’s values as defaults, so the same tests can run against a Redshift cluster on AWS by setting the variables, as Chapter 18 does.
The tests write events to S3, run the pipeline and read the table back:
import json
from decimal import Decimal as D
import boto3
import pytest
from pipeline import run
from tests.conftest import ROLE
from tests.test_summarise import event
pytestmark = pytest.mark.integration
def put_events(bucket, day, part, events):
body = "\n".join(json.dumps(e) for e in events).encode()
key = f"raw/orders/date={day}/part-{part}.jsonl"
s3 = boto3.client("s3")
s3.put_object(Bucket=bucket, Key=key, Body=body)
def totals(connect, schema, day):
sql = (
"SELECT customer, orders, revenue"
f" FROM {schema}.daily_totals"
" WHERE day = %s ORDER BY customer"
)
with connect() as conn:
cur = conn.cursor()
cur.execute(sql, (day,))
return [tuple(row) for row in cur.fetchall()]
def test_loads_a_day_from_two_files(bucket, schema, connect):
day = "2026-10-01"
put_events(
bucket,
day,
0,
[
event("c1", "order", "10.00"),
event("c2", "order", "5.00"),
],
)
put_events(bucket, day, 1, [event("c1", "refund", "2.50")])
loaded = run(
day,
bucket=bucket,
schema=schema,
role=ROLE,
connect=connect,
)
assert loaded == 2
assert totals(connect, schema, day) == [
("c1", 1, D("7.50")),
("c2", 1, D("5.00")),
]
def test_rerunning_a_day_replaces_it(bucket, schema, connect):
day = "2026-10-02"
put_events(bucket, day, 0, [event("c1", "order", "3.00")])
for _ in range(2):
run(
day,
bucket=bucket,
schema=schema,
role=ROLE,
connect=connect,
)
assert totals(connect, schema, day) == [("c1", 1, D("3.00"))]
# the other day's rows are untouched
assert len(totals(connect, schema, "2026-10-01")) == 2export AWS_PROFILE=oblako
uv run pytest -m integration -vtests/test_pipeline.py::test_loads_a_day_from_two_files PASSED
tests/test_pipeline.py::test_rerunning_a_day_replaces_it PASSED
================ 2 passed, 3 deselected in 1.27s =================The first test checks that the pipeline reads every file under the day’s prefix and that a refund in one file reduces the revenue of an order in another. The second runs the same day twice. Without the DELETE in load, it fails: the second run appends a second copy of the day’s rows, and the assertion reports the extra row.
16.6 Running the tests in CI
GitHub Actions runs a workflow on a fresh virtual machine for each push and pull request. GitHub’s hosted Linux runners have Docker installed, which is all oblako needs, so the integration tests run there as they do on your machine:
name: tests
on: [push, pull_request]
jobs:
unit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: astral-sh/setup-uv@v6
- run: uv run pytest -m "not integration"
integration:
runs-on: ubuntu-latest
env:
AWS_PROFILE: oblako
steps:
- uses: actions/checkout@v4
- uses: astral-sh/setup-uv@v6
- name: Install oblako
run: >
uv tool install
git+https://github.com/almostly/oblako.git
- name: Start S3 and Redshift
run: |
oblako up s3
oblako up redshift
oblako configure
- name: Trust the Redshift certificate
run: |
uv sync
oblako trust --python .venv/bin/python
- name: Integration tests
run: uv run pytest -m integrationThe workflow has two jobs, which run in parallel (Figure 16.1). The unit job needs nothing but Python. The integration job installs oblako as a tool, starts S3 and Redshift, writes the oblako profile, trusts the Redshift certificate in the project’s environment and runs the integration tests with the profile selected for the whole job.
oblako up waits for each service it starts to accept requests, up to 120 seconds by default, and fails the step when a service is not ready in time; --timeout changes the limit. The first start on a runner downloads the service images, so the step takes longer there than on a machine that already has them. A test step that starts before the services are ready fails with connection errors that look like bugs in the code, so failing at the start step makes the cause clear.
The trust step must run after uv sync, because it patches the certificate bundle of the installed redshift-connector. Without it, every Redshift connection fails:
redshift_connector.error.InterfaceError: ('communication error',
SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED]
certificate verify failed: self-signed certificate (_ssl.c:1081)'))The install step takes oblako’s latest code. For a build that does not change under you, pin a commit by appending @ and the commit’s hash to the URL.
Which services to start
Start only the services the tests use. Each takes memory and adds to the start-up time, and a runner has less of both than most laptops. A pipeline that also reads from DynamoDB adds oblako up dynamodb; one that queries through the Redshift Data API adds oblako up redshift-data, which starts a process on the runner instead of a container.
16.7 Where oblako differs from AWS
Permissions are not checked. The tests pass whatever the IAM role and the credentials allow, so a missing permission shows up only on AWS.
The cluster is single-node unless the tests create a multi-node one (Chapter 13). Distribution styles and sort keys are accepted, and on a single node they do not change where rows are stored.
Speed is the runner’s. Timings in CI say nothing about the same load on Redshift.
16.8 Summary
Unit tests cover a pipeline’s logic with inputs written in the test; integration tests run the pipeline against S3 and Redshift and catch what only the services can reject. Fixtures that create a uniquely named bucket and schema for each test session, and remove them afterwards, keep runs from interfering with each other. In GitHub Actions, the integration job installs oblako, starts the services the tests need, writes the oblako profile and trusts the Redshift certificate. It then runs the tests with that profile selected, on the Docker that every hosted Linux runner has.