Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions examples/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,14 @@ The code within the examples is intended to be well-documented and you are encou

## Available Examples

### Patterns Catalog

#### [Pact Patterns Catalog](./catalog/README.md)

- **Location**: `examples/catalog/`
- **Purpose**: Focused code snippets demonstrating specific Pact patterns
- **Content**: Well-documented use cases and techniques without full application context

### HTTP Examples

#### [aiohttp and Flask](./http/aiohttp_and_flask/README.md)
Expand Down
34 changes: 34 additions & 0 deletions examples/catalog/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
# Pact Patterns Catalog

This catalog contains focused, well-documented code snippets demonstrating specific Pact patterns and use cases. Unlike the full examples in the parent directory, catalog entries are designed to showcase a single pattern or technique with minimal application context.

## Available

- [Multipart Form Data with Matching Rules](./multipart_matching_rules/README.md)

## Using Catalog Entries

Catalog entries are intended as a reference for learning and adapting Pact patterns. To get the most value:

- **Read the documentation** in each entry's README to understand the pattern and its intended use.
- **Review the code** to see how the pattern is implemented in practice.
- **Explore the tests** to see example usages and edge cases.

If you want to experiment or adapt a pattern, you can run the tests for any entry:

```console
cd examples/catalog
uv run --group test pytest <catalog_entry_directory>
```

## Contributing Patterns

When adding a new catalog entry:

1. Focus on a single pattern or technique
2. Provide minimal but complete code, emphasizing the Pact aspects over application logic
3. Document the pattern thoroughly
4. Include working pytest tests
5. Add it to this README

For complete examples with full application context, consider adding to the main examples directory instead.
1 change: 1 addition & 0 deletions examples/catalog/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# noqa: D104
11 changes: 11 additions & 0 deletions examples/catalog/multipart_matching_rules/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@

# Multipart Form Data with Matching Rules

This entry demonstrates how to use Pact matching rules with multipart/form-data requests. Specifically, it demonstrates how multipart data is specified on the consumer side, and how matching rules can be applied to different parts of the multipart body.

For full implementation details, see the code and docstrings in this directory. For general catalog usage, prerequisites, and test-running instructions, see the [main catalog README](../README.md).

Related documentation:

* [Pact Matching Rules](https://docs.pact.io/getting_started/matching)
* [Multipart Form Data (RFC 7578)](https://tools.ietf.org/html/rfc7578)
1 change: 1 addition & 0 deletions examples/catalog/multipart_matching_rules/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# noqa: D104
187 changes: 187 additions & 0 deletions examples/catalog/multipart_matching_rules/test_consumer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,187 @@
"""
Consumer test demonstrating multipart/form-data with matching rules.

This test shows how to use Pact matching rules with multipart requests. The
examples illustrates this with a request containing both JSON metadata and
binary data (an image). The contract uses matching rules to validate
structure and types rather than exact values, allowing flexibility in the data
sent by the consumer and accepted by the provider.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import TYPE_CHECKING

import httpx
import pytest

from pact import Pact, match

if TYPE_CHECKING:
from collections.abc import Generator


# Minimal JPEG for testing. The important part is the magic bytes. The rest is
# not strictly valid JPEG data.
# fmt: off
JPEG_BYTES = bytes([
0xFF, 0xD8, # Start of Image (SOI) marker
0xFF, 0xE0, # JFIF APP0 Marker
0x00, 0x10, # Length of APP0 segment
0x4A, 0x46, 0x49, 0x46, 0x00, # "JFIF\0"
0x01, 0x02, # Major and minor versions
])
# fmt: on
"""
Some minimal JPEG bytes for testing multipart uploads.

In this example, we only need the JPEG magic bytes to validate the file type.
This is not a complete JPEG file, but is sufficient for testing purposes.
"""


@pytest.fixture
def pact() -> Generator[Pact, None, None]:
"""
Set up Pact for consumer contract testing.

This fixture initializes a Pact instance for the consumer tests, specifying
the consumer and provider names, and ensuring that the generated Pact files
are written to the appropriate directory after the tests run.
"""
pact = Pact(
"multipart-consumer",
"multipart-provider",
)
yield pact
pact.write_file(Path(__file__).parents[1] / "pacts")


def test_multipart_upload_with_matching_rules(pact: Pact) -> None:
"""
Test multipart upload with matching of the contents.

This test builds a `multipart/form-data` request by hand, and then uses a
library (`httpx`) to send the request to the mock server started by Pact.
Unlike simpler payloads, the matching rules _cannot_ be embedded within the
body itself. Instead, the body and matching rules are defined in separate
calls.

Some key points about this example:

- We use a matching rule for the `Content-Type` header to allow any valid
multipart boundary. This is important because many HTTP libraries
generate random boundaries automatically without user control.
- The body includes arbitrary binary data (a JPEG image) which cannot be
represented as a string. Therefore, it is critical that
`with_binary_body` is used to define the payload.
- Matching rules are defined for both the JSON metadata and the image part
to allow flexibility in the values sent by the consumer. The general
form to match a part within the multipart body is `$.<part name>`. So
to match a field in the `metadata` part, we use `$.metadata.<field>`; or
to match the content type of the `image` part, we use `$.image`:

```python
from pact import match

{
"body": {
"$.image": match.content_type("image/jpeg"),
"$.metadata.name": match.regex(regex=r"^[a-zA-Z]+$"),
},
}
```

/// warning

Proper content types are essential when working with multipart data. This
ensures that Pact can correctly identify and apply matching rules to each
part of the multipart body. If content types are missing or incorrect, the
matching rules may not be applied as expected, leading to test failures or
incorrect behavior.

///

To view the implementation, expand the source code below.
"""
# It is recommended to use a fixed boundary for testing, this ensures that
# the generated Pact is consistent across test runs.
boundary = "test-boundary-12345"

metadata = {
"name": "test",
"size": 100,
}

# Build multipart body with both JSON and binary parts. Note that since we
# are combining text and binary data, the strings must be encoded to bytes.
expected_body = (
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="metadata"\r\n'
f"Content-Type: application/json\r\n"
f"\r\n"
f"{json.dumps(metadata)}\r\n"
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="image"; filename="test.jpg"\r\n'
f"Content-Type: image/jpeg\r\n"
f"\r\n"
).encode()
expected_body += JPEG_BYTES
expected_body += f"\r\n--{boundary}--\r\n".encode()

# Define the interaction with matching rules
(
pact.upon_receiving("a multipart upload with JSON metadata and image")
.with_request("POST", "/upload")
.with_header(
"Content-Type",
# The matcher here is important if you don't have the ability to fix
# the boundary in the actual request (e.g., when using a library
# that generates it automatically).
match.regex(
f"multipart/form-data; boundary={boundary}",
regex=r"multipart/form-data;\s*boundary=.*",
),
)
.with_binary_body(
expected_body,
f"multipart/form-data; boundary={boundary}",
)
# Matching rules make the contract flexible
.with_matching_rules({
"body": {
"$.image": match.content_type("image/jpeg"),
"$.metadata": match.type({}),
"$.metadata.name": match.regex(regex=r"^[a-zA-Z]+$"),
"$.metadata.size": match.int(),
},
})
.will_respond_with(201)
.with_body({
"id": "upload-1",
"message": "Upload successful",
"metadata": {"name": "test", "size": 100},
"image_size": len(JPEG_BYTES),
})
)

# Execute the test. Note that the matching rules take effect here, so we can
# send data that differs from the example in the contract.
with pact.serve() as srv:
# Simple inline consumer: just make the multipart request
files = {
"metadata": (
None,
json.dumps({"name": "different", "size": 200}).encode(),
"application/json",
),
"image": ("test.jpg", JPEG_BYTES, "image/jpeg"),
}
response = httpx.post(f"{srv.url}/upload", files=files, timeout=5)

assert response.status_code == 201
result = response.json()
assert result["message"] == "Upload successful"
assert result["id"] == "upload-1"
140 changes: 140 additions & 0 deletions examples/catalog/multipart_matching_rules/test_provider.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
"""
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.
"""

from __future__ import annotations

import time
from pathlib import Path
from threading import Thread
from typing import TYPE_CHECKING, Annotated

import pytest
import uvicorn
from fastapi import FastAPI, Form, HTTPException, UploadFile
from pydantic import BaseModel

import pact._util
from pact import Verifier

if TYPE_CHECKING:
from typing import Any


# Simple FastAPI provider for handling multipart uploads
app = FastAPI()
"""
FastAPI application to handle multipart/form-data uploads.
"""

uploads: dict[str, dict[str, Any]] = {}
"""
In-memory store for uploaded files and metadata.
"""


class UploadMetadata(BaseModel):
"""
Model for the JSON metadata part of the upload.
"""

name: str
size: int


class UploadResponse(BaseModel):
"""
Model for the response returned after a successful upload.
"""

id: str
message: str
metadata: UploadMetadata
image_size: int


@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),
)


@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}"


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