A test fixture supplies setup that a test can reuse. In pytest, a test requests a fixture by naming it as a function argument. Fixtures can be mixed and matched: one might provide a temporary directory while another supplies a database container. section 10.1.5 section 10.2
Give a file-producing operator a safe output path
To test a JSON-to-CSV operator, accept `tmp_path` as a test argument. Write input JSON under that path, pass an output path there to the operator, execute it, and read the CSV to check the result. pytest creates the temporary directory for the test, and its contents are removed afterward. The default fixture scope is one test function, so another run gets its own path. section 10.1.5
Listing 10.20 Testing using temporary paths
import csv
import json
from pathlib import Path
from custom.json_to_csv_operator import JsonToCsvOperator
def test_json_to_csv_operator(tmp_path: Path): #1
input_path = tmp_path / "input.json" #2
output_path = tmp_path / "output.csv" #2
input_data = [ #3
{"name": "bob", "age": "41", "sex": "M"}, #3
{"name": "alice", "age": "24", "sex": "F"}, #3
{"name": "carol", "age": "60", "sex": "F"}, #3
] #3
with open(input_path, "w") as f: #3
f.write(json.dumps(input_data)) #3
operator = JsonToCsvOperator(
task_id="test",
input_path=input_path,
output_path=output_path,
)
operator.execute(context={}) #4
with open(output_path, "r") as f: #5
reader = csv.DictReader(f) #5
result = [dict(row) for row in reader] #5
assert result == input_data #6
#7
The test uses `tmp_path` for both input and output, then checks the generated CSV against the input data. section 10.1.5
Fixtures also support integration tests. With pytest-docker-tools, `fetch` obtains a PostgreSQL image and `container` starts a database for a test. The test requests these fixtures by argument name. Mapping the container port to a random open host port avoids conflicts with services already running locally. Docker must be installed and accessible to the test environment. section 10.2
Listing 10.24 Starting a Docker container for a test with pytest_docker_tools fixture
from pytest_docker_tools import container
postgres_container = container(
image="{postgres_image.id}",
ports={"5432/tcp": None},
)
def test_call_fixture(postgres_container):
print(
f"Running Postgres container named {postgres_container.name} "
f"on port {postgres_container.ports['5432/tcp'][0]}."
)
The container fixture uses the fetched image and exposes PostgreSQL on an available host port. section 10.2
Using a real database lets an integration test exercise hook behavior without mocking its database calls. pytest-docker-tools manages the container lifecycle; the test still needs to run the operator and verify its results. Fixture scopes and further pytest features offer more control over reuse, but are not needed for the basic pattern. section 10.2 section 10.1.5
