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

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.fetch_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.fetch_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.fetch_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.fetch_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,26 @@
 import json
+from inspect import isfunction
 
 from .constants import TYPES
 from .headers import HTTPHeaderDict
 from .helpers import trigger_methods
 
+class WrapJSON:
+    def __init__(self, data):
+        self.data = data
+        # Marker for easier detection in fetch_body
+        self.can_fetch_body = True
+
+    def __call__(self, request, response):
+        data = self.data(request, response)
+        return self.encode_json(data)
+
+    @staticmethod
+    def encode_json(data):
+        if not isinstance(data, str) and not isinstance(data, bytes):
+            data = json.dumps(data, indent=4)
+        return data
+
 
 class Response:
     """
@@ -14,9 +31,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.
     """
@@ -165,7 +182,9 @@ def body(self, body, *, chunked=False):
         Returns:
             self: ``pook.Response`` current instance.
         """
-        if hasattr(body, "encode"):
+        if isfunction(body):
+            pass
+        elif hasattr(body, "encode"):
             body = body.encode("utf-8", "backslashreplace")
         elif isinstance(body, list):
             for i, chunk in enumerate(body):
@@ -178,6 +197,11 @@ def body(self, body, *, chunked=False):
             self.header("Transfer-Encoding", "chunked")
         return self
 
+    def fetch_body(self, request):
+        if isfunction(self._body) or hasattr(self._body, 'can_fetch_body'):
+            self._body = self._body(request, self)
+        return self._body
+
     def json(self, data):
         """
         Defines the mock response JSON body.
@@ -189,16 +213,18 @@ def json(self, data):
             self: ``pook.Response`` current instance.
         """
         self._headers["Content-Type"] = "application/json"
-        if not isinstance(data, str):
-            data = json.dumps(data, indent=4)
+        if isfunction(data):
+            data = WrapJSON(data)
+        else:
+            data = WrapJSON.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.

tests/unit/mock_test.py

@@ -71,6 +71,37 @@ def test_mock_constructor(param_kwargs, query_string, url_404):
         assert res.status == 200
         assert json.loads(res.read()) == {"hello": "from pook"}
 
+def test_dynamic_mock_response_body():
+    def resp_builder(req, resp):
+        return b"hello from pook"
+
+    mock = Mock(
+        url='https://example.com/fetch',
+        reply_status=200,
+        response_body=resp_builder,
+    )
+
+    with pook.use():
+        pook.engine().add_mock(mock)
+        res = urlopen('https://example.com/fetch')
+        assert res.status == 200
+        assert res.read() == b"hello from pook"
+
+def test_dynamic_mock_response_json():
+    def resp_builder(req, resp):
+        return {"hello": "from pook"}
+
+    mock = Mock(
+        url='https://example.com/fetch',
+        reply_status=200,
+        response_json=resp_builder,
+    )
+
+    with pook.use():
+        pook.engine().add_mock(mock)
+        res = urlopen('https://example.com/fetch')
+        assert res.status == 200
+        assert json.loads(res.read()) == {"hello": "from pook"}
 
 @pytest.mark.parametrize(
     "params, req_params, expected",