From 41572984459e2233b0fe80d4eab9b7459b020211 Mon Sep 17 00:00:00 2001 From: Benjamin Aduo Date: Fri, 20 Mar 2026 00:44:29 +0100 Subject: [PATCH] feat(examples): add XML contract testing example (Issue #372) - Add examples/http/xml_example/ with consumer, provider, and tests - Consumer uses xml.etree.ElementTree to parse XML responses - Provider is a FastAPI app returning application/xml responses - Tests cover GET user (200) and unknown user (404) scenarios - Add local conftest.py with pacts_path fixture - Update examples/http/README.md to list new example Co-Authored-By: Abacus.AI CLI --- examples/http/README.md | 1 + examples/http/xml_example/__init__.py | 0 examples/http/xml_example/conftest.py | 13 ++ examples/http/xml_example/consumer.py | 110 +++++++++++++++++ examples/http/xml_example/provider.py | 103 ++++++++++++++++ examples/http/xml_example/pyproject.toml | 27 +++++ examples/http/xml_example/test_consumer.py | 93 ++++++++++++++ examples/http/xml_example/test_provider.py | 135 +++++++++++++++++++++ 8 files changed, 482 insertions(+) create mode 100644 examples/http/xml_example/__init__.py create mode 100644 examples/http/xml_example/conftest.py create mode 100644 examples/http/xml_example/consumer.py create mode 100644 examples/http/xml_example/provider.py create mode 100644 examples/http/xml_example/pyproject.toml create mode 100644 examples/http/xml_example/test_consumer.py create mode 100644 examples/http/xml_example/test_provider.py diff --git a/examples/http/README.md b/examples/http/README.md index 4f97723ca..1579c6312 100644 --- a/examples/http/README.md +++ b/examples/http/README.md @@ -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 diff --git a/examples/http/xml_example/__init__.py b/examples/http/xml_example/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/examples/http/xml_example/conftest.py b/examples/http/xml_example/conftest.py new file mode 100644 index 000000000..13e9dee94 --- /dev/null +++ b/examples/http/xml_example/conftest.py @@ -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" diff --git a/examples/http/xml_example/consumer.py b/examples/http/xml_example/consumer.py new file mode 100644 index 000000000..5f080609b --- /dev/null +++ b/examples/http/xml_example/consumer.py @@ -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"), + ) diff --git a/examples/http/xml_example/provider.py b/examples/http/xml_example/provider.py new file mode 100644 index 000000000..1a051874f --- /dev/null +++ b/examples/http/xml_example/provider.py @@ -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", + ) diff --git a/examples/http/xml_example/pyproject.toml b/examples/http/xml_example/pyproject.toml new file mode 100644 index 000000000..3275a020e --- /dev/null +++ b/examples/http/xml_example/pyproject.toml @@ -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" diff --git a/examples/http/xml_example/test_consumer.py b/examples/http/xml_example/test_consumer.py new file mode 100644 index 000000000..a7f758046 --- /dev/null +++ b/examples/http/xml_example/test_consumer.py @@ -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 = "123Alice" + ( + 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) diff --git a/examples/http/xml_example/test_provider.py b/examples/http/xml_example/test_provider.py new file mode 100644 index 000000000..11c1e8b89 --- /dev/null +++ b/examples/http/xml_example/test_provider.py @@ -0,0 +1,135 @@ +""" +Provider contract tests using Pact (XML). + +This module demonstrates how to test a FastAPI provider (see +[`provider.py`][examples.http.xml_example.provider]) against a mock consumer +using Pact. The mock consumer replays the requests defined by the consumer +contract, and Pact validates that the provider responds as expected. + +Provider state handlers set up the in-memory database before each interaction +is verified, ensuring repeatable and isolated contract verification. +""" + +from __future__ import annotations + +import contextlib +import logging +from threading import Thread +from typing import TYPE_CHECKING, Any, Literal + +import pytest +import uvicorn + +import pact._util +from examples.http.xml_example.provider import User, UserDb, app +from pact import Verifier + +if TYPE_CHECKING: + from pathlib import Path + +logger = logging.getLogger(__name__) + + +@pytest.fixture(scope="session") +def app_server() -> str: + """ + Run the FastAPI server for provider verification. + + Returns: + The base URL of the running FastAPI server. + """ + hostname = "localhost" + port = pact._util.find_free_port() # noqa: SLF001 + Thread( + target=uvicorn.run, + args=(app,), + kwargs={"host": hostname, "port": port}, + daemon=True, + ).start() + return f"http://{hostname}:{port}" + + +def test_provider(app_server: str, pacts_path: Path) -> None: + """ + Test the provider against the consumer contract. + + Runs the Pact verifier against the FastAPI provider using the contract + generated by the consumer tests. State handlers ensure the database is + in the correct state for each interaction. + """ + verifier = ( + Verifier("xml-provider") + .add_source(pacts_path) + .add_transport(url=app_server) + .state_handler( + { + "the user exists": mock_user_exists, + "the user doesn't exist": mock_user_does_not_exist, + }, + teardown=True, + ) + ) + + verifier.verify() + + +def mock_user_exists( + action: Literal["setup", "teardown"], + parameters: dict[str, Any], +) -> None: + """ + Mock the provider state where a user exists. + + Args: + action: + Either "setup" or "teardown". + parameters: + Must contain "id" and optionally "name". + """ + user = User( + id=int(parameters.get("id", 1)), + name=str(parameters.get("name", "Alice")), + ) + + if action == "setup": + UserDb.create(user) + return + + if action == "teardown": + with contextlib.suppress(KeyError): + UserDb.delete(user.id) + return + + msg = f"Unknown action: {action}" + raise ValueError(msg) + + +def mock_user_does_not_exist( + action: Literal["setup", "teardown"], + parameters: dict[str, Any], +) -> None: + """ + Mock the provider state where a user does not exist. + + Args: + action: + Either "setup" or "teardown". + parameters: + Must contain "id". + """ + if "id" not in parameters: + msg = "State must contain an 'id' field to mock user non-existence" + raise ValueError(msg) + + uid = int(parameters["id"]) + + if action == "setup": + if user := UserDb.get(uid): + UserDb.delete(user.id) + return + + if action == "teardown": + return + + msg = f"Unknown action: {action}" + raise ValueError(msg)