diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b961ce17f..def7e7222 100644 --- a/.pre-commit-config.yaml +++ b/.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] diff --git a/examples/.ruff.toml b/examples/.ruff.toml index 7e5b439eb..6b2ed4c15 100644 --- a/examples/.ruff.toml +++ b/examples/.ruff.toml @@ -2,9 +2,10 @@ extend = "../pyproject.toml" [lint] ignore = [ - "S101", # Forbid assert statements - "D103", # Require docstring in public function - "D104", # Require docstring in public package + "S101", # Forbid assert statements + "D103", # Require docstring in public function + "D104", # Require docstring in public package + "PLR2004", # Forbid Magic Numbers ] [lint.per-file-ignores] diff --git a/examples/src/consumer.py b/examples/src/consumer.py index fa260d214..453297923 100644 --- a/examples/src/consumer.py +++ b/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. @@ -102,3 +109,47 @@ def get_user(self, user_id: int) -> User: name=data["name"], created_on=datetime.fromisoformat(data["created_on"]), ) + + def create_user( + self, + *, + name: str, + ) -> User: + """ + Create a new user on the server. + + Args: + name: The name of the user to create. + + Returns: + 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, json={"name": name}, timeout=5) + response.raise_for_status() + data: Dict[str, Any] = response.json() + return User( + id=data["id"], + name=data["name"], + created_on=datetime.fromisoformat(data["created_on"]), + ) + + def delete_user(self, uid: int | User) -> None: + """ + Delete a user by ID from the server. + + Args: + uid: The user ID or user object to delete. + + Raises: + requests.HTTPError: If the server returns a non-200 response. + """ + if isinstance(uid, User): + uid = uid.id + + uri = f"{self.base_uri}/users/{uid}" + response = requests.delete(uri, timeout=5) + response.raise_for_status() diff --git a/examples/src/fastapi.py b/examples/src/fastapi.py index b25d3c1a8..013e95f13 100644 --- a/examples/src/fastapi.py +++ b/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. @@ -19,12 +27,53 @@ from __future__ import annotations +import logging +from dataclasses import dataclass +from datetime import UTC, datetime from typing import Any, Dict -from fastapi import FastAPI -from fastapi.responses import JSONResponse +from fastapi import FastAPI, HTTPException app = FastAPI() +logger = logging.getLogger(__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})" + """ As this is a simple example, we'll use a simple dict to represent a database. @@ -34,11 +83,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_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. @@ -50,5 +99,48 @@ 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: dict[str, Any]) -> User: + """ + 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 "id" in user: + raise HTTPException(status_code=400, detail="ID should not be provided.") + 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: + uid: The ID of the user to delete + + Returns: + The status code 204, HTTP 404 if not + """ + if uid not in FAKE_DB: + raise HTTPException(status_code=404, detail="User not found") + + del FAKE_DB[uid] diff --git a/examples/src/flask.py b/examples/src/flask.py index 31d614042..98107e3c1 100644 --- a/examples/src/flask.py +++ b/examples/src/flask.py @@ -19,12 +19,68 @@ from __future__ import annotations -from typing import Any, Dict, Union +import logging +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any, Dict, Tuple -from flask import Flask +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. @@ -33,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/") -def get_user_by_id(uid: int) -> Union[Dict[str, Any], tuple[Dict[str, Any], int]]: +@app.route("/users/") +def get_user_by_id(uid: int) -> Response | Tuple[Response, int]: """ Fetch a user by their ID. @@ -49,5 +105,32 @@ def get_user_by_id(uid: int) -> Union[Dict[str, Any], tuple[Dict[str, Any], int] """ 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() -> Response: + if request.json is None: + abort(400, description="Invalid JSON data") + + 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/", 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 diff --git a/examples/tests/test_00_consumer.py b/examples/tests/test_00_consumer.py index 907678508..273b4405d 100644 --- a/examples/tests/test_00_consumer.py +++ b/examples/tests/test_00_consumer.py @@ -31,7 +31,7 @@ from pact.pact import Pact -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) MOCK_URL = URL("http://localhost:8080") @@ -127,7 +127,7 @@ def test_get_existing_user(pact: Pact, user_consumer: UserConsumer) -> None: def test_get_unknown_user(pact: Pact, user_consumer: UserConsumer) -> None: - expected = {"error": "User not found"} + expected = {"detail": "User not found"} ( pact.given("user 123 doesn't exist") @@ -142,3 +142,65 @@ 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_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. + """ + body = {"name": "Verna Hampton"} + expected_response: Dict[str, Any] = { + "id": 124, + "name": "Verna Hampton", + "created_on": Format().iso_8601_datetime(), + } + + ( + pact.given("create user 124") + .upon_receiving("A request to create a new user") + .with_request( + method="POST", + path="/users/", + body=body, + headers={"Content-Type": "application/json"}, + ) + .will_respond_with( + status=200, + body=Like(expected_response), + ) + ) + + with pact: + user = user_consumer.create_user(name="Verna Hampton") + assert user.id > 0 + assert user.name == "Verna Hampton" + assert user.created_on + + 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. + """ + ( + pact.given("delete the user 124") + .upon_receiving("a request for deleting user") + .with_request(method="DELETE", path="/users/124") + .will_respond_with(status=204) + ) + + with pact: + user_consumer.delete_user(124) + + pact.verify() diff --git a/examples/tests/test_01_provider_fastapi.py b/examples/tests/test_01_provider_fastapi.py index a95b5b5f8..0825b1c4c 100644 --- a/examples/tests/test_01_provider_fastapi.py +++ b/examples/tests/test_01_provider_fastapi.py @@ -25,6 +25,7 @@ from __future__ import annotations import time +from datetime import UTC, datetime from multiprocessing import Process from typing import Any, Dict, Generator, Union from unittest.mock import MagicMock @@ -34,8 +35,8 @@ from pydantic import BaseModel from yarl import URL -from examples.src.fastapi import app -from pact import Verifier +from examples.src.fastapi import User, app +from pact import Verifier # type: ignore[import-untyped] PROVIDER_URL = URL("http://localhost:8080") @@ -68,8 +69,11 @@ 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]()} + mapping[state.state]() + return {"result": f"{state} set"} def run_server() -> None: @@ -123,16 +127,78 @@ def mock_user_123_exists() -> None: """ import examples.src.fastapi - examples.src.fastapi.FAKE_DB = MagicMock() - examples.src.fastapi.FAKE_DB.get.return_value = { - "id": 123, - "name": "Verna Hampton", - "created_on": "2016-12-15T20:16:01", - "ip_address": "10.1.2.3", - "hobbies": ["hiking", "swimming"], - "admin": False, + mock_db = MagicMock() + mock_db.get.return_value = User( + id=123, + name="Verna Hampton", + email="verna@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.3", + hobbies=["hiking", "swimming"], + admin=False, + ) + examples.src.fastapi.FAKE_DB = mock_db + + +def mock_post_request_to_create_user() -> None: + """ + Mock the database for the post request to create a user. + """ + import examples.src.fastapi + + local_db: Dict[int, User] = {} + + def local_setitem(key: int, value: User) -> None: + local_db[key] = value + + def local_getitem(key: int) -> User: + return local_db[key] + + mock_db = MagicMock() + mock_db.__len__.return_value = 124 + mock_db.__setitem__.side_effect = local_setitem + mock_db.__getitem__.side_effect = local_getitem + examples.src.fastapi.FAKE_DB = mock_db + + +def mock_delete_request_to_delete_user() -> None: + """ + Mock the database for the delete request to delete a user. + """ + import examples.src.fastapi + + local_db = { + 123: User( + id=123, + name="Verna Hampton", + email="verna@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.3", + hobbies=["hiking", "swimming"], + admin=False, + ), + 124: User( + id=124, + name="Jane Doe", + email="jane@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.5", + hobbies=["running", "dancing"], + admin=False, + ), } + def local_delitem(key: int) -> None: + del local_db[key] + + def local_contains(key: int) -> bool: + return key in local_db + + mock_db = MagicMock() + mock_db.__delitem__.side_effect = local_delitem + mock_db.__contains__.side_effect = local_contains + examples.src.fastapi.FAKE_DB = mock_db + def test_against_broker(broker: URL, verifier: Verifier) -> None: """ diff --git a/examples/tests/test_01_provider_flask.py b/examples/tests/test_01_provider_flask.py index 58cb02458..a26c1bdbe 100644 --- a/examples/tests/test_01_provider_flask.py +++ b/examples/tests/test_01_provider_flask.py @@ -25,6 +25,7 @@ from __future__ import annotations import time +from datetime import UTC, datetime from multiprocessing import Process from typing import Any, Dict, Generator, Union from unittest.mock import MagicMock @@ -32,9 +33,9 @@ import pytest from yarl import URL -from examples.src.flask import app +from examples.src.flask import User, app from flask import request -from pact import Verifier +from pact import Verifier # type: ignore[import-untyped] PROVIDER_URL = URL("http://localhost:8080") @@ -55,11 +56,17 @@ async def mock_pact_provider_states() -> Dict[str, Union[str, None]]: endpoint is called by Pact before each test to ensure that the provider is in the correct state. """ + if request.json is None: + msg = "Request must be JSON" + raise ValueError(msg) + state: str = request.json["state"] 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[request.json["state"]]()} # type: ignore[index] + return {"result": mapping[state]()} # type: ignore[index] def run_server() -> None: @@ -70,7 +77,10 @@ def run_server() -> None: lambda cannot be used as the target of a `multiprocessing.Process` as it cannot be pickled. """ - app.run(host=PROVIDER_URL.host, port=PROVIDER_URL.port) + app.run( + host=PROVIDER_URL.host, + port=PROVIDER_URL.port, + ) @pytest.fixture(scope="module") @@ -112,15 +122,77 @@ def mock_user_123_exists() -> None: import examples.src.flask examples.src.flask.FAKE_DB = MagicMock() - examples.src.flask.FAKE_DB.get.return_value = { - "id": 123, - "name": "Verna Hampton", - "created_on": "2016-12-15T20:16:01", - "ip_address": "10.1.2.3", - "hobbies": ["hiking", "swimming"], - "admin": False, + examples.src.flask.FAKE_DB.get.return_value = User( + id=123, + name="Verna Hampton", + email="verna@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.3", + hobbies=["hiking", "swimming"], + admin=False, + ) + + +def mock_post_request_to_create_user() -> None: + """ + Mock the database for the post request to create a user. + """ + import examples.src.flask + + local_db: Dict[int, User] = {} + + def local_setitem(key: int, value: User) -> None: + local_db[key] = value + + def local_getitem(key: int) -> User: + return local_db[key] + + mock_db = MagicMock() + mock_db.__len__.return_value = 124 + mock_db.__setitem__.side_effect = local_setitem + mock_db.__getitem__.side_effect = local_getitem + examples.src.flask.FAKE_DB = mock_db + + +def mock_delete_request_to_delete_user() -> None: + """ + Mock the database for the delete request to delete a user. + """ + import examples.src.flask + + local_db = { + 123: User( + id=123, + name="Verna Hampton", + email="verna@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.3", + hobbies=["hiking", "swimming"], + admin=False, + ), + 124: User( + id=124, + name="Jane Doe", + email="jane@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.5", + hobbies=["running", "dancing"], + admin=False, + ), } + def local_delitem(key: int) -> None: + del local_db[key] + + def local_contains(key: int) -> bool: + return key in local_db + + mock_db = MagicMock() + mock_db.__delitem__.side_effect = local_delitem + mock_db.__contains__.side_effect = local_contains + mock_db.is_mocked = True + examples.src.flask.FAKE_DB = mock_db + def test_against_broker(broker: URL, verifier: Verifier) -> None: """ diff --git a/examples/tests/v3/conftest.py b/examples/tests/v3/conftest.py new file mode 100644 index 000000000..485ff82bb --- /dev/null +++ b/examples/tests/v3/conftest.py @@ -0,0 +1,15 @@ +""" +Common Pytest configuration for the V3 examples. +""" + +import pytest + + +@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") diff --git a/examples/tests/v3/test_00_consumer.py b/examples/tests/v3/test_00_consumer.py index 634ce7ffa..2807d1fc4 100644 --- a/examples/tests/v3/test_00_consumer.py +++ b/examples/tests/v3/test_00_consumer.py @@ -25,6 +25,7 @@ import pytest import requests +from examples.src.consumer import UserConsumer from pact.v3 import Pact @@ -71,7 +72,6 @@ def test_get_existing_user(pact: Pact) -> None: code as shown in [`test_get_non_existent_user`](#test_get_non_existent_user). """ - expected_response_code = 200 expected: Dict[str, Any] = { "id": 123, "name": "Verna Hampton", @@ -96,11 +96,10 @@ def test_get_existing_user(pact: Pact) -> None: ) with pact.serve() as srv: - response = requests.get(f"{srv.url}/users/123", timeout=5) - - assert response.status_code == expected_response_code - assert expected["name"] == "Verna Hampton" - datetime.fromisoformat(expected["created_on"]["value"]) + client = UserConsumer(str(srv.url)) + user = client.get_user(123) + assert user.id == 123 + assert user.name == "Verna Hampton" def test_get_non_existent_user(pact: Pact) -> None: @@ -131,3 +130,67 @@ def test_get_non_existent_user(pact: Pact) -> None: response = requests.get(f"{srv.url}/users/2", timeout=5) assert response.status_code == expected_response_code + + +def test_create_user(pact: Pact) -> 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. + """ + body = {"name": "Verna Hampton"} + expected_response: Dict[str, Any] = { + "id": 124, + "name": "Verna Hampton", + "created_on": { + # This structure is using the Integration JSON format as described + # in the link below. The preview of V3 currently does not have + # built-in support for matchers and generators, though this is on + # the roadmap and will be available before the final release. + # + # + "pact:matcher:type": "regex", + "regex": r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}(Z|(\+|-)\d{2}:\d{2})", + "value": datetime.now(tz=timezone.utc).isoformat(), + }, + } + + ( + pact.upon_receiving("a request to create a new user") + .given("the specified user doesn't exist") + .with_request(method="POST", path="/users/") + .with_body(json.dumps(body), content_type="application/json") + .will_respond_with(status=200) + .with_body(content_type="application/json", body=json.dumps(expected_response)) + ) + + with pact.serve() as srv: + client = UserConsumer(str(srv.url)) + user = client.create_user(name="Verna Hampton") + assert user.id > 0 + assert user.name == "Verna Hampton" + assert user.created_on + + +def test_delete_request_to_delete_user(pact: Pact) -> 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. + """ + ( + pact.upon_receiving("a request for deleting user") + .given("user is present in DB") + .with_request(method="DELETE", path="/users/124") + .will_respond_with(204) + ) + + with pact.serve() as srv: + client = UserConsumer(str(srv.url)) + client.delete_user(124) diff --git a/examples/tests/v3/test_01_fastapi_provider.py b/examples/tests/v3/test_01_fastapi_provider.py index f77cc071f..c998488f6 100644 --- a/examples/tests/v3/test_01_fastapi_provider.py +++ b/examples/tests/v3/test_01_fastapi_provider.py @@ -27,7 +27,7 @@ from __future__ import annotations import time -from datetime import datetime, timezone +from datetime import UTC, datetime from multiprocessing import Process from typing import TYPE_CHECKING, Callable, Dict, Literal from unittest.mock import MagicMock @@ -35,7 +35,7 @@ import uvicorn from yarl import URL -from examples.src.fastapi import app +from examples.src.fastapi import User, app from pact.v3 import Verifier PROVIDER_URL = URL("http://localhost:8000") @@ -83,10 +83,14 @@ async def mock_pact_provider_states( mapping["setup"] = { "user doesn't exists": mock_user_doesnt_exist, "user exists": mock_user_exists, + "the specified user doesn't exist": mock_post_request_to_create_user, + "user is present in DB": mock_delete_request_to_delete_user, } mapping["teardown"] = { "user doesn't exists": verify_user_doesnt_exist_mock, "user exists": verify_user_exists_mock, + "the specified user doesn't exist": verify_mock_post_request_to_create_user, + "user is present in DB": verify_mock_delete_request_to_delete_user, } mapping[action][state]() @@ -192,14 +196,91 @@ def mock_user_exists() -> None: import examples.src.fastapi mock_db = MagicMock() - mock_db.get.return_value = { - "id": 123, - "name": "Verna Hampton", - "created_on": datetime.now(tz=timezone.utc).isoformat(), - "ip_address": "10.1.2.3", - "hobbies": ["hiking", "swimming"], - "admin": False, + mock_db.get.return_value = User( + id=123, + name="Verna Hampton", + email="verna@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.3", + hobbies=["hiking", "swimming"], + admin=False, + ) + examples.src.fastapi.FAKE_DB = mock_db + + +def mock_post_request_to_create_user() -> None: + """ + Mock the database for the post request to create a user. + + While the `FAKE_DB` is a dictionary in this example, one should imagine that + this is a real database. In this instance, we are replacing the calls to the + database with a local dictionary to avoid side effects; thereby eliminating + the need to stand up a real database for the tests. + + The added benefit of using this approach is that the mock can subsequently + be inspected to ensure that the correct calls were made to the database. For + example, asserting that the correct user ID was retrieved from the database. + These checks are performed as part of the `teardown` action. This action can + also be used to reset the mock, or in the case were a real database is used, + to clean up any side effects. + """ + import examples.src.fastapi + + local_db: Dict[int, User] = {} + + def local_setitem(key: int, value: User) -> None: + local_db[key] = value + + def local_getitem(key: int) -> User: + return local_db[key] + + mock_db = MagicMock() + mock_db.__len__.return_value = 124 + mock_db.__setitem__.side_effect = local_setitem + mock_db.__getitem__.side_effect = local_getitem + examples.src.fastapi.FAKE_DB = mock_db + + +def mock_delete_request_to_delete_user() -> None: + """ + Mock the database for the delete request to delete a user. + + As with the `mock_post_request_to_create_user` function, we are using a + local dictionary to avoid side effects. This function replaces the calls to + the database with a local dictionary to avoid side effects. + """ + import examples.src.fastapi + + local_db = { + 123: User( + id=123, + name="Verna Hampton", + email="verna@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.3", + hobbies=["hiking", "swimming"], + admin=False, + ), + 124: User( + id=124, + name="Jane Doe", + email="jane@example.com", + created_on=datetime.now(tz=UTC), + ip_address="10.1.2.5", + hobbies=["running", "dancing"], + admin=False, + ), } + + def local_delitem(key: int) -> None: + del local_db[key] + + def local_contains(key: int) -> bool: + return key in local_db + + mock_db = MagicMock() + mock_db.__delitem__.side_effect = local_delitem + mock_db.__contains__.side_effect = local_contains examples.src.fastapi.FAKE_DB = mock_db @@ -207,23 +288,21 @@ def verify_user_doesnt_exist_mock() -> None: """ Verify the mock calls for the 'user doesn't exist' state. - This function checks that the mock for `FAKE_DB.get` was called, - verifies that it returned `None`, - and ensures that it was called with an integer argument. - It then resets the mock for future tests. - - Returns: - str: A message indicating that the 'user doesn't exist' mock has been verified. + This function checks that the mock for `FAKE_DB.get` was called, verifies + that it returned `None`, and ensures that it was called with an integer + argument. It then resets the mock for future tests. """ import examples.src.fastapi if TYPE_CHECKING: + # During setup, the `FAKE_DB` is replaced with a MagicMock object. + # We need to inform the type checker that this has happened. examples.src.fastapi.FAKE_DB = MagicMock() - examples.src.fastapi.FAKE_DB.get.assert_called_once() + assert len(examples.src.fastapi.FAKE_DB.mock_calls) == 1 + examples.src.fastapi.FAKE_DB.get.assert_called_once() args, kwargs = examples.src.fastapi.FAKE_DB.get.call_args - assert len(args) == 1 assert isinstance(args[0], int) assert kwargs == {} @@ -235,25 +314,64 @@ def verify_user_exists_mock() -> None: """ Verify the mock calls for the 'user exists' state. - This function checks that the mock for `FAKE_DB.get` was called, - verifies that it returned the expected user data, - and ensures that it was called with the integer argument `1`. - It then resets the mock for future tests. - - Returns: - str: A message indicating that the 'user exists' mock has been verified. + This function checks that the mock for `FAKE_DB.get` was called, verifies + that it returned the expected user data, and ensures that it was called with + the integer argument `1`. It then resets the mock for future tests. """ import examples.src.fastapi if TYPE_CHECKING: examples.src.fastapi.FAKE_DB = MagicMock() - examples.src.fastapi.FAKE_DB.get.assert_called_once() + assert len(examples.src.fastapi.FAKE_DB.mock_calls) == 1 + examples.src.fastapi.FAKE_DB.get.assert_called_once() args, kwargs = examples.src.fastapi.FAKE_DB.get.call_args + assert len(args) == 1 + assert isinstance(args[0], int) + assert kwargs == {} + + examples.src.fastapi.FAKE_DB.reset_mock() + + +def verify_mock_post_request_to_create_user() -> None: + import examples.src.fastapi + + if TYPE_CHECKING: + examples.src.fastapi.FAKE_DB = MagicMock() + + assert len(examples.src.fastapi.FAKE_DB.mock_calls) == 2 + examples.src.fastapi.FAKE_DB.__getitem__.assert_called_once() + args, kwargs = examples.src.fastapi.FAKE_DB.__getitem__.call_args assert len(args) == 1 assert isinstance(args[0], int) assert kwargs == {} + examples.src.fastapi.FAKE_DB.__len__.assert_called_once() + args, kwargs = examples.src.fastapi.FAKE_DB.__len__.call_args + assert len(args) == 0 + assert kwargs == {} + examples.src.fastapi.FAKE_DB.reset_mock() + + +def verify_mock_delete_request_to_delete_user() -> None: + import examples.src.fastapi + + if TYPE_CHECKING: + examples.src.fastapi.FAKE_DB = MagicMock() + + assert len(examples.src.fastapi.FAKE_DB.mock_calls) == 2 + + examples.src.fastapi.FAKE_DB.__delitem__.assert_called_once() + args, kwargs = examples.src.fastapi.FAKE_DB.__delitem__.call_args + assert len(args) == 1 + assert isinstance(args[0], int) + assert kwargs == {} + + examples.src.fastapi.FAKE_DB.__contains__.assert_called_once() + args, kwargs = examples.src.fastapi.FAKE_DB.__contains__.call_args + assert len(args) == 1 + assert isinstance(args[0], int) + assert kwargs == {}