chore: add https methods; updated v2 examples

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

@@ -24,18 +24,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 +88,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 +152,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": "[email protected]",
+        "created_on": Format().iso_8601_datetime(),
+    }
+    header = {"Content-Type": "application/json"}
+    payload = {
+        "name": "Jane Doe",
+        "email": "[email protected]",
+        "created_on": "1991-02-20T06:35:26+00:00",
+    }
+    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()

examples/tests/test_01_provider_fastapi.py

@@ -35,7 +35,7 @@
 from yarl import URL
 
 from examples.src.fastapi import app
-from pact import Verifier
+from pact import Verifier  # type: ignore[import-untyped]
 
 PROVIDER_URL = URL("http://localhost:8080")
 
@@ -68,6 +68,8 @@ async def mock_pact_provider_states(
     mapping = {
         "user 123 doesn't exist": mock_user_123_doesnt_exist,
         "user 123 exists": mock_user_123_exists,
+        "create user 124": mock_post_request_to_create_user,
+        "delete the user 124": mock_delete_request_to_delete_user,
     }
     return {"result": mapping[state.state]()}
 
@@ -134,6 +136,61 @@ def mock_user_123_exists() -> None:
     }
 
 
+def mock_post_request_to_create_user() -> None:
+    """
+    Mock the database for the post request to create a user.
+    """
+    import examples.src.fastapi
+
+    examples.src.fastapi.FAKE_DB = MagicMock()
+    examples.src.fastapi.FAKE_DB.__len__.return_value = 124
+    examples.src.fastapi.FAKE_DB.__setitem__.return_value = None
+    examples.src.fastapi.FAKE_DB.__getitem__.return_value = {
+        "id": 124,
+        "created_on": "1991-02-20T06:35:26+00:00",
+        "email": "[email protected]",
+        "name": "Jane Doe",
+        "ip_address": "10.1.2.3",
+        "hobbies": ["hiking", "swimming"],
+        "admin": False,
+    }
+
+
+def mock_delete_request_to_delete_user() -> None:
+    """
+    Mock the database for the delete request to delete a user.
+    """
+    import examples.src.fastapi
+
+    db_values = {
+        123: {
+            "id": 123,
+            "name": "Verna Hampton",
+            "email": "[email protected]",
+            "created_on": "1991-02-20T06:35:26+00:00",
+            "ip_address": "10.1.2.3",
+            "hobbies": ["hiking", "swimming"],
+            "admin": False,
+        },
+        124: {
+            "id": 124,
+            "name": "Jane Doe",
+            "email": "[email protected]",
+            "created_on": "1991-02-20T06:35:26+00:00",
+            "ip_address": "10.1.2.5",
+            "hobbies": ["running", "dancing"],
+            "admin": False,
+        },
+    }
+
+    examples.src.fastapi.FAKE_DB = MagicMock()
+    examples.src.fastapi.FAKE_DB.__delitem__.side_effect = (
+        lambda key: db_values.__delitem__(key)
+    )
+    examples.src.fastapi.FAKE_DB.__getitem__.side_effect = lambda key: db_values[key]
+    examples.src.fastapi.FAKE_DB.__contains__.side_effect = lambda key: key in db_values
+
+
 def test_against_broker(broker: URL, verifier: Verifier) -> None:
     """
     Test the provider against the broker.