Merge pull request #85 from SlavaSkvortsov/add_last_request_property

Refactor call stats with easier api

respx/api.py

@@ -1,13 +1,13 @@
 from typing import Callable, Optional, Pattern, Union, overload
 
 from .mocks import MockTransport
-from .models import ContentDataTypes, DefaultType, HeaderTypes, RequestPattern
+from .models import CallList, ContentDataTypes, DefaultType, HeaderTypes, RequestPattern
 
 mock = MockTransport(assert_all_called=False)
 
 aliases = mock.aliases
 stats = mock.stats
-calls = mock.calls
+calls: CallList = mock.calls
 
 
 def start() -> None:

respx/models.py

@@ -2,12 +2,15 @@
 import re
 from functools import partial
 from typing import (
+    TYPE_CHECKING,
     Any,
     AsyncIterable,
     Callable,
     Dict,
+    Generator,
     Iterable,
     List,
+    NamedTuple,
     Optional,
     Pattern,
     Sequence,
@@ -21,6 +24,10 @@
 import httpx
 from httpcore import AsyncByteStream, SyncByteStream
 
+if TYPE_CHECKING:
+    from unittest.mock import _CallList  # pragma: nocover
+
+
 URL = Tuple[bytes, bytes, Optional[int], bytes]
 Headers = List[Tuple[bytes, bytes]]
 Request = Tuple[
@@ -91,6 +98,24 @@ def decode_response(
     )
 
 
+class Call(NamedTuple):
+    request: httpx.Request
+    response: Optional[httpx.Response]
+
+
+class CallList(list):
+    def __iter__(self) -> Generator[Call, None, None]:
+        yield from super().__iter__()
+
+    @classmethod
+    def from_unittest_call_list(cls, call_list: "_CallList") -> "CallList":
+        return cls(Call(request, response) for (request, response), _ in call_list)
+
+    @property
+    def last(self) -> Optional[Call]:
+        return self[-1] if self else None
+
+
 class ResponseTemplate:
     _content: Optional[ContentDataTypes]
     _text: Optional[str]
@@ -284,18 +309,16 @@ def __init__(
         self.stats = mock.MagicMock()
 
     @property
-    def called(self):
+    def called(self) -> bool:
         return self.stats.called
 
     @property
-    def call_count(self):
+    def call_count(self) -> int:
         return self.stats.call_count
 
     @property
-    def calls(self):
-        return [
-            (request, response) for (request, response), _ in self.stats.call_args_list
-        ]
+    def calls(self) -> CallList:
+        return CallList.from_unittest_call_list(self.stats.call_args_list)
 
     def get_url(self) -> Optional[URLPatternTypes]:
         return self._url

respx/transports.py

@@ -11,13 +11,13 @@
 from .models import (
     URL,
     AsyncResponse,
+    CallList,
     ContentDataTypes,
     DefaultType,
     Headers,
     HeaderTypes,
     Request,
     RequestPattern,
-    Response,
     ResponseTemplate,
     SyncResponse,
     decode_request,
@@ -40,7 +40,7 @@ def __init__(
         self.aliases: Dict[str, RequestPattern] = {}
 
         self.stats = mock.MagicMock()
-        self.calls: List[Tuple[Request, Optional[Response]]] = []
+        self.calls: CallList = CallList()
 
     def clear(self):
         """
@@ -293,9 +293,7 @@ def record(
         self.stats(request, response)
 
         # Copy stats due to unwanted use of property refs in the high-level api
-        self.calls[:] = (
-            (request, response) for (request, response), _ in self.stats.call_args_list
-        )
+        self.calls[:] = CallList.from_unittest_call_list(self.stats.call_args_list)
 
     def assert_all_called(self) -> None:
         assert all(

tests/test_mock.py

@@ -226,9 +226,11 @@ async def test(respx_mock):
         assert respx.stats.call_count == 0
         assert respx_mock.stats.call_count == 1
 
-        _request, _response = respx_mock.calls[-1]
+        _request, _response = respx_mock.calls.last
         assert _request is not None
         assert _response is not None
+        assert respx_mock.calls.last.request is _request
+        assert respx_mock.calls.last.response is _response
         assert _request.url == "https://some.thing/"
 
     await test()

tests/test_stats.py

@@ -36,6 +36,7 @@ async def test(backend):
         assert foobar1.called is False
         assert foobar1.call_count == len(foobar1.calls)
         assert foobar1.call_count == 0
+        assert foobar1.calls.last is None
         assert respx.stats.call_count == len(respx.calls)
         assert respx.stats.call_count == 0
 
@@ -51,6 +52,8 @@ async def test(backend):
         _request, _response = foobar1.calls[-1]
         assert isinstance(_request, httpx.Request)
         assert isinstance(_response, httpx.Response)
+        assert foobar1.calls.last.request is _request
+        assert foobar1.calls.last.response is _response
         assert _request.method == "GET"
         assert _request.url == url
         assert _response.status_code == get_response.status_code == 202
@@ -82,8 +85,8 @@ async def test(backend):
     if isinstance(backend, TrioBackend):
         trio.run(test, backend)
     else:
+        loop = asyncio.new_event_loop()
         try:
-            loop = asyncio.new_event_loop()
             loop.run_until_complete(test(backend))
         finally:
             loop.close()