feat(examples): add post and delete

Extend the existing examples to showcase both HTTP POST and DELETE
requests, and how they are handled in Pact. Specifically showcasing how
the verifying can ensure that any side-effects have taken place.

Co-authored-by: Amit Singh <amit.828.as@gmail.com>
Signed-off-by: JP-Ellis <josh@jpellis.me>

Author: amit828as

.pre-commit-config.yaml

@@ -81,5 +81,5 @@ repos:
         entry: hatch run mypy
         language: system
         types: [python]
-        exclude: ^(src/pact|tests)/(?!v3/).*\.py$
+        exclude: ^(src/pact|tests|examples/tests)/(?!v3/).*\.py$
         stages: [pre-push]

examples/.ruff.toml

@@ -5,6 +5,7 @@ ignore = [
   "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

@@ -21,7 +21,7 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Dict
+from typing import Any, Dict, Tuple
 
 import requests
 
@@ -102,3 +102,44 @@ def get_user(self, user_id: int) -> User:
             name=data["name"],
             created_on=datetime.fromisoformat(data["created_on"]),
         )
+
+    def create_user(
+        self, user: Dict[str, Any], header: Dict[str, str]
+    ) -> Tuple[int, User]:
+        """
+        Create a new user on the server.
+
+        Args:
+            user: The user data to create.
+            header: The headers to send with the request.
+
+        Returns:
+            The user data including the ID assigned by the server; Error if user exists.
+        """
+        uri = f"{self.base_uri}/users/"
+        response = requests.post(uri, headers=header, json=user, 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"]),
+            ),
+        )
+
+    def delete_user(self, user_id: int) -> int:
+        """
+        Delete a user by ID from the server.
+
+        Args:
+            user_id: The ID of the user to delete.
+
+        Returns:
+            The response status code.
+        """
+        uri = f"{self.base_uri}/users/{user_id}"
+        response = requests.delete(uri, timeout=5)
+        response.raise_for_status()
+        return response.status_code

examples/src/fastapi.py

@@ -19,12 +19,30 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Any, Dict
 
-from fastapi import FastAPI
+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.
+    """
+
+    id: int | None = None
+    name: str
+    email: str
+
 
 """
 As this is a simple example, we'll use a simple dict to represent a database.
@@ -52,3 +70,39 @@ async def get_user_by_id(uid: int) -> JSONResponse:
     if not user:
         return JSONResponse(status_code=404, content={"error": "User not found"})
     return JSONResponse(status_code=200, content=user)
+
+
+@app.post("/users/")
+async def create_new_user(user: User) -> JSONResponse:
+    """
+    Create a new user .
+
+    Args:
+        user: The user data to create
+
+    Returns:
+        The status code 200 and user data if successfully created, HTTP 404 if not
+    """
+    if user.id is not None:
+        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
+    """
+     Delete an existing user .
+
+    Args:
+        user_id: The ID of the user to delete
+
+    Returns:
+        The status code 204, HTTP 404 if not
+    """
+    if user_id not in FAKE_DB:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    del FAKE_DB[user_id]

examples/src/flask.py

@@ -19,12 +19,14 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Union
+import logging
+from typing import Any, Dict, Tuple, Union
 
-from flask import Flask
+from flask import Flask, Response, abort, jsonify, request
 
-app = Flask(__name__)
+logger = logging.getLogger(__name__)
 
+app = Flask(__name__)
 """
 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.
@@ -47,7 +49,29 @@ 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(uid)
+    user = FAKE_DB.get(int(uid))
     if not user:
         return {"error": "User not found"}, 404
     return user
+
+
+@app.route("/users/", methods=["POST"])
+def create_user() -> Tuple[Response, int]:
+    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

examples/tests/test_00_consumer.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import logging
+from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, Dict, Generator
 
@@ -24,18 +25,28 @@
 from yarl import URL
 
 from examples.src.consumer import User, UserConsumer
-from pact import Consumer, Format, Like, Provider
+from pact import Consumer, Format, Like, Provider  # type: ignore[attr-defined]
 
 if TYPE_CHECKING:
     from pathlib import Path
 
-    from pact.pact import Pact
+    from pact.pact import Pact  # type: ignore[import-untyped]
 
-log = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 MOCK_URL = URL("http://localhost:8080")
 
 
+@pytest.fixture(scope="session", autouse=True)
+def _setup_pact_logging() -> None:
+    """
+    Set up logging for the pact package.
+    """
+    from pact.v3 import ffi
+
+    ffi.log_to_stderr("INFO")
+
+
 @pytest.fixture
 def user_consumer() -> UserConsumer:
     """
@@ -78,7 +89,7 @@ def pact(broker: URL, pact_dir: Path) -> Generator[Pact, Any, None]:
     pact = consumer.has_pact_with(
         Provider("UserProvider"),
         pact_dir=pact_dir,
-        publish_to_broker=True,
+        publish_to_broker=False,
         # Mock service configuration
         host_name=MOCK_URL.host,
         port=MOCK_URL.port,
@@ -142,3 +153,66 @@ def test_get_unknown_user(pact: Pact, user_consumer: UserConsumer) -> None:
         assert excinfo.value.response is not None
         assert excinfo.value.response.status_code == HTTPStatus.NOT_FOUND
         pact.verify()
+
+
+def test_post_request_to_create_user(pact: Pact, user_consumer: UserConsumer) -> None:
+    """
+    Test the POST request for creating a new user.
+
+    This test defines the expected interaction for a POST request to create
+    a new user. It sets up the expected request and response from the provider,
+    including the request body and headers, and verifies that the response
+    status code is 200 and the response body matches the expected user data.
+    """
+    expected: Dict[str, Any] = {
+        "id": 124,
+        "name": "Jane Doe",
+        "email": "jane@example.com",
+        "created_on": Format().iso_8601_datetime(),
+    }
+    header = {"Content-Type": "application/json"}
+    payload: dict[str, str] = {
+        "name": "Jane Doe",
+        "email": "jane@example.com",
+        "created_on": (datetime.now(tz=UTC) - timedelta(days=318)).isoformat(),
+    }
+    expected_response_code: int = 200
+
+    (
+        pact.given("create user 124")
+        .upon_receiving("A request to create a new user")
+        .with_request(method="POST", path="/users/", headers=header, body=payload)
+        .will_respond_with(status=200, headers=header, body=Like(expected))
+    )
+
+    with pact:
+        response = user_consumer.create_user(user=payload, header=header)
+        assert response[0] == expected_response_code
+        assert response[1].id == 124
+        assert response[1].name == "Jane Doe"
+
+        pact.verify()
+
+
+def test_delete_request_to_delete_user(pact: Pact, user_consumer: UserConsumer) -> None:
+    """
+    Test the DELETE request for deleting a user.
+
+    This test defines the expected interaction for a DELETE request to delete
+    a user. It sets up the expected request and response from the provider,
+    including the request body and headers, and verifies that the response
+    status code is 200 and the response body matches the expected user data.
+    """
+    expected_response_code: int = 204
+    (
+        pact.given("delete the user 124")
+        .upon_receiving("a request for deleting user")
+        .with_request(method="DELETE", path="/users/124")
+        .will_respond_with(204)
+    )
+
+    with pact:
+        response_status_code = user_consumer.delete_user(124)
+        assert response_status_code == expected_response_code
+
+        pact.verify()