Skip to content
Open
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
1 change: 1 addition & 0 deletions examples/http/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,3 +6,4 @@ This directory contains examples of HTTP-based contract testing with Pact.

- [`aiohttp_and_flask/`](aiohttp_and_flask/) - Async aiohttp consumer with Flask provider
- [`requests_and_fastapi/`](requests_and_fastapi/) - requests consumer with FastAPI provider
- [`xml_example/`](xml_example/) - requests consumer with FastAPI provider using XML bodies
Empty file.
13 changes: 13 additions & 0 deletions examples/http/xml_example/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from __future__ import annotations

from pathlib import Path

import pytest


EXAMPLE_DIR = Path(__file__).parent.resolve()


@pytest.fixture(scope="session")
def pacts_path() -> Path:
return EXAMPLE_DIR / "pacts"
110 changes: 110 additions & 0 deletions examples/http/xml_example/consumer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
"""
Requests XML consumer example.

This module defines a simple
[consumer](https://docs.pact.io/getting_started/terminology#service-consumer)
using the synchronous [`requests`][requests] library which will be tested with
Pact in the [consumer test][examples.http.xml_example.test_consumer].

The consumer sends requests expecting XML responses and parses them using the
standard library [`xml.etree.ElementTree`][xml.etree.ElementTree] module.

Note that the code in this module is agnostic of Pact (i.e., this would be your
production code). The `pact-python` dependency only appears in the tests.
"""

from __future__ import annotations

import logging
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import TYPE_CHECKING

import requests

if TYPE_CHECKING:
from types import TracebackType

from typing_extensions import Self

logger = logging.getLogger(__name__)


@dataclass()
class User:
"""
Represents a user as seen by the consumer.
"""

id: int
name: str


class UserClient:
"""
HTTP client for interacting with a user provider service via XML.
"""

def __init__(self, hostname: str) -> None:
"""
Initialise the user client.

Args:
hostname:
The base URL of the provider (must include scheme, e.g.,
`http://`).

Raises:
ValueError:
If the hostname does not start with 'http://' or `https://`.
"""
if not hostname.startswith(("http://", "https://")):
msg = "Invalid base URI"
raise ValueError(msg)
self._hostname = hostname
self._session = requests.Session()

def __enter__(self) -> Self:
"""
Begin the context for the client.
"""
self._session.__enter__()
return self

def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
"""
Exit the context for the client.
"""
self._session.__exit__(exc_type, exc_val, exc_tb)

def get_user(self, user_id: int) -> User:
"""
Fetch a user by ID from the provider, expecting an XML response.

Args:
user_id:
The ID of the user to fetch.

Returns:
A `User` instance parsed from the XML response.

Raises:
requests.HTTPError:
If the server returns a non-2xx response.
"""
logger.debug("Fetching user %s", user_id)
response = self._session.get(
f"{self._hostname}/users/{user_id}",
headers={"Accept": "application/xml"},
)
response.raise_for_status()
root = ET.fromstring(response.text) # noqa: S314
return User(
id=int(root.findtext("id")),
name=root.findtext("name"),
)
103 changes: 103 additions & 0 deletions examples/http/xml_example/provider.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
"""
FastAPI XML provider example.

This module defines a simple
[provider](https://docs.pact.io/getting_started/terminology#service-provider)
implemented with [`fastapi`](https://fastapi.tiangolo.com/) which will be tested
with Pact in the [provider test][examples.http.xml_example.test_provider].

The provider receives requests from the consumer and returns XML responses built
using the standard library [`xml.etree.ElementTree`][xml.etree.ElementTree]
module.

Note that the code in this module is agnostic of Pact (i.e., this would be your
production code). The `pact-python` dependency only appears in the tests.
"""

from __future__ import annotations

import logging
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import ClassVar

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import Response

logger = logging.getLogger(__name__)


@dataclass()
class User:
"""
Represents a user in the provider system.
"""

id: int
name: str


class UserDb:
"""
A simple in-memory user database abstraction for demonstration purposes.
"""

_db: ClassVar[dict[int, User]] = {}

@classmethod
def create(cls, user: User) -> None:
"""
Add a new user to the database.
"""
cls._db[user.id] = user

@classmethod
def delete(cls, user_id: int) -> None:
"""
Delete a user from the database by their ID.

Raises:
KeyError: If the user does not exist.
"""
if user_id not in cls._db:
msg = f"User {user_id} does not exist."
raise KeyError(msg)
del cls._db[user_id]

@classmethod
def get(cls, user_id: int) -> User | None:
"""
Retrieve a user by their ID.
"""
return cls._db.get(user_id)


app = FastAPI()


@app.get("/users/{uid}")
async def get_user_by_id(uid: int) -> Response:
"""
Retrieve a user by their ID, returning an XML response.

Args:
uid:
The user ID to retrieve.

Raises:
HTTPException: If the user is not found, a 404 error is raised.
"""
logger.debug("GET /users/%s", uid)
user = UserDb.get(uid)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
root = ET.Element("user")
ET.SubElement(root, "id").text = str(user.id)
ET.SubElement(root, "name").text = user.name
return Response(
content=ET.tostring(root, encoding="unicode"),
media_type="application/xml",
)
27 changes: 27 additions & 0 deletions examples/http/xml_example/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
#:schema https://www.schemastore.org/pyproject.json
[project]
name = "example-xml"

description = "Example of XML contract testing with Pact Python"

dependencies = ["requests~=2.0", "fastapi~=0.0", "typing-extensions~=4.0"]
requires-python = ">=3.10"
version = "1.0.0"

[dependency-groups]
test = ["pact-python", "pytest~=9.0", "uvicorn~=0.29"]

[tool.uv.sources]
pact-python = { path = "../../../" }

[tool.ruff]
extend = "../../../pyproject.toml"

[tool.pytest]
addopts = ["--import-mode=importlib"]

asyncio_default_fixture_loop_scope = "session"

log_date_format = "%H:%M:%S"
log_format = "%(asctime)s.%(msecs)03d [%(levelname)-8s] %(name)s: %(message)s"
log_level = "NOTSET"
93 changes: 93 additions & 0 deletions examples/http/xml_example/test_consumer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
"""
Consumer contract tests using Pact (XML).

This module demonstrates how to test a consumer (see
[`consumer.py`][examples.http.xml_example.consumer]) against a mock provider
using Pact. The key difference from JSON-based examples is that the response
body is specified as a plain XML string — no matchers are used, as XML matchers
do not exist in pact-python. The `Accept` header is set via a separate
`.with_header()` call after `.with_request()`.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

import pytest
import requests

from examples.http.xml_example.consumer import UserClient
from pact import Pact

if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path

logger = logging.getLogger(__name__)


@pytest.fixture
def pact(pacts_path: Path) -> Generator[Pact, None, None]:
"""
Set up a Pact mock provider for consumer tests.

Args:
pacts_path:
The path where the generated pact file will be written.

Yields:
A Pact object for use in tests.
"""
pact = Pact("xml-consumer", "xml-provider").with_specification("V4")
yield pact
pact.write_file(pacts_path)


def test_get_user(pact: Pact) -> None:
"""
Test the GET request for a user, expecting an XML response.

The response body is a plain XML string. Note that `.with_header()` is
called as a separate chain step — `with_request()` does not accept a
headers argument.
"""
response = "<user><id>123</id><name>Alice</name></user>"
(
pact
.upon_receiving("A request for a user as XML")
.given("the user exists", id=123, name="Alice")
.with_request("GET", "/users/123")
.with_header("Accept", "application/xml")
.will_respond_with(200)
.with_body(response, content_type="application/xml")
)

with (
pact.serve() as srv,
UserClient(str(srv.url)) as client,
):
user = client.get_user(123)
assert user.id == 123
assert user.name == "Alice"


def test_get_unknown_user(pact: Pact) -> None:
"""
Test the GET request for an unknown user, expecting a 404 response.
"""
(
pact
.upon_receiving("A request for an unknown user as XML")
.given("the user doesn't exist", id=123)
.with_request("GET", "/users/123")
.with_header("Accept", "application/xml")
.will_respond_with(404)
)

with (
pact.serve() as srv,
UserClient(str(srv.url)) as client,
pytest.raises(requests.HTTPError),
):
client.get_user(123)
Loading