Add ParsedRequest assertions helper

Makefile

@@ -1,26 +1,31 @@
 PYTHON_VERSION ?= 3.9
 
-dist: clean-dist venv
+dist: clean-dist setup
 	. venv/bin/activate && python3 -m build .
 
-setup: venv
+.PHONY: setup
+setup: venv/setup.txt
 
-venv: dev-requirements.txt requirements.txt
+venv:
 	virtualenv venv --python=${PYTHON_VERSION}
+
+venv/setup.txt: venv dev-requirements.txt requirements.txt
 	. venv/bin/activate && \
 	pip3 install --upgrade pip && \
 	pip3 install \
 	--requirement dev-requirements.txt \
 	--requirement requirements.txt
+	touch venv/setup.txt
 
 .PHONY: test
-test: venv
-	@ . venv/bin/activate && PYTHONPATH=src/ pytest -rsx tests src --cov ./src/requtests --no-cov-on-fail --cov-report term-missing --doctest-modules --doctest-continue-on-failure
+test: setup
+	@ . venv/bin/activate && PYTHONPATH=src/ pytest -vv -rsx tests src --cov ./src/requtests --no-cov-on-fail --cov-report term-missing --doctest-modules --doctest-continue-on-failure
 	@ . venv/bin/activate && flake8 src --exclude '#*,~*,.#*'
 	@ . venv/bin/activate && black --check src tests
+	@ . venv/bin/activate && mypy src
 
 .PHONY: test-focus
-test-focus: venv/setup.txt
+test-focus: setup
 	@ . venv/bin/activate && PYTHONPATH=src/ pytest -vv -m focus -rsx tests src --cov ./src/requtests --no-cov-on-fail --cov-report term-missing --doctest-modules --doctest-continue-on-failure
 	@ . venv/bin/activate && flake8 src --exclude '#*,~*,.#*'
 	@ . venv/bin/activate && black --check src tests
@@ -34,3 +39,12 @@ clean-dist:
 	rm -rf build
 	rm -rf src/requtests.egg-info
 	rm -rf dist
+
+.PHONY: release
+release: test dist
+	. venv/bin/activate && twine upload dist/*
+
+.PHONY: test-release
+test-release: test dist
+	. venv/bin/activate && twine upload -r testpypi dist/*
+

README.md

@@ -10,8 +10,8 @@ Test helpers for the [requests](https://docs.python-requests.org) library
 
 ## Installation
 
-Install the package `requtests` version `1.1+` from PyPI.
-The recommended `requirements.txt` line is `requtests~=1.1`.
+Install the package `requtests` version `1.2+` from PyPI.
+The recommended `requirements.txt` line is `requtests~=1.2`.
 
 ### `FakeAdapter`
 
@@ -23,8 +23,8 @@ The faked adapter can be mounted using the standard `mount` method on an instanc
 #### Example
 
 ```python3
-from requtests import FakeAdapter, fake_response
 from requests import Session
+from requtests import FakeAdapter, fake_response
 
 
 class Client:
@@ -96,9 +96,34 @@ def test_login():
     password = "my-password"
     request_func = fake_request_with_response(json={"token": "my-login-token"})
     assert login(username, password, request_func=request_func) == "my-login-token"
-
 ```
 
 ### `fake_response`
 
 Returns a `requests.Response` object with either the return value of its `json()` method set to a python data structure or its `text` property set.
+
+### `ParsedRequest`
+
+A test helper object wrapping a `PreparedRequest` object to make it easier to write assertions. In addition to wrapping the `PreparedRequest`'s `body`, `headers`, `method`, and `url` properties, it also provides the following convenience properties.
+
+* `endpoint` - the URL without any query parameters.
+* `query` - any query parameters, parsed and decoded.
+* `json` - the body parsed as JSON. 
+* `text` - the body decoded as a string.
+
+#### Example
+
+```python3
+from requtests import ParsedRequest
+
+def _create_user_assertions(prepared_request, **kwargs):
+    parsed_request = ParsedRequest(prepared_request)
+    assert parsed_request.method == "POST"
+    assert parsed_request.url == "https://example.com/users?action=create"
+    assert parsed_request.endpoint == "https://example.com/users"
+    assert parsed_request.query == {"action": "create"}
+    assert parsed_request.headers["Authorization"] == "Bearer token"
+    assert parsed_request.body == b'{"username": "my_username"}'
+    assert parsed_request.json == {"username": "my_username"}
+    assert parsed_request.text == '{"username": "my_username"}'
+```

dev-requirements.txt

@@ -1,7 +1,9 @@
 black
 build
 flake8
+mypy
 pytest
 pytest-clarity
 pytest-cov
 twine
+types-requests

src/requtests/__init__.py

@@ -11,6 +11,7 @@
     fake_request_with_response,
 )
 from .fake_response import fake_response
+from .parsed_request import ParsedRequest
 
 __all__ = [
     "FakeAdapter",
@@ -24,7 +25,8 @@
     "fake_request",
     "fake_request_with_response",
     "fake_response",
+    "ParsedRequest",
 ]
-__version__ = "1.1.0"
+__version__ = "1.2.0"
 
 VERSION = __version__

src/requtests/fake_adapter.py

@@ -1,9 +1,11 @@
 from itertools import cycle
+from requests import Response
 from requests.adapters import BaseAdapter
+from .protocols import OptionalAssertions
 
 
 class FakeAdapter(BaseAdapter):
-    def __init__(self, *responses, assertions=None):
+    def __init__(self, *responses: Response, assertions: OptionalAssertions = None):
         super().__init__()
         self.closed = 0
         self.responses = _to_generator(responses)

src/requtests/fake_request.py

@@ -1,50 +1,54 @@
 from functools import partial
 from requests import Session
-from requtests.fake_adapter import FakeAdapter
-from requtests.fake_response import fake_response
+from .fake_adapter import FakeAdapter
+from .fake_response import fake_response
+from .protocols import OptionalAssertions, Responder
 
 
-def fake_delete(*responses, assertions=None):
+def fake_delete(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "delete")
 
 
-def fake_get(*responses, assertions=None):
+def fake_get(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "get")
 
 
-def fake_head(*responses, assertions=None):
+def fake_head(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "head")
 
 
-def fake_options(*responses, assertions=None):
+def fake_options(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "options")
 
 
-def fake_patch(*responses, assertions=None):
+def fake_patch(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "patch")
 
 
-def fake_post(*responses, assertions=None):
+def fake_post(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "post")
 
 
-def fake_put(*responses, assertions=None):
+def fake_put(*responses, assertions: OptionalAssertions = None) -> Responder:
     return partial(fake_request(*responses, assertions=assertions), "put")
 
 
-def fake_request_with_response(assertions=None, **response_config):
+def fake_request_with_response(
+    assertions: OptionalAssertions = None,
+    **response_config,
+) -> Responder:
     """
     Creates a request function that returns a response given the response_config.
     """
     return fake_request(fake_response(**response_config), assertions=assertions)
 
 
-def fake_request(*responses, assertions=None):
+def fake_request(*responses, assertions: OptionalAssertions = None) -> Responder:
     """
     Creates a request function that returns the supplied responses, one at a time.
     Making a new request after the last response has been returned results in a StopIteration error.
     """
     adapter = FakeAdapter(*responses, assertions=assertions)
     session = Session()
-    session.get_adapter = lambda url: adapter
+    setattr(session, "get_adapter", lambda url: adapter)
     return session.request

src/requtests/fake_response.py

@@ -1,5 +1,6 @@
 from json import dumps as to_json
 from requests.models import Response
+from requests.structures import CaseInsensitiveDict
 
 
 def fake_response(
@@ -8,18 +9,18 @@ def fake_response(
     status_code=200,
     text=None,
     url=None,
-    headers={},
-):
+    headers=None,
+) -> Response:
     """
-    Returns a populated instance of Response.
+    Returns a populated instance of requests.models.Response.
     """
 
     response = Response()
     response._content = _content(json, text)
     response.reason = reason
     response.status_code = status_code
     response.url = url
-    response.headers = headers
+    response.headers = CaseInsensitiveDict(**(headers or {}))
     return response
 
 

src/requtests/parsed_request.py

@@ -0,0 +1,70 @@
+import json
+from json import JSONDecodeError
+from typing import Any, Dict, List, Optional, Union
+from urllib.parse import parse_qs, urlparse
+
+
+class CannotParseBodyAsJSON(RuntimeError):
+    def __init__(self, error):
+        super().__init__(error)
+        self.error = error
+
+
+class ParsedRequest:
+    def __init__(self, prepared_request):
+        self.prepared_request = prepared_request
+        self._parsed_url = urlparse(self.prepared_request.url)
+        self._parsed_query = parse_qs(self._parsed_url.query)
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__} [{self.method}]>"
+
+    @property
+    def body(self) -> Optional[bytes]:
+        return self.prepared_request.body
+
+    @property
+    def endpoint(self) -> str:
+        return f"{self._parsed_url.scheme}://{self._parsed_url.netloc}{self._parsed_url.path}"
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        return self.prepared_request.headers
+
+    @property
+    def json(self) -> Any:
+        """
+        The body of the prepared request, parsed as JSON.
+
+        Raises a CannotParseBodyAsJSON error if the body is not valid JSON.
+        """
+        try:
+            return json.loads(self.prepared_request.body)
+        except (TypeError, JSONDecodeError) as e:
+            raise CannotParseBodyAsJSON(e)
+
+    @property
+    def method(self) -> str:
+        return self.prepared_request.method
+
+    @property
+    def query(self) -> Dict[str, Any]:
+        return {key: _delist(value) for key, value in self._parsed_query.items()}
+
+    @property
+    def text(self) -> str:
+        """
+        The body of the prepared request, decoded as Unicode.
+        """
+        return self.prepared_request.body.decode() if self.prepared_request.body else ""
+
+    @property
+    def url(self) -> str:
+        return self.prepared_request.url
+
+
+def _delist(value: List[Any]) -> Union[Any, List[Any]]:
+    """
+    Extracts the value from a list with a single value, leaving other lists unmodifed.
+    """
+    return value[0] if len(value) == 1 else value

src/requtests/protocols.py

@@ -0,0 +1,15 @@
+from typing import Any, Callable, List, Optional, Protocol, Union
+from requests.models import PreparedRequest, Response
+
+
+class AssertionFunction(Protocol):
+    def __call__(self, prepared_request: PreparedRequest, **kwargs) -> Any:
+        """
+        An assertion function is expected to raise an error if any of its assertions fail.
+        """
+        pass
+
+
+Assertions = Union[AssertionFunction, List[AssertionFunction]]
+OptionalAssertions = Optional[Assertions]
+Responder = Callable[..., Response]

tests/fake_adapter_test.py

@@ -28,7 +28,7 @@ def test_fake_adapter_with_assert_step():
         response,
         assertions=assert_prepared_request(url=TEST_URL, body=TEST_DATA),
     )
-    assert adapter.send(build_request(url=TEST_URL, body=TEST_DATA)) == response
+    assert adapter.send(build_request(url=TEST_URL, data=TEST_DATA)) == response
 
 
 def test_fake_adapter_with_failing_assert_step():
@@ -37,8 +37,8 @@ def test_fake_adapter_with_failing_assert_step():
         response,
         assertions=assert_prepared_request(url=TEST_URL, body=TEST_DATA),
     )
-    with pytest.raises(AssertionError, match="assert 'unexpected data' == 'some data'"):
-        adapter.send(build_request(url=TEST_URL, body="unexpected data")) == response
+    with pytest.raises(AssertionError, match="some data"):
+        adapter.send(build_request(url=TEST_URL, data="unexpected data")) == response
 
 
 def test_fake_adapter_with_multiple_responses():
@@ -49,26 +49,24 @@ def test_fake_adapter_with_multiple_responses():
         response_2,
         assertions=assert_prepared_request(url=TEST_URL, body=TEST_DATA),
     )
-    request = build_request(url=TEST_URL, body=TEST_DATA)
+    request = build_request(url=TEST_URL, data=TEST_DATA)
     assert adapter.send(request) is response_1
     assert adapter.send(request) is response_2
 
 
 def test_fake_adapter_with_multiple_responses_and_assertions():
-    data_1 = TEST_DATA
-    data_2 = "some more data"
     response_1 = fake_response(status_code=429)
     response_2 = fake_response()
     adapter = FakeAdapter(
         response_1,
         response_2,
         assertions=[
-            assert_prepared_request(url=TEST_URL, body=data_1),
-            assert_prepared_request(url=TEST_URL, body=data_2),
+            assert_prepared_request(url=TEST_URL, body=TEST_DATA),
+            assert_prepared_request(url=TEST_URL, body=b'{"even": "more data"}'),
         ],
     )
-    request_1 = build_request(url=TEST_URL, body=data_1)
-    request_2 = build_request(url=TEST_URL, body=data_2)
+    request_1 = build_request(url=TEST_URL, data=TEST_DATA)
+    request_2 = build_request(url=TEST_URL, json={"even": "more data"})
     assert adapter.send(request_1) is response_1
     assert adapter.send(request_2) is response_2
 

tests/fake_request_test.py

@@ -1,6 +1,4 @@
-import pytest
 from requests.models import PreparedRequest
-
 from requtests import fake_request, fake_request_with_response, fake_response
 from tests.test_utils import assert_response
 

tests/fake_response_test.py

@@ -1,5 +1,5 @@
 from requtests import fake_response
-from .test_utils import assert_response
+from tests.test_utils import assert_response
 
 
 def test_fake_response_with_json_data():