Skip to content

Test provider

Provider verification for multipart/form-data contract with matching rules.

This test demonstrates provider verification for contracts with multipart requests and matching rules. It uses FastAPI to create a simple server that can process multipart uploads containing JSON metadata and a JPEG image. The test verifies that the provider complies with the contract generated by the consumer tests, ensuring that it can handle variations in the data while maintaining compatibility.

Attributes

app = FastAPI() module-attribute

FastAPI application to handle multipart/form-data uploads.

uploads: dict[str, dict[str, Any]] = {} module-attribute

In-memory store for uploaded files and metadata.

Classes

UploadMetadata pydantic-model

Bases: BaseModel

Model for the JSON metadata part of the upload.

Fields:

Attributes

name: str pydantic-field
size: int pydantic-field

UploadResponse pydantic-model

Bases: BaseModel

Model for the response returned after a successful upload.

Fields:

Attributes

id: str pydantic-field
image_size: int pydantic-field
message: str pydantic-field
metadata: UploadMetadata pydantic-field

Functions

app_server() -> str

Start FastAPI server for provider verification.

Source code in examples/catalog/multipart_matching_rules/test_provider.py
@pytest.fixture
def app_server() -> str:
    """
    Start FastAPI server for provider verification.
    """
    hostname = "localhost"
    port = pact._util.find_free_port()  # noqa: SLF001
    Thread(
        target=uvicorn.run,
        args=(app,),
        kwargs={"host": hostname, "port": port},
        daemon=True,
    ).start()
    time.sleep(0.1)  # Allow server time to start
    return f"http://{hostname}:{port}"

test_provider_multipart(app_server: str) -> None

Verify the provider against the multipart upload contract.

In general, there are no special considerations for verifying providers with multipart requests. The Pact verifier will read the contract file generated by the consumer tests and ensure that the provider can handle requests that conform to the specified matching rules.

As with any provider verification, the test needs to ensure that provider states are set up correctly. This example does not include any provider states to ensure simplicity.

Source code in examples/catalog/multipart_matching_rules/test_provider.py
def test_provider_multipart(app_server: str) -> None:
    """
    Verify the provider against the multipart upload contract.

    In general, there are no special considerations for verifying providers with
    multipart requests. The Pact verifier will read the contract file generated
    by the consumer tests and ensure that the provider can handle requests that
    conform to the specified matching rules.

    As with any provider verification, the test needs to ensure that provider
    states are set up correctly. This example does not include any provider
    states to ensure simplicity.
    """
    verifier = (
        Verifier("multipart-provider")
        .add_source(Path(__file__).parents[1] / "pacts")
        .add_transport(url=app_server)
    )

    verifier.verify()

    assert len(uploads) > 0, "No uploads were processed by the provider"

upload_file(metadata: Annotated[str, Form()], image: Annotated[UploadFile, Form()]) -> UploadResponse async

Handle multipart upload with JSON metadata and image.

This endpoint processes a multipart/form-data request containing a JSON metadata part and an image file part. It validates the metadata structure and the image content type, then stores the upload in memory.

Source code in examples/catalog/multipart_matching_rules/test_provider.py
@app.post("/upload", status_code=201)
async def upload_file(
    metadata: Annotated[str, Form()],
    image: Annotated[UploadFile, Form()],
) -> UploadResponse:
    """
    Handle multipart upload with JSON metadata and image.

    This endpoint processes a multipart/form-data request containing a JSON
    metadata part and an image file part. It validates the metadata structure
    and the image content type, then stores the upload in memory.
    """
    metadata_dict = UploadMetadata.model_validate_json(metadata)
    if image.content_type != "image/jpeg":
        msg = f"Expected image/jpeg, got {image.content_type}"
        raise HTTPException(status_code=400, detail=msg)

    content = await image.read()
    if not content.startswith((b"\xff\xd8\xff\xdb", b"\xff\xd8\xff\xe0")):
        msg = "Invalid/malformed JPEG file"
        raise HTTPException(status_code=400, detail=msg)

    upload_id = f"upload-{len(uploads) + 1}"
    uploads[upload_id] = {
        "id": upload_id,
        "metadata": metadata_dict,
        "filename": image.filename,
        "content_type": image.content_type,
        "size": len(content),
    }

    return UploadResponse(
        id=upload_id,
        message="Upload successful",
        metadata=metadata_dict,
        image_size=len(content),
    )