An example on how to capture and test logs with pytest.

test_capture_log.py

import time
import logging
import sqlite3

import pytest

logger = logging.getLogger("ExampleDBClient")
RECONNECT_SLEEP = 30
RECONNECT_ATTEMPTS = 3


def mock_return(*args, **kwargs):
    raise KeyError("Mock Error")


def mock_sleep(*args):
    return None


class ExampleDBClient:

    def __init__(self, user="dummy", host="localhost", port="1234", database="example"):
        self.user = user
        self.host = host
        self.port = port
        self.database = database
        self.full_config = dict(user=user, host=host, port=port, database=database)
        self.connection = self.connect()

    def connect(self, attempts=1):
        try:
            self.connection = sqlite3.connect(**self.full_config)
            logger.info("Connected to %s@%s:%s/%s in attempt %d.",
                        self.user, self.host, self.port, self.database, attempts)
            return self.connection

        except Exception as e:
            if attempts == RECONNECT_ATTEMPTS:
                logger.error("Tried to connect too many times (%u), stopping %s@%s:%s/%s",
                             attempts, self.user, self.host, self.port, self.database)
                raise e

            logger.error("Error connecting to %s @ %s : %s / %s",
                         self.user, self.host, self.port, self.database)
            logger.exception(e)

        # Reconnection
        time.sleep(RECONNECT_SLEEP)
        self.connection = self.connect(attempts=attempts + 1)
        return self.connection


def test_reconnection_fails(monkeypatch, caplog):
    monkeypatch.setattr(time, "sleep", mock_sleep)  # Will not sleep
    monkeypatch.setattr(sqlite3, "connect", mock_return)

    with pytest.raises(Exception):
        _ = ExampleDBClient()

    print(caplog.records)
    assert caplog.records[0].msg == "Error connecting to %s @ %s : %s / %s"
    assert caplog.records[1].msg.args[0] == "Mock Error"
    assert caplog.records[2].msg == "Error connecting to %s @ %s : %s / %s"
    assert caplog.records[3].msg.args[0] == "Mock Error"
    assert caplog.records[4].msg == "Tried to connect too many times (%u), stopping %s@%s:%s/%s"