Test Provider

Provider test using Protobuf plugin with Pact Python v3.

This module demonstrates how to write a provider test using the Pact protobuf plugin with Pact Python's v3 API. The provider test verifies that the provider service correctly handles the contract defined by the consumer test.

The provider test runs the actual provider service and uses Pact to replay the consumer's interactions against the provider, verifying that the provider responds correctly with protobuf-serialized messages.

This example shows how to:

• Set up a FastAPI provider that handles protobuf responses
• Use the Pact Verifier with the protobuf plugin
• Handle provider states for setting up test data
• Verify protobuf serialization in the provider responses

Attributes

MOCK_ADDRESS_BOOK: AddressBook | None = None module-attribute

PROVIDER_URL = URL('http://localhost:8001') module-attribute

app = FastAPI(title='Protobuf Address Book API') module-attribute

FastAPI application

This application serves as the provider for the address book service, handling requests to retrieve person data by ID. It uses Protocol Buffers for serialization of the response data.

This code would typically be in a separate module within your application, but for the sake of this example, it is included directly within the test module.

Classes

Server

Bases: Server

Custom server class to run the FastAPI server in a separate thread.

This allows the provider test to run the FastAPI server in the background while Pact verifies the interactions against it.

Functions

install_signal_handlers() -> None

Prevent the server from installing signal handlers.

This is required to run the FastAPI server in a separate process.

Source code in examples/v3/plugins/protobuf/test_provider.py

def install_signal_handlers(self) -> None:
    """
    Prevent the server from installing signal handlers.

    This is required to run the FastAPI server in a separate process.
    """

run_in_thread() -> Generator[str, None, None]

Run the FastAPI server in a separate thread.

YIELDS	DESCRIPTION
str	The URL of the running server.

Source code in examples/v3/plugins/protobuf/test_provider.py

@contextlib.contextmanager
def run_in_thread(self) -> Generator[str, None, None]:
    """
    Run the FastAPI server in a separate thread.

    Yields:
        The URL of the running server.
    """
    thread = Thread(target=self.run)
    thread.start()
    try:
        while not self.started:
            time.sleep(0.01)
        yield f"http://{self.config.host}:{self.config.port}"
    finally:
        self.should_exit = True
        thread.join()

Functions

get_person(person_id: int) -> Response async

Get a person by ID, returning protobuf-serialized data.

PARAMETER	DESCRIPTION
person_id	The ID of the person to retrieve. TYPE: int

RETURNS	DESCRIPTION
Response	Response containing protobuf-serialized Person data.

RAISES	DESCRIPTION
HTTPException	If person is not found.

Source code in examples/v3/plugins/protobuf/test_provider.py

@app.get("/person/{person_id}")
async def get_person(person_id: int) -> Response:
    """
    Get a person by ID, returning protobuf-serialized data.

    Args:
        person_id: The ID of the person to retrieve.

    Returns:
        Response containing protobuf-serialized Person data.

    Raises:
        HTTPException: If person is not found.
    """
    if MOCK_ADDRESS_BOOK is None:
        raise HTTPException(status_code=404, detail="Person not found")

    # Find person by ID
    for person in MOCK_ADDRESS_BOOK.people:
        if person.id == person_id:
            # Serialize person to protobuf bytes
            protobuf_data = person.SerializeToString()
            return Response(
                content=protobuf_data,
                media_type="application/x-protobuf",
            )

    raise HTTPException(status_code=404, detail="Person not found")

provider_state_handler(state: str, action: Literal['setup', 'teardown'], parameters: dict[str, Any] | None = None) -> None

Handle provider state setup and teardown.

This function is called by Pact to set up the provider's internal state before each interaction is replayed. It ensures that the provider has the necessary data to respond correctly to the consumer's requests.

PARAMETER	DESCRIPTION
state	The provider state name from the consumer test. TYPE: str
action	Either "setup" or "teardown". TYPE: Literal['setup', 'teardown']
parameters	Additional parameters (not used in this example). TYPE: dict[str, Any] | None DEFAULT: None

Source code in examples/v3/plugins/protobuf/test_provider.py

def provider_state_handler(
    state: str,
    action: Literal["setup", "teardown"],
    parameters: dict[str, Any] | None = None,  # noqa: ARG001
) -> None:
    """
    Handle provider state setup and teardown.

    This function is called by Pact to set up the provider's internal state
    before each interaction is replayed. It ensures that the provider has
    the necessary data to respond correctly to the consumer's requests.

    Args:
        state:
            The provider state name from the consumer test.

        action:
            Either `"setup"` or `"teardown"`.

        parameters:
            Additional parameters (not used in this example).
    """
    if action == "setup":
        {
            "person with ID 1 exists": setup_person_exists,
            "person with ID 999 does not exist": setup_person_doesnt_exist,
        }[state]()

    if action == "teardown":
        {
            "person with ID 1 exists": teardown_person_exists,
            "person with ID 999 does not exist": teardown_person_doesnt_exist,
        }[state]()

server() -> Generator[str, None, None]

Fixture to start the FastAPI server for testing.

YIELDS	DESCRIPTION
str	The URL of the running server.

Source code in examples/v3/plugins/protobuf/test_provider.py

@pytest.fixture(scope="session")
def server() -> Generator[str, None, None]:
    """
    Fixture to start the FastAPI server for testing.

    Yields:
        The URL of the running server.
    """
    assert PROVIDER_URL.host is not None
    assert PROVIDER_URL.port is not None
    server = Server(
        uvicorn.Config(
            app,
            host=PROVIDER_URL.host,
            port=PROVIDER_URL.port,
        )
    )
    with server.run_in_thread() as url:
        yield url

setup_person_doesnt_exist() -> None

Set up the provider state where person with ID 999 doesn't exist.

This ensures the mock address book is empty so the provider will return a 404 error as expected.

Source code in examples/v3/plugins/protobuf/test_provider.py

def setup_person_doesnt_exist() -> None:
    """
    Set up the provider state where person with ID 999 doesn't exist.

    This ensures the mock address book is empty so the provider will
    return a 404 error as expected.
    """
    global MOCK_ADDRESS_BOOK  # noqa: PLW0603
    MOCK_ADDRESS_BOOK = AddressBook()  # Empty address book

setup_person_exists() -> None

Set up the provider state where person with ID 1 exists.

This creates a mock address book containing Alice (ID 1) so that the provider can return the expected protobuf data.

Source code in examples/v3/plugins/protobuf/test_provider.py

def setup_person_exists() -> None:
    """
    Set up the provider state where person with ID 1 exists.

    This creates a mock address book containing Alice (ID 1) so that
    the provider can return the expected protobuf data.
    """
    global MOCK_ADDRESS_BOOK  # noqa: PLW0603
    MOCK_ADDRESS_BOOK = address_book()

teardown_person_doesnt_exist() -> None

Clean up after testing person doesn't exist scenario.

Source code in examples/v3/plugins/protobuf/test_provider.py

def teardown_person_doesnt_exist() -> None:
    """
    Clean up after testing person doesn't exist scenario.
    """
    global MOCK_ADDRESS_BOOK  # noqa: PLW0603
    MOCK_ADDRESS_BOOK = None

teardown_person_exists() -> None

Clean up after testing person exists scenario.

Source code in examples/v3/plugins/protobuf/test_provider.py

def teardown_person_exists() -> None:
    """
    Clean up after testing person exists scenario.
    """
    global MOCK_ADDRESS_BOOK  # noqa: PLW0603
    MOCK_ADDRESS_BOOK = None

test_provider(server: str) -> None

Test the protobuf provider against the consumer contract.

This test uses the Pact Verifier to replay the consumer's interactions against the running provider service. It verifies that the provider correctly handles protobuf serialization and responds appropriately to both successful and error scenarios.

The test:

1. Configures the Verifier with the protobuf plugin
2. Points the verifier to the pact file generated by the consumer
3. Sets up state handlers to prepare test data
4. Verifies all interactions match the contract

Source code in examples/v3/plugins/protobuf/test_provider.py

def test_provider(server: str) -> None:
    """
    Test the protobuf provider against the consumer contract.

    This test uses the Pact Verifier to replay the consumer's interactions
    against the running provider service. It verifies that the provider
    correctly handles protobuf serialization and responds appropriately
    to both successful and error scenarios.

    The test:

    1.   Configures the Verifier with the protobuf plugin
    2.   Points the verifier to the pact file generated by the consumer
    3.   Sets up state handlers to prepare test data
    4.   Verifies all interactions match the contract
    """
    pact_file = (
        Path(__file__).parents[3] / "pacts" / "protobuf_consumer-protobuf_provider.json"
    )

    verifier = (
        Verifier("protobuf_provider")
        .add_transport(url=server)
        .add_source(pact_file)
        .state_handler(provider_state_handler, teardown=True)
    )

    verifier.verify()