Testing the Paths That Fail

Error-handling code is the least-run code in any library and the most consequential when it’s wrong. It is also the easiest to leave untested, because making things fail on demand takes deliberate effort.

This lesson is about that effort: how to force a failure, and what to assert once you’ve forced one.

A test that proves nothing

Here’s the shape almost everyone writes first:

    def test_save_fails_on_a_duplicate(self):
        Post(title="first", views=1).save(self.connection)
        with self.assertRaises(IntegrityError):
            Post(title="first", views=2).save(self.connection)

It asserts that an exception came out. It says nothing about the state of the database afterwards — which is the entire reason the transaction code exists. A version of save() that wrote half the row, corrupted the cache and then raised would pass this test.

The question to ask about a failure test is: what should be true after the failure? Usually three things:

the caller got a specific exception, carrying a useful message and cause
the data is unchanged — no partial write
the library’s own state is unchanged — no half-updated cache, no leaked entry

Three assertions, one failure Python

import sqlite3
import unittest
from contextlib import contextmanager


class IntegrityError(Exception):
    pass


@contextmanager
def transaction(connection):
    connection.execute("begin")
    try:
        yield connection
    except sqlite3.IntegrityError as error:
        connection.execute("rollback")
        raise IntegrityError(str(error)) from error
    except BaseException:
        connection.execute("rollback")
        raise
    else:
        connection.execute("commit")


CACHE = {}


def save_all(connection, rows):
    with transaction(connection):
        for title in rows:
            connection.execute("insert into post (title) values (?)", (title,))
    CACHE.update({title: True for title in rows})     # only after the commit


class SaveAll(unittest.TestCase):
    def setUp(self):
        self.connection = sqlite3.connect(":memory:", isolation_level=None)
        self.connection.execute("create table post (id integer primary key, title text unique)")
        self.connection.execute("insert into post (title) values ('first')")
        CACHE.clear()

    def titles(self):
        return [row[0] for row in self.connection.execute("select title from post order by id")]

    def test_a_duplicate_is_reported_and_changes_nothing(self):
        with self.assertRaises(IntegrityError) as caught:
            save_all(self.connection, ["second", "first"])

        self.assertIn("UNIQUE constraint failed", str(caught.exception))
        self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
        self.assertEqual(self.titles(), ["first"])          # "second" was rolled back
        self.assertEqual(CACHE, {})                         # nothing was remembered


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(SaveAll)
)

Four assertions, and each one fails for a different bug: a wrong exception type, a lost cause, a partial write, a cache that ran ahead of the commit.

assertRaises as a context manager

with self.assertRaises(X) as caught: gives you caught.exception afterwards, which is how you assert on the message and the cause. The one-line form, self.assertRaises(X, function, arg), can’t. There’s also assertRaisesRegex(X, "pattern") when the message is all you care about.

Making the failure happen

Real failures are inconvenient: a disk fills, a network drops, a constraint trips on the third of five rows. Three ways to produce one on demand, in order of preference.

1. Cause it for real. A unique constraint and a duplicate row is a real IntegrityError from a real driver. Nothing about it is simulated, so nothing about it can be wrong. Prefer this whenever the real thing is reachable.

2. Write a substitute that breaks on cue. When the real failure is hard to provoke — an I/O error mid-write — a small fake is honest and clear:

A connection that fails on the third statement Python

import sqlite3
import unittest
from contextlib import contextmanager


class Unreliable:
    """Passes everything through, then fails once, at a chosen statement."""

    def __init__(self, connection, fail_on):
        self.real = connection
        self.fail_on = fail_on
        self.count = 0

    def execute(self, sql, parameters=(), /):
        self.count += 1
        if self.count == self.fail_on:
            raise sqlite3.OperationalError("disk I/O error")
        return self.real.execute(sql, parameters)

    def commit(self):
        self.real.commit()


@contextmanager
def transaction(connection):
    connection.execute("begin")
    try:
        yield connection
    except BaseException:
        connection.real.execute("rollback")     # the fake is already broken
        raise
    else:
        connection.execute("commit")


class MidWriteFailure(unittest.TestCase):
    def test_a_failure_in_the_middle_leaves_nothing(self):
        real = sqlite3.connect(":memory:", isolation_level=None)
        real.execute("create table post (id integer primary key, title text)")
        # statements: 1 begin, 2 first insert, 3 second insert -> boom
        connection = Unreliable(real, fail_on=3)

        with self.assertRaises(sqlite3.OperationalError):
            with transaction(connection):
                connection.execute("insert into post (title) values ('one')")
                connection.execute("insert into post (title) values ('two')")

        self.assertEqual(real.execute("select count(*) from post").fetchone()[0], 0)


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(MidWriteFailure)
)

Counting statements is crude and it is also exactly what you want here: the test says fail at the third statement, and if the library starts issuing a different number of statements, the test tells you by failing. A fake that is too clever hides that.

3. Patch something. unittest.mock.patch can make any function raise. It’s the most powerful and the least trustworthy, because it asserts against your belief about the code rather than the code. Keep it for cases the other two can’t reach — and patch the thing your code calls, not the thing your code is.

A fake that can't fail the way the real thing does

Unreliable raises sqlite3.OperationalError because that’s what a real disk error raises. If you invent class FakeError(Exception) for a test, you’ve tested that your code handles an exception that can never occur. Check what the real thing raises — the driver’s documentation, or a quick experiment — before inventing one.

Cleanup is part of the contract

The other half of failure testing is what happens when nothing failed. Every resource the library acquires should be gone afterwards:

    def test_nothing_is_left_behind_between_transactions(self) -> None:
        Post(title="first", views=1).save(self.connection)
        self.assertEqual(transactions._depth, {})

And addCleanup is how a test tidies up after itself even when it fails partway:

        self.folder = tempfile.TemporaryDirectory()
        self.addCleanup(self.folder.cleanup)

addCleanup runs in reverse order, runs even if setUp later raises, and survives a failing assertion — all things a tearDown method gets subtly wrong when there’s more than one resource.

Exercise

Test a retry that gives up

fetch_with_retry retries a flaky call, then raises. The implementation is given and correct; the tests are the exercise. Three are written for you as examples, and three are missing.

import unittest


class Retryable(Exception):
    pass


class GaveUp(Exception):
    def __init__(self, attempts, last):
        super().__init__(f"gave up after {attempts} attempts")
        self.attempts = attempts
        self.last = last


def fetch_with_retry(fetch, attempts=3):
    """Call fetch() until it succeeds, up to `attempts` times."""
    for attempt in range(1, attempts + 1):
        try:
            return fetch()
        except Retryable as error:
            last = error
    raise GaveUp(attempts, last) from last


class Flaky:
    """Fails `failures` times, then returns "ok"."""

    def __init__(self, failures):
        self.failures = failures
        self.calls = 0

    def __call__(self):
        self.calls += 1
        if self.calls <= self.failures:
            raise Retryable(f"attempt {self.calls} failed")
        return "ok"


class Retries(unittest.TestCase):
    def test_a_first_time_success_is_returned(self):
        self.assertEqual(fetch_with_retry(Flaky(failures=0)), "ok")

    def test_it_succeeds_after_a_failure(self):
        self.assertEqual(fetch_with_retry(Flaky(failures=2)), "ok")

    def test_it_stops_at_the_limit(self):
        fetch = Flaky(failures=99)
        with self.assertRaises(GaveUp):
            fetch_with_retry(fetch, attempts=3)
        self.assertEqual(fetch.calls, 3, "it should not try a fourth time")

    # TODO: it doesn't retry more than it has to (a success on attempt 2 means
    #       exactly 2 calls)
    # TODO: GaveUp keeps the last Retryable as its cause, and as .last
    # TODO: an error that isn't Retryable comes straight out, with no retries


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(Retries)
)

Hint 1: the idea

Each missing test needs something it can count or inspect. Flaky already counts its calls, so “doesn’t retry more than it has to” is a claim about calls. assertRaises used with as hands you the exception to inspect — its __cause__ and .last. For the non-retryable case Flaky won’t do, since it only raises Retryable: write a tiny fetch function that raises something else and counts its own calls.

Hint 2: a first step

Copy test_it_stops_at_the_limit and change three things: Flaky(failures=1), expect "ok", and assert calls == 2. Then, for the second test, with self.assertRaises(GaveUp) as caught: — afterwards, caught.exception is the error.

Show one solution

    def test_it_does_not_retry_more_than_it_has_to(self):
        fetch = Flaky(failures=1)
        self.assertEqual(fetch_with_retry(fetch), "ok")
        self.assertEqual(fetch.calls, 2)

    def test_gave_up_keeps_the_last_failure(self):
        with self.assertRaises(GaveUp) as caught:
            fetch_with_retry(Flaky(failures=99), attempts=2)
        self.assertIsInstance(caught.exception.__cause__, Retryable)
        self.assertEqual(str(caught.exception.last), "attempt 2 failed")
        self.assertEqual(caught.exception.attempts, 2)

    def test_other_errors_are_not_retried(self):
        calls = 0

        def fetch():
            nonlocal calls
            calls += 1
            raise ValueError("not retryable")

        with self.assertRaises(ValueError):
            fetch_with_retry(fetch)
        self.assertEqual(calls, 1)

The third test is the one that matters most and gets written least. A retry loop that catches too much turns a permanent bug — a TypeError from a typo in your own code — into three identical failures and a confusing GaveUp. Pinning “only Retryable is retried” is what stops a later “let’s be more robust” edit from doing that.

The second test asserts on the last failure, not just that something was raised. GaveUp exists to carry that information, and information a class carries but no test reads is information that quietly stops being correct.

Notice what none of these tests do: sleep. fetch_with_retry has no backoff, and if it had one, the delay would be a parameter with a default, so tests could pass zero. Never test a retry loop that sleeps for real — it makes the suite slow, and slow suites get skipped.

What you learned

A failure test asserts on the exception, the data and the library’s state, not just that something was raised.
Prefer a real failure, then a small fake that breaks on cue, then patching.
A fake must raise what the real thing raises.
addCleanup releases resources even when a test fails partway, and cleanup deserves its own assertions.