Integration and Contract Testing

Testing how parts work together: databases in tests, test containers, API tests, consumer-driven contract tests, test data management and keeping integration tests fast.

What integration tests add

A unit test replaces a function's neighbors with test doubles, so it runs in milliseconds and points at one function. The price is that the neighbors are imagined: a fake repository accepts any query, and a stub for meter-api returns exactly the JSON the developer expected. An integration test puts the real neighbor back and checks the interface between the two (the pipeline page shows how a pipeline starts the database for it). ISTQB (the International Software Testing Qualifications Board) distinguishes component integration testing, between parts of one system, from system integration testing, between whole systems such as the portal and meter-api; see Test Levels and Test Types.

Integration tests catch the bugs that live in the seams:

  • an SQL (Structured Query Language) column that was renamed, or a query that only works in one database engine;
  • a field that is a string in the real API and a number in the stub;
  • time zones, character encodings and decimal separators that change on the way through;
  • behavior under failure: what happens when the database or the other service is slow.

The last point is not theoretical at Kestrel. In INC-22047 on 6 October 2026, meter-api 2.15.0 did not return a database connection to its pool when a write timed out (see the postmortem). No unit test could have seen that: the connection pool, the timeout and the database only meet in an integration test, and only if that test makes a write slow on purpose and then checks that the connection came back.

Databases in tests

The reading-validator (change CHG-0761) stores readings through a small repository class. The test below runs that class against a real SQL engine: SQLite, in memory, created fresh for every test by a pytest fixture. A small builder function makes test readings with sensible defaults, so each test only states what matters to it. 

import sqlite3
from decimal import Decimal

import pytest

SCHEMA = """
CREATE TABLE readings (
    meter_id       TEXT NOT NULL,
    interval_start TEXT NOT NULL,
    kwh            TEXT NOT NULL,
    PRIMARY KEY (meter_id, interval_start)
)
"""


class ReadingRepository:
    def __init__(self, connection):
        self.connection = connection

    def add(self, meter_id, interval_start, kwh):
        self.connection.execute(
            "INSERT INTO readings VALUES (?, ?, ?)",
            (meter_id, interval_start, str(kwh)),
        )

    def total_for_day(self, meter_id, day):
        rows = self.connection.execute(
            "SELECT kwh FROM readings WHERE meter_id = ? AND interval_start LIKE ?",
            (meter_id, day + "T%"),
        )
        return sum((Decimal(kwh) for (kwh,) in rows), Decimal(0))


def reading(**overrides):
    """Test data builder: a valid reading, changed only where a test says so."""
    values = {"meter_id": "M-30007", "interval_start": "2026-10-05T23:15",
              "kwh": Decimal("1.500")}
    values.update(overrides)
    return values


@pytest.fixture
def repo():
    connection = sqlite3.connect(":memory:")  # a new, empty database per test
    connection.execute(SCHEMA)
    yield ReadingRepository(connection)
    connection.close()


def test_total_for_one_day(repo):
    repo.add(**reading())
    repo.add(**reading(interval_start="2026-10-05T23:30", kwh=Decimal("0.750")))
    repo.add(**reading(interval_start="2026-10-06T00:00"))  # next day
    assert repo.total_for_day("M-30007", "2026-10-05") == Decimal("2.250")


def test_other_meters_do_not_count(repo):
    repo.add(**reading(meter_id="M-30002"))
    assert repo.total_for_day("M-30007", "2026-10-05") == 0


def test_the_same_quarter_hour_cannot_be_stored_twice(repo):
    repo.add(**reading())
    with pytest.raises(sqlite3.IntegrityError, match="UNIQUE"):
        repo.add(**reading(kwh=Decimal("9.999")))
============================= test session starts ==============================
collecting ... collected 3 items

test_example.py::test_total_for_one_day PASSED                           [ 33%]
test_example.py::test_other_meters_do_not_count PASSED                   [ 66%]
test_example.py::test_the_same_quarter_hour_cannot_be_stored_twice PASSED [100%]

============================== 3 passed in 0.36s ===============================

These are real integration tests: the SQL is parsed and executed, the primary key is enforced, and a typo in a column name fails. The second test would pass on an empty database too, which is the point of the fixture: each test gets its own database, so no test can see what another test stored.

Test isolation

Test isolation means that a test's result does not depend on any other test: not on the order they run in, not on data another test left behind, not on whether another test ran at all. Without it, a suite that passes in full fails when you run one test, or fails only on Tuesdays when the order changes. The common ways to get isolation with a database:

  • A fresh database per test, as above. Trivial with SQLite in memory; too slow for a real database server.
  • A transaction per test, rolled back at the end. The fixture opens a transaction, the test runs inside it, and the rollback removes everything. Fast, but code under test that commits on its own breaks out of it.
  • Unique data per test, such as a meter ID that contains the test's name, when tests share one database and run in parallel.

SQLite is not PostgreSQL

Kestrel's real readings live in PostgreSQL on db01. SQLite is a different engine with a different idea of types. By default it accepts text in a numeric column without complaint: 

import sqlite3

connection = sqlite3.connect(":memory:")
connection.execute("CREATE TABLE readings (meter_id TEXT, kwh REAL NOT NULL)")
connection.execute("INSERT INTO readings VALUES ('M-30007', 'one and a half')")
print(connection.execute("SELECT kwh, typeof(kwh) FROM readings").fetchall())
[('one and a half', 'text')]

PostgreSQL rejects the same statement:

ERROR:  invalid input syntax for type real: "one and a half"
LINE 1: INSERT INTO readings VALUES ('M-30007', 'one and a half')
                                                ^

Other differences bite more quietly: time zone handling in timestamptz columns, NUMERIC precision, INSERT … ON CONFLICT variants, locking, case sensitivity of LIKE, and every PostgreSQL-only function. A test suite that passes on SQLite says little about those. SQLite in memory is fine for logic that uses plain SQL; for the repository that runs against db01, test against PostgreSQL.

Real databases with Testcontainers

The practical way to get a real PostgreSQL in a test run is a container: started when the tests begin, thrown away when they end, the same version as production, and nothing to clean up. Testcontainers is a family of libraries (for Java, Python, .NET, Go, Node.js and more) that starts such throwaway containers from test code and tells the tests how to connect. It is open source; AtomicJar, the company behind it, has been part of Docker since December 2023.

Ravi Kumar has proposed ADR-0010, "Test the reading repository against PostgreSQL in containers, not SQLite". A session-wide fixture would look like this with testcontainers-python 4.15 (the current version in October 2026). It needs Docker on the machine or CI (continuous integration) runner, so it is not run on this page: 

import psycopg2
import pytest
from testcontainers.community.postgres import PostgresContainer

SCHEMA = open("db/schema.sql").read()  # the same schema file production uses


@pytest.fixture(scope="session")
def postgres_url():
    # One container for the whole test session: starting PostgreSQL takes seconds.
    with PostgresContainer("postgres:16", driver=None) as postgres:
        yield postgres.get_connection_url()


@pytest.fixture
def connection(postgres_url):
    connection = psycopg2.connect(postgres_url)
    with connection.cursor() as cursor:
        cursor.execute(SCHEMA)
    yield connection
    connection.rollback()  # isolation: undo everything this test did
    connection.close()

Testing against a stand-in service

The reading-validator also calls meter-api: GET /v1/meters/{meter_id}/readings. Calling the real meter-api from tests would make them slow, dependent on another team's deployments, and unable to test failures on demand. The usual answer is service virtualization: replacing a service the code depends on with a stand-in that speaks the same protocol over the network and can be told to behave well, badly or slowly. A simple version is a mock server; commercial tools record and replay real traffic and simulate whole landscapes of systems.

Unlike a mock object, the stand-in is a real HTTP server, so the client's real HTTP code runs: URL building, headers, JSON parsing, timeouts. In the test below a fixture starts a fake meter-api on a free local port; one meter answers normally, one does not exist and one is too slow: 

import json
import threading
import time
import urllib.error
import urllib.request
from decimal import Decimal
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import pytest


class MeterNotFound(Exception):
    pass


class MeterApiUnavailable(Exception):
    pass


def fetch_readings(base_url, meter_id, timeout=0.5):
    """The reading-validator's client for meter-api (the code under test)."""
    url = f"{base_url}/v1/meters/{meter_id}/readings"
    try:
        with urllib.request.urlopen(url, timeout=timeout) as response:
            body = json.load(response)
    except urllib.error.HTTPError as error:
        if error.code == 404:
            raise MeterNotFound(meter_id) from error
        raise MeterApiUnavailable(f"HTTP {error.code}") from error
    except (urllib.error.URLError, TimeoutError) as error:
        raise MeterApiUnavailable("no answer in time") from error
    return [(r["intervalStart"], Decimal(str(r["kwh"]))) for r in body["readings"]]


class FakeMeterApi(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass

    def do_GET(self):
        if self.path == "/v1/meters/M-30009/readings":
            time.sleep(1.0)  # far slower than the client's timeout
        if self.path in ("/v1/meters/M-30007/readings", "/v1/meters/M-30009/readings"):
            body = {"meterId": "M-30007", "readings": [
                {"intervalStart": "2026-10-05T23:15", "kwh": 1.5, "quality": "measured"}]}
            status = 200
        else:
            body, status = {"title": "Not found"}, 404
        data = json.dumps(body).encode()
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)
        except (BrokenPipeError, ConnectionResetError):
            pass  # the client already gave up (the slow meter)


@pytest.fixture(scope="module")
def meter_api():
    server = ThreadingHTTPServer(("127.0.0.1", 0), FakeMeterApi)  # port 0: any free port
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{server.server_port}"
    server.shutdown()


def test_readings_are_parsed_as_decimals(meter_api):
    assert fetch_readings(meter_api, "M-30007") == [("2026-10-05T23:15", Decimal("1.5"))]


def test_unknown_meter(meter_api):
    with pytest.raises(MeterNotFound):
        fetch_readings(meter_api, "M-39999")


def test_slow_meter_api_gives_a_clear_error(meter_api):
    with pytest.raises(MeterApiUnavailable, match="no answer in time"):
        fetch_readings(meter_api, "M-30009")
============================= test session starts ==============================
collecting ... collected 3 items

test_example.py::test_readings_are_parsed_as_decimals PASSED             [ 33%]
test_example.py::test_unknown_meter PASSED                               [ 66%]
test_example.py::test_slow_meter_api_gives_a_clear_error PASSED          [100%]

============================== 3 passed in 1.27s ===============================

The third test is a failure test of the same kind that INC-22047 called for, on the client side: it proves that the reading-validator gives up after half a second with a clear error, instead of hanging. The test INC-22047 itself needed belongs in meter-api's suite: make a write time out on purpose and check that the pool gets its connection back. The weakness of any stand-in is the same as that of a stub: it is only as right as the person who wrote it. If meter-api renames intervalStart next month, this fake will not notice. That is what contract tests are for.

Contract testing with Pact

Contract testing checks the agreement between a consumer and a provider without running them together, and a consumer-driven contract lets each consumer state which parts of the provider's responses it relies on. The architecture page shows the idea with a hand-written checker. In practice most teams use a tool for it.

Pact is the most widely used open-source tool for consumer-driven contract testing, with libraries for many languages (pact-python 3.4.1 as of October 2026; the current specification is version 4). Its documentation describes it as "a code-first consumer-driven contract testing tool": the contract is not written by hand but generated while the consumer's tests run. The flow:

The Pact flow. On the consumer side, the reading-validator's tests run against a Pact mock server and write a pact file, the contract. The pact file is published to a Pact Broker. On the provider side, meter-api's pipeline fetches the pact and replays every recorded request against the real meter-api, comparing the responses: provider verification. The results go back to the broker, and before a release the can-i-deploy check asks the broker whether the versions are compatible. consumer pipeline provider pipeline reading-validator tests vs Pact mock pact file (JSON) Pact Broker contracts, results provider verification replay against meter-api meter-api real code, test data verification result can-i-deploy?
Neither side needs the other running: the pact file carries the consumer's expectations to the provider's pipeline.

The consumer side

In the reading-validator's tests, Pact replaces the stand-in server of the previous section. The test describes the interaction it expects, Pact serves exactly that from a mock server, the client code runs against it, and at the end Pact writes the interaction to a pact file. Matchers say what kind of value is expected rather than the exact value. pact-python 3 is a complete rework of version 2; the code below uses the version 3 interface (not run here: Pact is not installed on the machine that checks this site): 

from decimal import Decimal
from pathlib import Path

import pytest
from pact import Pact, match

from validator.meter_api import fetch_readings


@pytest.fixture
def pact():
    pact = Pact("reading-validator", "meter-api").with_specification("V4")
    yield pact
    pact.write_file(Path(__file__).parent / "pacts")


def test_get_readings(pact):
    body = {
        "meterId": match.str("M-30007"),
        "readings": match.each_like(
            {"intervalStart": match.str("2026-10-05T23:15"), "kwh": match.number(1.5)}
        ),
    }
    (
        pact.upon_receiving("a request for the readings of a meter")
        .given("meter M-30007 has readings")
        .with_request("GET", "/v1/meters/M-30007/readings")
        .will_respond_with(200)
        .with_body(body, content_type="application/json")
    )
    with pact.serve() as server:
        readings = fetch_readings(str(server.url), "M-30007")
    assert readings == [("2026-10-05T23:15", Decimal("1.5"))]

Note what is not in the contract: the quality field. The reading-validator does not use it, so meter-api may change or remove it freely. A contract lists what the consumer needs, not everything the provider sends; that is what keeps the provider free to evolve (compare the tolerant reader). The generated pact file, shortened: 

{
  "consumer": { "name": "reading-validator" },
  "provider": { "name": "meter-api" },
  "interactions": [
    {
      "type": "Synchronous/HTTP",
      "description": "a request for the readings of a meter",
      "providerStates": [{ "name": "meter M-30007 has readings" }],
      "request": { "method": "GET", "path": "/v1/meters/M-30007/readings" },
      "response": {
        "status": 200,
        "body": {
          "contentType": "application/json",
          "content": {
            "meterId": "M-30007",
            "readings": [{ "intervalStart": "2026-10-05T23:15", "kwh": 1.5 }]
          }
        }
      }
    }
  ],
  "metadata": { "pactSpecification": { "version": "4.0" } }
}

(The real file also contains the matching rules, which say that kwh may be any number and readings any non-empty list of such objects.)

Provider verification

Provider verification is the provider's half of contract testing: the provider's pipeline takes every contract its consumers published, replays each recorded request against the real provider, and checks the responses against the expectations. Before each interaction, a provider state handler puts the matching test data in place ("meter M-30007 has readings"). In meter-api's pipeline: 

from pact import Verifier


def test_meter_api_honors_its_contracts(meter_api_url, set_up_state):
    verifier = (
        Verifier("meter-api")
        .add_transport(url=meter_api_url)        # a meter-api started for this test
        .add_source("./pacts/")                  # or .broker_source(...) for a Pact Broker
        .state_handler(set_up_state, teardown=True)
    )
    verifier.verify()

Suppose a developer renames intervalStart to start in meter-api. Its own unit tests are updated along with the code and pass; the verification fails, before release, and names the consumer that would have broken: 

Verifying a pact between reading-validator and meter-api

  a request for the readings of a meter
     Given meter M-30007 has readings
    returns a response which
      has status code 200 (OK)
      includes headers
        "Content-Type" with value "application/json" (OK)
      has a matching body (FAILED)

Failures:

1) Verifying a pact between reading-validator and meter-api Given meter M-30007 has readings - a request for the readings of a meter
    1.1) has a matching body
           $.readings[0] -> Actual map is missing the following keys: intervalStart

There were 1 pact failures

With a Pact Broker (open source, or the hosted PactFlow service from SmartBear, now sold as Swagger Contract Testing) the two pipelines never exchange files by hand: the consumer publishes its pacts, the provider fetches them and publishes its verification results, and a can-i-deploy check before each release asks the broker whether this version of meter-api has been verified against the versions of its consumers that are in production.

What contract tests are not

  • Not functional tests of the provider. A pact says "this request gets a response of this shape", not "the total for this meter is 2.25 kWh". meter-api's own tests check its business rules; a contract full of exact values breaks on every data change.
  • Not a replacement for one end-to-end check. Contracts prove that each pair of services agrees; a few end-to-end tests still prove that the deployed whole is wired together.
  • Not free for public APIs. Consumer-driven contracts need consumers who write tests and share them. For an API with unknown outside consumers, a published OpenAPI description and schema checks fit better (some tools, such as PactFlow's bi-directional contract testing, compare consumer pacts with an OpenAPI description).

Test data management

Integration tests need data, and where that data comes from decides whether they are reliable and legal. Test data management is the planning, creation, storage and clean-up of the data tests use: what each test needs, how it gets there, who may see it and when it is removed. Rules that hold up:

  • Each test creates what it needs, through builders such as reading() above or fixtures, and asserts only on that data. A test that relies on "customer C-10006 is in the test database" breaks the day someone edits that customer.
  • Synthetic, not copied. Kestrel's test customers are invented (names such as "Test Klant 06", addresses on example.com, made-up EAN codes and meter IDs). A copy of the production database contains personal data of real customers, and the GDPR applies to every copy, including the one on a developer's laptop.
  • If production-like data is unavoidable, for example to reproduce a performance problem, it goes through pseudonymization or anonymization first, with the data protection officer's agreement.
  • Fixed values for time. Readings have fixed dates (5 October 2026, 23:15), never "now", so results do not change with the calendar. Include the awkward days on purpose: 25 October 2026 with its 100 quarter-hours.

Test environments

A test environment is the set of hardware, software, configuration, data and connections in which tests run, kept separate from production. ITSM (IT service management) describes the chain of environments a release passes through; for integration testing three kinds matter:

Kind Example at Kestrel Good for Watch out for
Inside the test run SQLite in memory, a fake meter-api, a container started by the test fast, isolated tests on every push stand-ins that drift from reality
Ephemeral, per branch or pull request portal, meter-api and PostgreSQL started together for one pipeline run system integration tests of the real versions start-up time, cost
Shared, long-lived the portal test environment used for acceptance tests acceptance, exploratory and performance testing data that changes under you, versions that differ from production, "who broke test?"

The closer an environment is to production (same versions, same configuration, realistic volumes), the more its results mean. Keep its configuration in code like production's, so a difference is a visible line in a file and not a surprise. A request to refresh or rebuild a shared environment is a normal service request (REQ-6140 at Kestrel: "refresh the portal test environment with the new synthetic customer set").

Keeping integration tests fast

Integration tests are slower than unit tests by nature, but a slow suite is one people stop running. What keeps them quick:

  • Start expensive things once. One database container per test session (scope="session"), not per test; isolation through transactions or unique data.
  • Mark and separate them. A pytest marker such as @pytest.mark.integration lets developers run pytest -m "not integration" locally and the pipeline run both in separate jobs, unit tests first. See fast pipelines.
  • Test each seam once. The repository needs integration tests for its queries; the cost calculation that uses the repository does not need a database in every test. Push the rest down to unit tests, as the test strategy page explains.
  • Short timeouts in tests. A client timeout of half a second against a local stand-in keeps a failing test from hanging the pipeline for minutes.