Feature/dynamic response

History.rst

@@ -1,6 +1,17 @@
 History
 =======
 
+vX.Y.Z / 20xx-xx-xx
+-------------------------
+What's Changed
+^^^^^^^^^^^^^^
+
+  * Add in support for passing a function to response_json, response_body, response_xml
+
+New Contributors
+^^^^^^^^^^^^^^^^
+  * @urkle made their first contribution in https://github.com/h2non/pook/pull/165
+
 v2.1.4 / 2025-07-05
 -------------------------
 

README.rst

@@ -103,6 +103,25 @@ Basic mocking:
         assert resp.json() == {"error": "not found"}
         assert mock.calls == 1
 
+Support dynamic response generation
+
+.. code:: python
+
+    import pook
+    import requests
+
+    @pook.on
+    def test_dynamic():
+        def resp_build(req, resp):
+            return {'test': 1234, 'url': req.url}
+
+        mock = pook.get('http://example.com/test', reply=200, response_json=resp_build)
+
+        resp = requests.get('http://example.com/test')
+        assert resp.status_code == 200
+        assert resp.json() == {'test': 1234, 'url': 'http://example.com/test'}
+        assert mock.calls == 1
+
 Using the chainable API DSL:
 
 .. code:: python

pyproject.toml

@@ -74,6 +74,27 @@ dependencies = [
     "mocket[pook]~=3.12.2; platform_python_implementation != 'PyPy'",
 ]
 
+[dependency-groups]
+dev = [
+     "pre-commit~=4.0",
+    "mypy>=1.11.2",
+
+    "pytest~=8.3",
+    "pytest-asyncio~=0.24",
+    "pytest-pook==0.1.0b0",
+
+    "falcon~=4.0",
+
+    "requests~=2.20",
+    "urllib3~=2.2",
+    "httpx~=0.26",
+
+    "aiohttp~=3.10",
+    "async-timeout~=4.0",
+
+    "mocket[pook]~=3.12.2; platform_python_implementation != 'PyPy'",
+]
+
 [tool.hatch.envs.default.scripts]
 ci = [
     "lint-install",

src/pook/interceptors/_httpx.py

@@ -78,12 +78,12 @@ def _get_pook_request(self, httpx_request: httpx.Request) -> Request:
         return req
 
     def _get_httpx_response(
-        self, httpx_request: httpx.Request, mock_response: Response
+        self, httpx_request: httpx.Request, mock_response: Response, pook_request: Request
     ) -> httpx.Response:
         res = httpx.Response(
             status_code=mock_response._status,
             headers=mock_response._headers,
-            content=mock_response._body,
+            content=mock_response.get_body(pook_request),
             extensions={
                 # TODO: Add HTTP2 response support
                 "http_version": b"HTTP/1.1",
@@ -140,4 +140,4 @@ def handle_request(self, request):
             transport = self._original_transport_for_url(self._client, request.url)
             return transport.handle_request(request)
 
-        return self._get_httpx_response(request, mock._response)
+        return self._get_httpx_response(request, mock._response, pook_request)

src/pook/interceptors/aiohttp.py

@@ -145,7 +145,7 @@ async def _on_request(
         _res._headers = multidict.CIMultiDictProxy(multidict.CIMultiDict(headers))
 
         if res._body:
-            _res.content = SimpleContent(res._body)
+            _res.content = SimpleContent(res.get_body(req))
         else:
             # Define `_content` attribute with an empty string to
             # force do not read from stream (which won't exists)

src/pook/interceptors/http.py

@@ -88,9 +88,12 @@ def getresponse():
         conn.__response = mockres  # type: ignore[attr-defined]
         conn.__state = _CS_REQ_SENT  # type: ignore[attr-defined]
 
+
+        body = res.get_body(req)
+
         # Path reader
         def read():
-            return res._body or b""
+            return body or b""
 
         mockres.read = read
 

src/pook/interceptors/urllib3.py

@@ -138,7 +138,7 @@ def _on_request(
 
         # Shortcut to mock response and response body
         res = mock._response
-        body = res._body
+        body = res.get_body(req)
 
         # Aggregate headers as list of tuples for interface compatibility
         headers = []
@@ -152,7 +152,7 @@ def _on_request(
             body.fp = FakeChunkedResponseBody(body_chunks)  # type:ignore
         else:
             # Assume that the body is a bytes-like object
-            body = io.BytesIO(res._body)
+            body = io.BytesIO(body)
 
         # Return mocked HTTP response
         return HTTPResponse(

src/pook/response.py

@@ -1,9 +1,45 @@
 import json
+from inspect import isfunction
 
 from .constants import TYPES
 from .headers import HTTPHeaderDict
 from .helpers import trigger_methods
 
+def encode_body(body):
+    """
+    Shared code to handle encoding a response body into a byte object OR an array of byte objects.
+    """
+    if hasattr(body, "encode"):
+        body = body.encode("utf-8", "backslashreplace")
+    elif isinstance(body, list):
+        for i, chunk in enumerate(body):
+            if hasattr(chunk, "encode"):
+                body[i] = chunk.encode("utf-8", "backslashreplace")
+
+    return body
+
+def encode_json(data):
+    """
+    Shared method to encode "JSON" data into a string.
+    Note, this does not handle encoding a string into a JSON string (e.g. 'MyString'=>'"MyString"'), but does handle dict/list/numbers.
+    """
+    if not isinstance(data, str) and not isinstance(data, bytes):
+        data = json.dumps(data, indent=4)
+    return data
+
+
+class WrapJSON:
+    """
+    A Wrapper callable class that handles a function returning JSON data to be encoded.
+    This performs the same logic as the json() function would with a fixed value
+    """
+    def __init__(self, data):
+        self.data = data
+
+    def __call__(self, request, response):
+        data = self.data(request, response)
+        return encode_json(data)
+
 
 class Response:
     """
@@ -14,9 +50,9 @@ class Response:
     Arguments:
         status (int): HTTP response status code. Defaults to ``200``.
         headers (dict): HTTP response headers.
-        body (str|bytes): HTTP response body.
-        json (str|dict|list): HTTP response JSON body.
-        xml (str): HTTP response XML body.
+        body (str|bytes|function): HTTP response body.
+        json (str|bytes|dict|list|function): HTTP response JSON body.
+        xml (str|function): HTTP response XML body.
         type (str): HTTP response content MIME type.
         file (str): file path to HTTP body response.
     """
@@ -159,49 +195,49 @@ def body(self, body, *, chunked=False):
         Defines response body data.
 
         Arguments:
-            body (str|bytes|list): response body to use.
+            body (str|bytes|list|callable): response body to use.
             chunked (bool): return a chunked response.
 
         Returns:
             self: ``pook.Response`` current instance.
         """
-        if hasattr(body, "encode"):
-            body = body.encode("utf-8", "backslashreplace")
-        elif isinstance(body, list):
-            for i, chunk in enumerate(body):
-                if hasattr(chunk, "encode"):
-                    body[i] = chunk.encode("utf-8", "backslashreplace")
-
-        self._body = body
+        self._body = encode_body(body)
 
         if chunked:
             self.header("Transfer-Encoding", "chunked")
         return self
 
+    def get_body(self, request):
+        if callable(self._body):
+            return encode_body(self._body(request, self))
+        return self._body
+
     def json(self, data):
         """
         Defines the mock response JSON body.
 
         Arguments:
-            data (dict|list|str): JSON body data.
+            data (dict|list|str|callable): JSON body data.
 
         Returns:
             self: ``pook.Response`` current instance.
         """
         self._headers["Content-Type"] = "application/json"
-        if not isinstance(data, str):
-            data = json.dumps(data, indent=4)
+        if callable(data):
+            data = WrapJSON(data)
+        else:
+            data = encode_json(data)
 
         return self.body(data)
 
     def xml(self, xml):
         """
         Defines the mock response XML body.
 
-        For not it only supports ``str`` as input type.
+        For now it only supports ``str`` as input type.
 
         Arguments:
-            xml (str): XML body data to use.
+            xml (str|callable): XML body data to use.
 
         Returns:
             self: ``pook.Response`` current instance.

tests/unit/interceptors/base.py

@@ -177,6 +177,70 @@ def test_json_request_and_response(self, url_404):
         assert body
         assert json.loads(body) == json_response
 
+    @pytest.mark.pook
+    def test_dynamic_text_response_body(self, url_404):
+        """Dynamic mock response body as raw data."""
+
+        def resp_builder(req, resp):
+            return b"hello from pook"
+
+        pook.get(url_404).reply(200).body(resp_builder)
+
+        status, body, *_ = self.make_request("GET", url_404)
+        assert status == 200
+        assert body == b"hello from pook"
+
+    @pytest.mark.pook
+    def test_dynamic_json_response_body(self, url_404):
+        """Dynamic mock response body with JSON."""
+
+        def resp_builder(req, resp):
+            return {"hello": "from pook"}
+
+        pook.get(url_404).reply(200).json(resp_builder)
+
+        status, body, headers = self.make_request("GET", url_404)
+        assert status == 200
+        assert json.loads(body) == {"hello": "from pook"}
+        assert headers["Content-Type"] == "application/json"
+
+    @pytest.mark.pook
+    def test_dynamic_xml_response_body(self, url_404):
+        """Dynamic mock response body with XML."""
+
+        def resp_builder(req, resp):
+            return "<test>hello from pook</test>"
+
+        pook.get(url_404).reply(200).xml(resp_builder)
+
+        status, body, headers = self.make_request("GET", url_404)
+        assert status == 200
+        assert body == b"<test>hello from pook</test>"
+        # TODO what is the purpose of the XML() response method if it does nothing special over body()
+        # assert headers["Content-Type"] == "application/xml"
+
+    @pytest.mark.pook
+    def test_multiple_dynamic_response_body(self, url_404):
+        """Dynamic mock body with multiple requests."""
+
+        class RespBuilder:
+            def __init__(self):
+                self.counter = 0
+
+            def __call__(self, req, resp):
+                self.counter = self.counter + 1
+                return {"hello": "from pook", "value": self.counter}
+
+        pook.get(url_404).persist().reply(200).json(RespBuilder())
+
+        # Make 3 requests to test the dynamic response
+        for i in range(1, 4):
+            status, body, headers = self.make_request("GET", url_404)
+            print(status, body)
+            assert status == 200
+            assert json.loads(body) == {"hello": "from pook", "value": i}
+            assert headers["Content-Type"] == "application/json"
+
     @pytest.mark.pook
     def test_header_sent(self, url_404):
         """Sent headers can be matched."""

tests/unit/mock_test.py

@@ -10,6 +10,8 @@
 from pook.exceptions import PookNoMatches
 from pook.mock import Mock
 from pook.request import Request
+from pook.response import WrapJSON
+
 from tests.unit.fixtures import BINARY_FILE, BINARY_FILE_PATH
 
 
@@ -71,6 +73,30 @@ def test_mock_constructor(param_kwargs, query_string, url_404):
         assert res.status == 200
         assert json.loads(res.read()) == {"hello": "from pook"}
 
+def test_setting_dynamic_body(url_404):
+    def resp_builder(req, resp):
+        return b"hello from pook"
+
+    mock = Mock(
+        url=url_404,
+        reply_status=200,
+        response_body=resp_builder,
+    )
+
+    assert mock._response._body == resp_builder
+
+def test_dynamic_mock_response_json(url_404):
+    def resp_builder(req, resp):
+        return {"hello": "from pook"}
+
+    mock = Mock(
+        url=url_404,
+        reply_status=200,
+        response_json=resp_builder,
+    )
+
+    assert isinstance(mock._response._body, WrapJSON)
+    assert  mock._response._body.data == resp_builder
 
 @pytest.mark.parametrize(
     "params, req_params, expected",