Integration fixtures

Because we always need Docker to run the Postgres integration test, we can create a pytest fixture to handle the management of Docker. We will take the CLI command we used before and write a fixture around them.

# tests/fixtures.py
@pytest.fixture(scope="session", autouse=True)
def docker_compose():
    file_path = Path(__file__).absolute().parent / "docker-compose.yaml"

    # Tear down any existing containers and volumes to ensure clean state
    subprocess.run(
        ["docker", "compose", "-f", file_path, "down", "-v"],
        capture_output=True,
    )

    # Start Docker Compose
    subprocess.run(
        ["docker", "compose", "-f", file_path, "up", "--build", "-d"],
        check=True,
        capture_output=True,
    )

    # Wait for PostgreSQL to be ready to accept connections
    max_retries = 10
    for i in range(max_retries):
        result = subprocess.run(
            ["docker", "exec", "postgresql", "pg_isready"],
            capture_output=True,
        )
        if result.returncode == 0:
            break
        time.sleep(2)

    # Wait for the initialization scripts to complete (schema and table creation)
    for i in range(max_retries):
        result = subprocess.run(
            [
                "docker",
                "exec",
                "postgresql",
                "psql",
                "-U",
                "test_user",
                "-d",
                "test_db",
                "-c",
                "SELECT 1 FROM data.city_population LIMIT 1;",
            ],
            capture_output=True,
        )
        if result.returncode == 0:
            break
        time.sleep(2)

    yield

    # Tear down Docker Compose and remove volumes
    subprocess.run(["docker", "compose", "-f", file_path, "down", "-v"], check=True)

The code above does the following:

  1. Gets the file path of the docker-compose.yaml relative to the test (which is in the same directory).
  2. Starts the Docker service using the docker compose command.
  3. Waits for the service to be ready by checking when the database is ready using pg_isready.
  4. Yields to the test execution (running the test).
  5. Spins down the Docker service when testing has completed.

Now that fixture can be used within the parameters of the integration test and couple our test and the underlying services:

# tests/test_lesson_5.py
def test_state_population_database(docker_compose):  # noqa: F811
    postgres_resource = PostgresResource(
        host="localhost",
        user="test_user",
        password="test_pass",
        database="test_db",
    )

    result = lesson_5.state_population_database(postgres_resource)
    assert result == [
        ("New York", 8804190),
        ("Buffalo", 278349),
    ]

> pytest tests/test_lesson_5.py::test_state_population_database
...
tests/test_lesson_5.py .                                                          [100%]

Because this pytest fixture is scoped to the session, it will remain active for all the tests run. This can be very helpful because you may notice these tests take longer than the unit tests. Most of this is the set up time to spin up the necessary services (the tests themselves are relatively quick).

Scoping everything to the session ensures that we do not have to needlessly spin the services up and down for each test.