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")

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

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

Handle provider state for when a person with the given ID doesn't exist.

PARAMETER	DESCRIPTION
action	Either "setup" or "teardown". TYPE: Literal['setup', 'teardown']
parameters	Dictionary containing user_id key. TYPE: dict[str, Any] | None DEFAULT: None

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

def state_person_doesnt_exist(
    action: Literal["setup", "teardown"],
    parameters: dict[str, Any] | None = None,
) -> None:
    """
    Handle provider state for when a person with the given ID doesn't exist.

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

        parameters:
            Dictionary containing user_id key.
    """
    global MOCK_ADDRESS_BOOK  # noqa: PLW0603

    if action == "setup":
        MOCK_ADDRESS_BOOK = AddressBook()
        if user_id := parameters.get("user_id") if parameters else None:
            assert not any(
                person.id == user_id for person in MOCK_ADDRESS_BOOK.people
            ), f"Person with ID {user_id} should not exist in address book"
        else:
            msg = "User ID not provided"
            raise AssertionError(msg)
    elif action == "teardown":
        MOCK_ADDRESS_BOOK = None

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

Handle provider state for when a person with the given ID exists.

PARAMETER	DESCRIPTION
action	Either "setup" or "teardown". TYPE: Literal['setup', 'teardown']
parameters	Dictionary containing user_id key. TYPE: dict[str, Any] | None DEFAULT: None

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

def state_person_exists(
    action: Literal["setup", "teardown"],
    parameters: dict[str, Any] | None = None,
) -> None:
    """
    Handle provider state for when a person with the given ID exists.

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

        parameters:
            Dictionary containing user_id key.
    """
    global MOCK_ADDRESS_BOOK  # noqa: PLW0603

    if action == "setup":
        MOCK_ADDRESS_BOOK = address_book()
        if user_id := parameters.get("user_id") if parameters else None:
            assert any(person.id == user_id for person in MOCK_ADDRESS_BOOK.people), (
                f"Person with ID {user_id} does not exist in address book"
            )
        else:
            msg = "User ID not provided"
            raise AssertionError(msg)
    elif action == "teardown":
        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(
            {
                "person with the given ID exists": state_person_exists,
                "person with the given ID does not exist": state_person_doesnt_exist,
            },
            teardown=True,
        )
    )

    verifier.verify()