chore: refactor tests

Signed-off-by: JP-Ellis <josh@jpellis.me>

Author: JP-Ellis

examples/.ruff.toml

@@ -2,10 +2,10 @@ extend = "../pyproject.toml"
 
 [lint]
 ignore = [
-  "S101", # Forbid assert statements
-  "D103", # Require docstring in public function
-  "D104", # Require docstring in public package
-  "PLR2004" # Forbid Magic Numbers
+  "S101",    # Forbid assert statements
+  "D103",    # Require docstring in public function
+  "D104",    # Require docstring in public package
+  "PLR2004", # Forbid Magic Numbers
 ]
 
 [lint.per-file-ignores]

examples/src/consumer.py

@@ -12,6 +12,13 @@
 [`User`][examples.src.consumer.User] class and the consumer fetches a user's
 information from a HTTP endpoint.
 
+This also showcases how Pact tests differ from merely testing adherence to an
+OpenAPI specification. The Pact tests are more concerned with the practical use
+of the API, rather than the formally defined specification. So you will see
+below that as far as this consumer is concerned, the only information needed
+from the provider is the user's ID, name, and creation date. This is despite the
+provider having additional fields in the response.
+
 Note that the code in this module is agnostic of Pact. The `pact-python`
 dependency only appears in the tests. This is because the consumer is not
 concerned with Pact, only the tests are.
@@ -21,7 +28,7 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Dict, Tuple
+from typing import Any, Dict
 
 import requests
 
@@ -104,42 +111,45 @@ def get_user(self, user_id: int) -> User:
         )
 
     def create_user(
-        self, user: Dict[str, Any], header: Dict[str, str]
-    ) -> Tuple[int, User]:
+        self,
+        *,
+        name: str,
+    ) -> User:
         """
         Create a new user on the server.
 
         Args:
-            user: The user data to create.
-            header: The headers to send with the request.
+            name: The name of the user to create.
 
         Returns:
-            The user data including the ID assigned by the server; Error if user exists.
+            The user, if successfully created.
+
+        Raises:
+            requests.HTTPError: If the server returns a non-200 response.
         """
         uri = f"{self.base_uri}/users/"
-        response = requests.post(uri, headers=header, json=user, timeout=5)
+        response = requests.post(uri, json={"name": name}, timeout=5)
         response.raise_for_status()
         data: Dict[str, Any] = response.json()
-        return (
-            response.status_code,
-            User(
-                id=data["id"],
-                name=data["name"],
-                created_on=datetime.fromisoformat(data["created_on"]),
-            ),
+        return User(
+            id=data["id"],
+            name=data["name"],
+            created_on=datetime.fromisoformat(data["created_on"]),
         )
 
-    def delete_user(self, user_id: int) -> int:
+    def delete_user(self, uid: int | User) -> None:
         """
         Delete a user by ID from the server.
 
         Args:
-            user_id: The ID of the user to delete.
+            uid: The user ID or user object to delete.
 
-        Returns:
-            The response status code.
+        Raises:
+            requests.HTTPError: If the server returns a non-200 response.
         """
-        uri = f"{self.base_uri}/users/{user_id}"
+        if isinstance(uid, User):
+            uid = uid.id
+
+        uri = f"{self.base_uri}/users/{uid}"
         response = requests.delete(uri, timeout=5)
         response.raise_for_status()
-        return response.status_code

examples/src/fastapi.py

@@ -12,6 +12,14 @@
 (the consumer) and returns a response. In this example, we have a simple
 endpoint which returns a user's information from a (fake) database.
 
+This also showcases how Pact tests differ from merely testing adherence to an
+OpenAPI specification. The Pact tests are more concerned with the practical use
+of the API, rather than the formally defined specification. The User class
+defined here has additional fields which are not used by the consumer. Should
+the provider later decide to add or remove fields, Pact's consumer-driven
+testing will provide feedback on whether the consumer is compatible with the
+provider's changes.
+
 Note that the code in this module is agnostic of Pact. The `pact-python`
 dependency only appears in the tests. This is because the consumer is not
 concerned with Pact, only the tests are.
@@ -20,28 +28,51 @@
 from __future__ import annotations
 
 import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime
 from typing import Any, Dict
 
-from pydantic import BaseModel
-
 from fastapi import FastAPI, HTTPException
-from fastapi.responses import JSONResponse
 
 app = FastAPI()
 logger = logging.getLogger(__name__)
 
 
-class User(BaseModel):
-    """
-    User data class.
-
-    This class is used to represent a user in the application. It is used to
-    validate the incoming data and to dump the data to a dictionary.
-    """
+@dataclass()
+class User:
+    """User data class."""
 
-    id: int | None = None
+    id: int
     name: str
-    email: str
+    created_on: datetime
+    email: str | None
+    ip_address: str | None
+    hobbies: list[str]
+    admin: bool
+
+    def __post_init__(self) -> None:
+        """
+        Validate the User data.
+
+        This performs the following checks:
+
+        - The name cannot be empty
+        - The id must be a positive integer
+
+        Raises:
+            ValueError: If any of the above checks fail.
+        """
+        if not self.name:
+            msg = "User must have a name"
+            raise ValueError(msg)
+
+        if self.id < 0:
+            msg = "User ID must be a positive integer"
+            raise ValueError(msg)
+
+    def __repr__(self) -> str:
+        """Return the user's name."""
+        return f"User({self.id}:{self.name})"
 
 
 """
@@ -52,11 +83,11 @@ class User(BaseModel):
 be mocked out to avoid the need for a real database. An example of this can be
 found in the [test suite][examples.tests.test_01_provider_fastapi].
 """
-FAKE_DB: Dict[int, Dict[str, Any]] = {}
+FAKE_DB: Dict[int, User] = {}
 
 
 @app.get("/users/{uid}")
-async def get_user_by_id(uid: int) -> JSONResponse:
+async def get_user_by_id(uid: int) -> User:
     """
     Fetch a user by their ID.
 
@@ -68,12 +99,12 @@ async def get_user_by_id(uid: int) -> JSONResponse:
     """
     user = FAKE_DB.get(uid)
     if not user:
-        return JSONResponse(status_code=404, content={"error": "User not found"})
-    return JSONResponse(status_code=200, content=user)
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
 
 
 @app.post("/users/")
-async def create_new_user(user: User) -> JSONResponse:
+async def create_new_user(user: dict[str, Any]) -> User:
     """
     Create a new user .
 
@@ -83,26 +114,33 @@ async def create_new_user(user: User) -> JSONResponse:
     Returns:
         The status code 200 and user data if successfully created, HTTP 404 if not
     """
-    if user.id is not None:
+    if "id" in user:
         raise HTTPException(status_code=400, detail="ID should not be provided.")
-    new_uid = len(FAKE_DB)
-    FAKE_DB[new_uid] = user.model_dump()
-
-    return JSONResponse(status_code=200, content=FAKE_DB[new_uid])
-
-
-@app.delete("/users/{user_id}", status_code=204)
-async def delete_user(user_id: int):  # noqa: ANN201
+    uid = len(FAKE_DB)
+    FAKE_DB[uid] = User(
+        id=uid,
+        name=user["name"],
+        created_on=datetime.now(tz=UTC),
+        email=user.get("email"),
+        ip_address=user.get("ip_address"),
+        hobbies=user.get("hobbies", []),
+        admin=user.get("admin", False),
+    )
+    return FAKE_DB[uid]
+
+
+@app.delete("/users/{uid}", status_code=204)
+async def delete_user(uid: int):  # noqa: ANN201
     """
      Delete an existing user .
 
     Args:
-        user_id: The ID of the user to delete
+        uid: The ID of the user to delete
 
     Returns:
         The status code 204, HTTP 404 if not
     """
-    if user_id not in FAKE_DB:
+    if uid not in FAKE_DB:
         raise HTTPException(status_code=404, detail="User not found")
 
-    del FAKE_DB[user_id]
+    del FAKE_DB[uid]

examples/src/flask.py

@@ -20,13 +20,67 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, Tuple, Union
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Dict, Tuple
 
 from flask import Flask, Response, abort, jsonify, request
 
 logger = logging.getLogger(__name__)
-
 app = Flask(__name__)
+
+
+@dataclass()
+class User:
+    """User data class."""
+
+    id: int
+    name: str
+    created_on: datetime
+    email: str | None
+    ip_address: str | None
+    hobbies: list[str]
+    admin: bool
+
+    def __post_init__(self) -> None:
+        """
+        Validate the User data.
+
+        This performs the following checks:
+
+        - The name cannot be empty
+        - The id must be a positive integer
+
+        Raises:
+            ValueError: If any of the above checks fail.
+        """
+        if not self.name:
+            msg = "User must have a name"
+            raise ValueError(msg)
+
+        if self.id < 0:
+            msg = "User ID must be a positive integer"
+            raise ValueError(msg)
+
+    def __repr__(self) -> str:
+        """Return the user's name."""
+        return f"User({self.id}:{self.name})"
+
+    def dict(self) -> dict[str, Any]:
+        """
+        Return the user's data as a dict.
+        """
+        return {
+            "id": self.id,
+            "name": self.name,
+            "created_on": self.created_on.isoformat(),
+            "email": self.email,
+            "ip_address": self.ip_address,
+            "hobbies": self.hobbies,
+            "admin": self.admin,
+        }
+
+
 """
 As this is a simple example, we'll use a simple dict to represent a database.
 This would be replaced with a real database in a real application.
@@ -35,11 +89,11 @@
 be mocked out to avoid the need for a real database. An example of this can be
 found in the [test suite][examples.tests.test_01_provider_flask].
 """
-FAKE_DB: Dict[int, Dict[str, Any]] = {}
+FAKE_DB: Dict[int, User] = {}
 
 
-@app.route("/users/<uid>")
-def get_user_by_id(uid: int) -> Union[Dict[str, Any], tuple[Dict[str, Any], int]]:
+@app.route("/users/<int:uid>")
+def get_user_by_id(uid: int) -> Response | Tuple[Response, int]:
     """
     Fetch a user by their ID.
 
@@ -49,29 +103,34 @@ def get_user_by_id(uid: int) -> Union[Dict[str, Any], tuple[Dict[str, Any], int]
     Returns:
         The user data if found, HTTP 404 if not
     """
-    user = FAKE_DB.get(int(uid))
+    user = FAKE_DB.get(uid)
     if not user:
-        return {"error": "User not found"}, 404
-    return user
+        return jsonify({"detail": "User not found"}), 404
+    return jsonify(user.dict())
 
 
 @app.route("/users/", methods=["POST"])
-def create_user() -> Tuple[Response, int]:
+def create_user() -> Response:
     if request.json is None:
         abort(400, description="Invalid JSON data")
 
-    data: Dict[str, Any] = request.json
-    new_uid: int = len(FAKE_DB)
-    if new_uid in FAKE_DB:
-        abort(400, description="User already exists")
-
-    FAKE_DB[new_uid] = {"id": new_uid, "name": data["name"], "email": data["email"]}
-    return jsonify(FAKE_DB[new_uid]), 200
-
-
-@app.route("/users/<user_id>", methods=["DELETE"])
-def delete_user(user_id: int) -> Tuple[str, int]:
-    if user_id not in FAKE_DB:
-        abort(404, description="User not found")
-    del FAKE_DB[user_id]
-    return "", 204  # No Content status code
+    user: Dict[str, Any] = request.json
+    uid = len(FAKE_DB)
+    FAKE_DB[uid] = User(
+        id=uid,
+        name=user["name"],
+        created_on=datetime.now(tz=UTC),
+        email=user.get("email"),
+        ip_address=user.get("ip_address"),
+        hobbies=user.get("hobbies", []),
+        admin=user.get("admin", False),
+    )
+    return jsonify(FAKE_DB[uid].dict())
+
+
+@app.route("/users/<int:uid>", methods=["DELETE"])
+def delete_user(uid: int) -> Tuple[str | Response, int]:
+    if uid not in FAKE_DB:
+        return jsonify({"detail": "User not found"}), 404
+    del FAKE_DB[uid]
+    return "", 204