From 5948db4c7a5cbd95795c9bc243dba7bed0c85c39 Mon Sep 17 00:00:00 2001 From: Edward Rudd Date: Wed, 27 Aug 2025 11:26:38 -0400 Subject: [PATCH 1/3] Fix wrong word on comment --- src/pook/response.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pook/response.py b/src/pook/response.py index b07b75f..7e561d3 100644 --- a/src/pook/response.py +++ b/src/pook/response.py @@ -198,7 +198,7 @@ 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. From 0a9bd33a27787c552b4f9d5adbb9405709a19caf Mon Sep 17 00:00:00 2001 From: Edward Rudd Date: Wed, 27 Aug 2025 11:29:30 -0400 Subject: [PATCH 2/3] Add support for using a function for a body response ```python def dyn_resp(req, resp): # Can make the response dynamic based on the request return {'key': 'value'} pook.get('https://example.org/', persist=True, response_json=dyn_resp) ``` --- History.rst | 11 +++++++++ README.rst | 19 ++++++++++++++++ src/pook/interceptors/_httpx.py | 6 ++--- src/pook/interceptors/aiohttp.py | 2 +- src/pook/interceptors/http.py | 5 ++++- src/pook/interceptors/urllib3.py | 4 ++-- src/pook/response.py | 38 +++++++++++++++++++++++++++----- tests/unit/mock_test.py | 31 ++++++++++++++++++++++++++ 8 files changed, 103 insertions(+), 13 deletions(-) diff --git a/History.rst b/History.rst index d447984..efea68a 100644 --- a/History.rst +++ b/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 ------------------------- diff --git a/README.rst b/README.rst index c0393ab..bdb5550 100644 --- a/README.rst +++ b/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 diff --git a/src/pook/interceptors/_httpx.py b/src/pook/interceptors/_httpx.py index 07566ae..c800ac0 100644 --- a/src/pook/interceptors/_httpx.py +++ b/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) diff --git a/src/pook/interceptors/aiohttp.py b/src/pook/interceptors/aiohttp.py index 5646513..dc2314f 100644 --- a/src/pook/interceptors/aiohttp.py +++ b/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) diff --git a/src/pook/interceptors/http.py b/src/pook/interceptors/http.py index 567b971..f4a0ac0 100644 --- a/src/pook/interceptors/http.py +++ b/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 diff --git a/src/pook/interceptors/urllib3.py b/src/pook/interceptors/urllib3.py index 42c1fac..0fbc1d7 100644 --- a/src/pook/interceptors/urllib3.py +++ b/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( diff --git a/src/pook/response.py b/src/pook/response.py index 7e561d3..04c4e13 100644 --- a/src/pook/response.py +++ b/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,8 +213,10 @@ 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) diff --git a/tests/unit/mock_test.py b/tests/unit/mock_test.py index f195893..d830c2f 100644 --- a/tests/unit/mock_test.py +++ b/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", From e390992df2780ad2cfa76003f639b0df3eb6b22b Mon Sep 17 00:00:00 2001 From: Edward Rudd Date: Fri, 10 Oct 2025 16:39:49 -0400 Subject: [PATCH 3/3] fixup! Add support for using a function for a body response --- pyproject.toml | 21 +++++++++++ src/pook/interceptors/_httpx.py | 2 +- src/pook/interceptors/aiohttp.py | 2 +- src/pook/interceptors/http.py | 2 +- src/pook/interceptors/urllib3.py | 2 +- src/pook/response.py | 64 ++++++++++++++++++-------------- tests/unit/interceptors/base.py | 64 ++++++++++++++++++++++++++++++++ tests/unit/mock_test.py | 23 +++++------- 8 files changed, 135 insertions(+), 45 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 120bec1..38692a8 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/pook/interceptors/_httpx.py b/src/pook/interceptors/_httpx.py index c800ac0..4c19776 100644 --- a/src/pook/interceptors/_httpx.py +++ b/src/pook/interceptors/_httpx.py @@ -83,7 +83,7 @@ def _get_httpx_response( res = httpx.Response( status_code=mock_response._status, headers=mock_response._headers, - content=mock_response.fetch_body(pook_request), + content=mock_response.get_body(pook_request), extensions={ # TODO: Add HTTP2 response support "http_version": b"HTTP/1.1", diff --git a/src/pook/interceptors/aiohttp.py b/src/pook/interceptors/aiohttp.py index dc2314f..ca5f700 100644 --- a/src/pook/interceptors/aiohttp.py +++ b/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.fetch_body(req)) + _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) diff --git a/src/pook/interceptors/http.py b/src/pook/interceptors/http.py index f4a0ac0..227122a 100644 --- a/src/pook/interceptors/http.py +++ b/src/pook/interceptors/http.py @@ -89,7 +89,7 @@ def getresponse(): conn.__state = _CS_REQ_SENT # type: ignore[attr-defined] - body = res.fetch_body(req) + body = res.get_body(req) # Path reader def read(): diff --git a/src/pook/interceptors/urllib3.py b/src/pook/interceptors/urllib3.py index 0fbc1d7..545a03d 100644 --- a/src/pook/interceptors/urllib3.py +++ b/src/pook/interceptors/urllib3.py @@ -138,7 +138,7 @@ def _on_request( # Shortcut to mock response and response body res = mock._response - body = res.fetch_body(req) + body = res.get_body(req) # Aggregate headers as list of tuples for interface compatibility headers = [] diff --git a/src/pook/response.py b/src/pook/response.py index 04c4e13..55b55ec 100644 --- a/src/pook/response.py +++ b/src/pook/response.py @@ -5,21 +5,40 @@ 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 - # 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 + return encode_json(data) class Response: @@ -176,30 +195,21 @@ 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 isfunction(body): - pass - elif 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 fetch_body(self, request): - if isfunction(self._body) or hasattr(self._body, 'can_fetch_body'): - self._body = self._body(request, self) + def get_body(self, request): + if callable(self._body): + return encode_body(self._body(request, self)) return self._body def json(self, data): @@ -207,16 +217,16 @@ 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 isfunction(data): + if callable(data): data = WrapJSON(data) else: - data = WrapJSON.encode_json(data) + data = encode_json(data) return self.body(data) @@ -227,7 +237,7 @@ def xml(self, xml): 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. diff --git a/tests/unit/interceptors/base.py b/tests/unit/interceptors/base.py index ea49a61..2a3cdcd 100644 --- a/tests/unit/interceptors/base.py +++ b/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 "hello from pook" + + pook.get(url_404).reply(200).xml(resp_builder) + + status, body, headers = self.make_request("GET", url_404) + assert status == 200 + assert body == b"hello from pook" + # 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.""" diff --git a/tests/unit/mock_test.py b/tests/unit/mock_test.py index d830c2f..3a58eaa 100644 --- a/tests/unit/mock_test.py +++ b/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,37 +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_dynamic_mock_response_body(): +def test_setting_dynamic_body(url_404): def resp_builder(req, resp): return b"hello from pook" mock = Mock( - url='https://example.com/fetch', + url=url_404, 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" + assert mock._response._body == resp_builder -def test_dynamic_mock_response_json(): +def test_dynamic_mock_response_json(url_404): def resp_builder(req, resp): return {"hello": "from pook"} mock = Mock( - url='https://example.com/fetch', + url=url_404, 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"} + assert isinstance(mock._response._body, WrapJSON) + assert mock._response._body.data == resp_builder @pytest.mark.parametrize( "params, req_params, expected",